Merge pull request #51 from lf-lang/port-graph

Port graph

Author: cmnrd

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,31 @@ 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);
+
+  env.draw_connection(trigger1.trigger, counter1.trigger, ConnectionProperties{});
+  env.draw_connection(counter1.count, printer1.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);
+
+  // 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);
 
+  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

@@ -164,13 +164,14 @@ auto main() -> int {
   Brake brakes{&env};
   Engine engine{&env};
 
+  env.draw_connection(left_pedal.angle, brake_control.angle, ConnectionProperties{});
+  env.draw_connection(left_pedal.angle, engine_control.on_off, ConnectionProperties{});
+  env.draw_connection(brake_control.force, brakes.force, ConnectionProperties{});
+  env.draw_connection(right_pedal.angle, engine_control.angle, ConnectionProperties{});
+  env.draw_connection(engine_control.check, right_pedal.check, ConnectionProperties{});
+  env.draw_connection(engine_control.torque, engine.torque, ConnectionProperties{});
+
   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");
 

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_; }
@@ -81,8 +73,8 @@ template <class T> class Action : public BaseAction {
 private:
   ImmutableValuePtr<T> value_ptr_{nullptr};
 
-  std::map<Tag, ImmutableValuePtr<T>> events_;
-  std::mutex mutex_events_;
+  std::map<Tag, ImmutableValuePtr<T>> events_{};
+  std::mutex mutex_events_{};
 
 protected:
   void setup() noexcept override;
@@ -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/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,14 +114,14 @@ 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) {
@@ -128,7 +132,7 @@ public:
       // 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};
+      [[maybe_unused]] bool result{false}; // NOLINT value set is not used
       if constexpr (std::is_same<T, void>::value) {
         result = this->schedule_at(tag);
       } else {
@@ -186,21 +190,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,34 @@
+/*
+ * 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};
+
+  auto operator<(const ConnectionProperties& elem2) const noexcept -> bool {
+    return (this->type_ < elem2.type_) || (this->type_ == elem2.type_ && this->delay_ < elem2.delay_);
+  }
+
+  auto operator==(const ConnectionProperties& elem2) const noexcept -> bool {
+    return this->type_ == elem2.type_ && this->delay_ == elem2.delay_;
+  }
+};
+
+} // namespace reactor
+
+#endif // REACTOR_CPP_CONNECTION_PROPERTIES_HH

include/reactor-cpp/environment.hh

@@ -13,10 +13,11 @@
 #include <string>
 #include <vector>
 
+#include "connection_properties.hh"
 #include "fwd.hh"
+#include "graph.hh"
 #include "reactor-cpp/logging.hh"
 #include "reactor-cpp/time.hh"
-#include "reactor.hh"
 #include "scheduler.hh"
 
 namespace reactor {
@@ -58,6 +59,9 @@ private:
 
   const Duration timeout_{};
 
+  Graph<BasePort*, ConnectionProperties> graph_{};
+  Graph<BasePort*, ConnectionProperties> optimized_graph_{};
+
   void build_dependency_graph(Reactor* reactor);
   void calculate_indexes();
 
@@ -72,7 +76,24 @@ public:
 
   auto name() -> const std::string& { return name_; }
 
+  // this method draw a connection between two graph elements with some properties
+  template <class T> void draw_connection(Port<T>& source, Port<T>& sink, ConnectionProperties properties) {
+    this->draw_connection(&source, &sink, properties);
+  }
+
+  template <class T> void draw_connection(Port<T>* source, Port<T>* sink, ConnectionProperties properties) {
+    if (top_environment_ == nullptr || top_environment_ == this) {
+      log::Debug() << "drawing connection: " << source << " --> " << sink;
+      graph_.add_edge(source, sink, properties);
+    } else {
+      top_environment_->draw_connection(source, sink, properties);
+    }
+  }
+
+  void optimize();
+
   void register_reactor(Reactor* reactor);
+  void register_port(BasePort* port) noexcept;
   void register_input_action(BaseAction* action);
   void assemble();
   auto startup() -> std::thread;