[DRAFT] Declarative Port Graph

examples/ports/main.cc

@@ -1,6 +1,6 @@
 #include <iostream>
 
-#include "reactor-cpp/reactor-cpp.hh"
+#include <reactor-cpp/reactor-cpp.hh>
 
 using namespace reactor;
 using namespace std::chrono_literals;
@@ -38,6 +38,7 @@ class Counter : public Reactor {
   Output<int> count{"count", this};     // NOLINT
 
   void assemble() override {
+    std::cout << "assemble Counter" << std::endl << std::flush;
     r_trigger.declare_trigger(&trigger);
     r_trigger.declare_antidependency(&count);
   }
@@ -53,12 +54,15 @@ class Printer : public Reactor {
   Reaction r_value{"r_value", 1, this, [this]() { on_value(); }};
 
 public:
-  Input<int> value{"value", this}; // NOLINT
-
+  Input<int> value{"value", this};     // NOLINT
+  Input<int> forward{"forward", this}; // NOLINT
   Printer(const std::string& name, Environment* env)
       : Reactor(name, env) {}
 
-  void assemble() override { r_value.declare_trigger(&value); }
+  void assemble() override {
+    std::cout << "assemble Printer" << std::endl << std::flush;
+    r_value.declare_trigger(&value);
+  }
 
   void on_value() { std::cout << this->name() << ": " << *value.get() << std::endl; }
 };
@@ -84,6 +88,7 @@ class Adder : public Reactor {
   void add() {
     if (i1.is_present() && i2.is_present()) {
       sum.set(*i1.get() + *i2.get());
+      std::cout << "setting sum" << std::endl;
     }
   }
 };
@@ -94,23 +99,49 @@ auto main() -> int {
   Trigger trigger1{"t1", &env, 1s};
   Counter counter1{"c1", &env};
   Printer printer1{"p1", &env};
-  trigger1.trigger.bind_to(&counter1.trigger);
-  counter1.count.bind_to(&printer1.value);
+
+  auto trigger1_trigger = env.register_port(&trigger1.trigger);
+  auto counter1_trigger = env.register_port(&counter1.trigger);
+  auto counter1_count = env.register_port(&counter1.count);
+  auto print1_value = env.register_port(&printer1.value);
+  // trigger1.trigger.set_inward_binding(&counter1.trigger);
+  // counter1.count.set_inward_binding(&printer1.value);
+
+  env.draw_connection(trigger1_trigger, counter1_trigger, ConnectionProperties{});
+  env.draw_connection(counter1_count, print1_value, ConnectionProperties{});
 
   Trigger trigger2{"t2", &env, 2s};
   Counter counter2{"c2", &env};
   Printer printer2{"p2", &env};
-  trigger2.trigger.bind_to(&counter2.trigger);
-  counter2.count.bind_to(&printer2.value);
+
+  auto trigger2_trigger = env.register_port(&trigger2.trigger);
+  auto counter2_trigger = env.register_port(&counter2.trigger);
+  auto counter2_count = env.register_port(&counter2.count);
+  auto printer2_value = env.register_port(&printer2.value);
+
+  // trigger2.trigger.set_inward_binding(&counter2.trigger);
+  // counter2.count.set_inward_binding(&printer2.value);
+  env.draw_connection(trigger2_trigger, counter2_trigger, ConnectionProperties{});
+  env.draw_connection(counter2_count, printer2_value, ConnectionProperties{});
 
   Adder add{"add", &env};
   Printer p_add{"p_add", &env};
-  counter1.count.bind_to(&add.i1);
-  counter2.count.bind_to(&add.i2);
-  add.sum.bind_to(&p_add.value);
 
+  auto add_i1 = env.register_port(&add.i1);
+  auto add_i2 = env.register_port(&add.i2);
+  auto add_sum = env.register_port(&add.sum);
+  auto p_add_forward = env.register_port(&p_add.forward);
+  auto p_add_value = env.register_port(&p_add.value);
+
+  env.draw_connection(counter1_count, add_i1, ConnectionProperties{});
+  env.draw_connection(counter2_count, add_i2, ConnectionProperties{});
+  env.draw_connection(add_sum, p_add_forward, ConnectionProperties{ConnectionType::Delayed, 10s, nullptr});
+  env.draw_connection(p_add_forward, p_add_value, ConnectionProperties{ConnectionType::Delayed, 5s, nullptr});
+
+  std::cout << "assemble" << std::endl << std::flush;
   env.assemble();
 
+  std::cout << "optimize" << std::endl << std::flush;
   auto thread = env.startup();
   thread.join();
 

examples/power_train/main.cc

@@ -165,14 +165,14 @@ auto main() -> int {
   Engine engine{&env};
 
   env.assemble();
-  left_pedal.angle.bind_to(&brake_control.angle);
-  left_pedal.on_off.bind_to(&engine_control.on_off);
-  brake_control.force.bind_to(&brakes.force);
-  right_pedal.angle.bind_to(&engine_control.angle);
-  engine_control.check.bind_to(&right_pedal.check);
-  engine_control.torque.bind_to(&engine.torque);
-
-  env.export_dependency_graph("graph.dot");
+  left_pedal.angle.set_inward_binding(&brake_control.angle);
+  left_pedal.on_off.set_inward_binding(&engine_control.on_off);
+  brake_control.force.set_inward_binding(&brakes.force);
+  right_pedal.angle.set_inward_binding(&engine_control.angle);
+  engine_control.check.set_inward_binding(&right_pedal.check);
+  engine_control.torque.set_inward_binding(&engine.torque);
+
+  // TODO: env.export_dependency_graph("graph.dot");
 
   auto thread = env.startup();
   thread.join();

include/reactor-cpp/action.hh

@@ -14,6 +14,7 @@
 #include "fwd.hh"
 #include "logical_time.hh"
 #include "reactor.hh"
+#include "reactor_element.hh"
 #include "time_barrier.hh"
 #include "value_ptr.hh"
 
@@ -27,7 +28,7 @@ class BaseAction : public ReactorElement {
 private:
   std::set<Reaction*> triggers_{};
   std::set<Reaction*> schedulers_{};
-  const Duration min_delay_{};
+  const Duration min_delay_{0};
   const bool logical_{true};
   bool present_{false};
 
@@ -48,22 +49,13 @@ protected:
    * indicates that the tag is safe to process.
    */
   virtual auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock,
-                           const std::function<bool(void)>& abort_waiting) -> bool {
-    reactor_assert(!logical_);
-    reactor_assert(lock.owns_lock());
-    return PhysicalTimeBarrier::acquire_tag(tag, lock, environment()->scheduler(), abort_waiting);
-  }
+                           const std::function<bool(void)>& abort_waiting) -> bool;
 
   BaseAction(const std::string& name, Reactor* container, bool logical, Duration min_delay)
       : ReactorElement(name, ReactorElement::Type::Action, container)
       , min_delay_(min_delay)
       , logical_(logical) {}
-  BaseAction(const std::string& name, Environment* environment, bool logical, Duration min_delay)
-      : ReactorElement(name, ReactorElement::Type::Action, environment)
-      , min_delay_(min_delay)
-      , logical_(logical) {
-    environment->register_input_action(this);
-  }
+  BaseAction(const std::string& name, Environment* environment, bool logical, Duration min_delay);
 
 public:
   [[nodiscard]] auto inline triggers() const noexcept -> const auto& { return triggers_; }
@@ -150,12 +142,7 @@ public:
 
 template <class T> class PhysicalAction : public Action<T> {
 public:
-  PhysicalAction(const std::string& name, Reactor* container)
-      : Action<T>(name, container, false, Duration::zero()) {
-    // all physical actions act as input actions to the program as they can be
-    // scheduled from external threads
-    container->environment()->register_input_action(this);
-  }
+  PhysicalAction(const std::string& name, Reactor* container);
 };
 
 template <class T> class LogicalAction : public Action<T> {
@@ -166,8 +153,8 @@ public:
 
 class Timer : public BaseAction {
 private:
-  const Duration offset_{};
-  const Duration period_{};
+  const Duration offset_{0};
+  const Duration period_{0};
 
   void cleanup() noexcept final;
 

include/reactor-cpp/assert.hh

@@ -75,6 +75,7 @@ template <typename E> constexpr auto extract_value(E enum_value) -> typename std
 }
 
 void assert_phase([[maybe_unused]] const ReactorElement* ptr, [[maybe_unused]] Phase phase);
+void assert_phase([[maybe_unused]] const Environment* ptr, [[maybe_unused]] Phase phase);
 
 } // namespace reactor
 

include/reactor-cpp/connection.hh

@@ -39,15 +39,19 @@ protected:
   [[nodiscard]] auto upstream_port() -> auto* { return upstream_port_; }
   [[nodiscard]] auto upstream_port() const -> const auto* { return upstream_port_; }
 
-  virtual auto upstream_set_callback() noexcept -> PortCallback = 0;
-
 public:
+  virtual auto upstream_set_callback() noexcept -> PortCallback = 0;
   virtual void bind_upstream_port(Port<T>* port) {
     reactor_assert(upstream_port_ == nullptr);
     upstream_port_ = port;
-    port->register_set_callback(upstream_set_callback());
   }
 
+  virtual void bind_downstream_ports(const std::vector<BasePort*>& ports) {
+    // with C++23 we can use insert_rage here
+    for ([[maybe_unused]] auto* port : ports) { // NOLINT
+      this->downstream_ports_.insert(static_cast<Port<T>*>(port));
+    }
+  }
   virtual void bind_downstream_port(Port<T>* port) {
     [[maybe_unused]] bool result = this->downstream_ports_.insert(port).second;
     reactor_assert(result);
@@ -110,31 +114,28 @@ protected:
 
   EnclaveConnection(const std::string& name, Environment* enclave, const Duration& delay)
       : BaseDelayedConnection<T>(name, enclave, false, delay)
-      , log_{this->fqn()}
-      , logical_time_barrier_(enclave->scheduler()) {}
+      , logical_time_barrier_(enclave->scheduler())
+      , log_{this->fqn()} {}
 
 public:
   EnclaveConnection(const std::string& name, Environment* enclave)
       : BaseDelayedConnection<T>(name, enclave, false, Duration::zero())
-      , log_{this->fqn()}
-      , logical_time_barrier_(enclave->scheduler()) {}
+      , logical_time_barrier_(enclave->scheduler())
+      , log_{this->fqn()} {};
 
   inline auto upstream_set_callback() noexcept -> PortCallback override {
-    return [this](const BasePort& port) {
-      // We know that port must be of type Port<T>
-      auto& typed_port = reinterpret_cast<const Port<T>&>(port); // NOLINT
-      const auto* scheduler = port.environment()->scheduler();
+    return [&](const BasePort& port) { // NOLINT unused this
       // This callback will be called from a reaction executing in the context
       // of the upstream port. Hence, we can retrieve the current tag directly
       // without locking.
-      auto tag = Tag::from_logical_time(scheduler->logical_time());
-      bool result{false};
       if constexpr (std::is_same<T, void>::value) {
-        result = this->schedule_at(tag);
+        this->schedule_at(Tag::from_logical_time(port.environment()->scheduler()->logical_time()));
       } else {
-        result = this->schedule_at(std::move(typed_port.get()), tag);
+        // We know that port must be of type Port<T>
+        auto& typed_port = reinterpret_cast<const Port<T>&>(port); // NOLINT
+        this->schedule_at(std::move(typed_port.get()),
+                          Tag::from_logical_time(port.environment()->scheduler()->logical_time()));
       }
-      reactor_assert(result);
     };
   }
 
@@ -186,21 +187,21 @@ public:
       : EnclaveConnection<T>(name, enclave, delay) {}
 
   inline auto upstream_set_callback() noexcept -> PortCallback override {
-    return [this](const BasePort& port) {
-      // We know that port must be of type Port<T>
-      auto& typed_port = reinterpret_cast<const Port<T>&>(port); // NOLINT
-      const auto* scheduler = port.environment()->scheduler();
+    return [&](const BasePort& port) { // NOLINT unused this
       // This callback will be called from a reaction executing in the context
       // of the upstream port. Hence, we can retrieve the current tag directly
       // without locking.
-      auto tag = Tag::from_logical_time(scheduler->logical_time()).delay(this->min_delay());
-      bool result{false};
+
       if constexpr (std::is_same<T, void>::value) {
-        result = this->schedule_at(tag);
+        this->schedule_at(
+            Tag::from_logical_time(port.environment()->scheduler()->logical_time()).delay(this->min_delay()));
       } else {
-        result = this->schedule_at(std::move(typed_port.get()), tag);
+        // We know that port must be of type Port<T>
+        auto& typed_port = reinterpret_cast<const Port<T>&>(port); // NOLINT
+        this->schedule_at(
+            std::move(typed_port.get()),
+            Tag::from_logical_time(port.environment()->scheduler()->logical_time()).delay(this->min_delay()));
       }
-      reactor_assert(result);
     };
   }
 

include/reactor-cpp/connection_properties.hh

@@ -0,0 +1,26 @@
+/*
+ * Copyright (C) 2023 TU Dresden
+ * All rights reserved.
+ *
+ * Authors:
+ *   Tassilo Tanneberger
+ */
+
+#ifndef REACTOR_CPP_CONNECTION_PROPERTIES_HH
+#define REACTOR_CPP_CONNECTION_PROPERTIES_HH
+
+#include "fwd.hh"
+#include "logical_time.hh"
+
+namespace reactor {
+
+enum ConnectionType { Normal, Delayed, Enclaved, Physical, DelayedEnclaved, PhysicalEnclaved, Plugin, Invalid };
+struct ConnectionProperties {
+  ConnectionType type_ = ConnectionType::Normal;
+  Duration delay_{0};
+  Environment* enclave_{nullptr};
+};
+
+} // namespace reactor
+
+#endif // REACTOR_CPP_CONNECTION_PROPERTIES_HH