dag_fanout_impl.hpp

#pragma once
 
#include <functional_dag/error_codes.h>
 
#include <expected>
#include <functional_dag/core/dag_utils.hpp>
#include <functional_dag/impl/dag_node_impl.hpp>
#include <thread>
#include <vector>
 
#include "include/functional_dag/error_codes.h"
 
namespace fn_dag {
using namespace std;

template <typename Type, typename IDType>
class dag_fanout_node {
 private:
  const fn_dag::_dag_context &g_context;  // Shared state
  vector<_abstract_internal_dag_node<Type, IDType> *>
      m_children;  // Children to fan-out to

  void _add_node(_abstract_internal_dag_node<Type, IDType> *_new_node) {
    m_children.push_back(_new_node);
  }
 
 public:
  dag_fanout_node(const _dag_context &_context)
      : g_context(_context), m_children() {}

  ~dag_fanout_node() {
    for (auto internal_dag : m_children) delete internal_dag;
    m_children.clear();
  }

  void fan_out(unique_ptr<Type> _data) {
    if (_data.get() == nullptr) return;
    if (!g_context.run_single_threaded) {
      vector<thread> child_threads;
 
      for (auto it : m_children)
        child_threads.push_back(thread(
            &fn_dag::_abstract_internal_dag_node<Type, IDType>::run_filter, it,
            _data.get()));
 
      for (uint32_t i = 0; i < child_threads.size(); i++)
        child_threads[i].join();
    } else
      for (auto it : m_children) it->run_filter(_data.get());
  }

  void print(const string &_indent) {
    const string next_indent = _indent + string(g_context.indent_str);
 
    for (const auto child : m_children) {
      *g_context.log << _indent << "->";
      child->print(next_indent);
    }
  }

  template <typename In, typename Out>
  [[nodiscard]] expected<IDType, error_codes> add_node_to_subdag(
      _internal_dag_node<In, Out, IDType> *_node_to_add, const IDType _onto,
      const IDType _parent_id) {
    if (_onto == _parent_id) {
      _add_node((_abstract_internal_dag_node<In, IDType> *)_node_to_add);
      return _parent_id;
    } else {
      for (auto child = m_children.cbegin(); child != m_children.cend();
           child++) {
        auto *internal_child =
            static_cast<_internal_dag_node<In, Type, IDType> *>(*child);
        fn_dag::dag_fanout_node<Type, IDType> *fanout_node =
            internal_child->m_child;
        if (auto p = fanout_node->add_node_to_subdag(_node_to_add, _onto,
                                                     (*child)->get_id());
            p.has_value())
          return p.value();
      }
    }
 
    return unexpected(error_codes::PARENT_NOT_FOUND);
    ;
  }
};
}  // namespace fn_dag