dag_impl.hpp

#pragma once
#include <functional_dag/error_codes.h>
 
#include <expected>
#include <iostream>
#include <unordered_set>
 
#include "functional_dag/dag_interface.hpp"
#include "functional_dag/impl/dag_fanout_impl.hpp"
 
namespace fn_dag {
using namespace std;

template <typename IDType>
class _dag_base {
 public:
  virtual ~_dag_base() = default;

  virtual bool dag_contains(const IDType &_id) = 0;

  virtual void print() = 0;

  virtual const IDType &get_id() = 0;

  virtual void push_once() = 0;
};

template <typename OriginType, typename IDType>
class dag : public _dag_base<IDType> {
 private:
  const IDType m_id;                 // The ID of the DAG itself
  dag_source<OriginType> *m_source;  // The source generator that creates data
  dag_fanout_node<OriginType, IDType>
      m_children;  // The children of the source to propagate data across
  unordered_set<IDType> m_children_ids;  // An optimization: a quick O(1) set
                                         // lookup of the children IDs
  const _dag_context
      &g_context;   // The shared state across all of the children of this node.
  thread m_thread;  // Thread to run on if this DAG runs multi-threaded.
 
 public:
  dag(const IDType &_id, dag_source<OriginType> *_lsource,
      const _dag_context &_context, bool _startThread)
      : m_id(_id),
        m_source(_lsource),
        m_children(_context),
        m_children_ids(),
        g_context(_context) {
    if (_startThread) m_thread = thread(&dag::start_source, this);
  }

  ~dag() {
    if (m_thread.joinable()) m_thread.join();
    delete m_source;
  }

  const IDType &get_id() { return m_id; }

  void manual_pump(const unique_ptr<OriginType> _raw_dat) {
    m_children.fan_out(_raw_dat);
  }

  bool dag_contains(const IDType &_id) { return m_children_ids.count(_id) > 0; }

  template <typename In, typename Out>
  [[nodiscard]] expected<IDType, error_codes> add_filter(
      IDType _newID, dag_node<In, Out> *_new_filter, IDType _on_node) {
    _internal_dag_node<In, Out, IDType> *new_node;
    new_node =
        new _internal_dag_node<In, Out, IDType>(_newID, _new_filter, g_context);
    m_children_ids.insert(_newID);
    auto res = m_children.add_node_to_subdag(new_node, _on_node, m_id);
    if (!res) {
      delete new_node;
    }
    return res;
  }

  void print() {
    *g_context.log << "->" << m_id << endl;
    m_children.print(string(g_context.indent_str));
  }

  void push_once() {
    unique_ptr<OriginType> dat = m_source->update();
    if (dat.get() != nullptr) m_children.fan_out(std::move(dat));
  }
 
 private:
  void start_source() {
    while (!g_context.filter_off) push_once();
  }
};
};  // namespace fn_dag