Merge pull request #38 from lf-lang/network-interfaces

Native runtime implementation of delayed and physical connections

Author: cmnrd

include/reactor-cpp/action.hh

@@ -40,15 +40,10 @@ protected:
 
 public:
   [[nodiscard]] auto inline triggers() const noexcept -> const auto& { return triggers_; }
-
   [[nodiscard]] auto inline schedulers() const noexcept -> const auto& { return schedulers_; }
-
   [[nodiscard]] auto inline is_logical() const noexcept -> bool { return logical_; }
-
   [[nodiscard]] auto inline is_physical() const noexcept -> bool { return !logical_; }
-
   [[nodiscard]] auto inline min_delay() const noexcept -> Duration { return min_delay_; }
-
   [[nodiscard]] auto inline is_present() const noexcept -> bool { return present_; }
 
   friend class Reaction;
@@ -62,14 +57,29 @@ private:
   std::map<Tag, ImmutableValuePtr<T>> events_;
   std::mutex mutex_events_;
 
-  void setup() noexcept final;
+protected:
+  void setup() noexcept override;
   void cleanup() noexcept final;
 
-protected:
   Action(const std::string& name, Reactor* container, bool logical, Duration min_delay)
       : BaseAction(name, container, logical, min_delay) {}
 
 public:
+  // Normally, we should lock the mutex while moving to make this
+  // fully thread safe. However, we rely assembly happening before
+  // execution and hence can ignore the mutex.
+  Action(Action&& action) noexcept
+      : BaseAction(std::move(action)) {}
+  auto operator=(Action&& action) noexcept -> Action& {
+    BaseAction::operator=(std::move(action));
+    return *this;
+  }
+
+  Action(const Action& action) = delete;
+  auto operator=(const Action& action) -> Action& = delete;
+
+  ~Action() override = default;
+
   void startup() final {}
   void shutdown() final {}
 

include/reactor-cpp/connection.hh

@@ -0,0 +1,93 @@
+/*
+ * Copyright (C) 2023 TU Dresden
+ * All rights reserved.
+ *
+ * Authors:
+ *   Tassilo Tanneberger
+ *   Christian Menard
+ */
+
+#ifndef REACTOR_CPP_CONNECTION_HH
+#define REACTOR_CPP_CONNECTION_HH
+
+#include "action.hh"
+#include "assert.hh"
+#include "environment.hh"
+#include "port.hh"
+#include "reaction.hh"
+#include "reactor.hh"
+#include "time.hh"
+
+namespace reactor {
+
+template <class T> class Connection : public Action<T> {
+private:
+  Port<T>* upstream_port_{nullptr};
+  std::set<Port<T>*> downstream_ports_{};
+
+protected:
+  Connection(const std::string& name, Reactor* container, bool is_logical, Duration min_delay)
+      : Action<T>(name, container, is_logical, min_delay) {}
+
+  [[nodiscard]] auto downstream_ports() -> auto& { return downstream_ports_; }
+  [[nodiscard]] auto downstream_ports() const -> const auto& { return downstream_ports_; }
+
+  virtual auto upstream_set_callback() noexcept -> PortCallback = 0;
+
+public:
+  void bind_upstream_port(Port<T>* port) {
+    reactor_assert(upstream_port_ == nullptr);
+    upstream_port_ = port;
+    port->register_set_callback(upstream_set_callback());
+  }
+
+  void bind_downstream_port(Port<T>* port) { reactor_assert(this->downstream_ports_.insert(port).second); };
+};
+
+template <class T> class BaseDelayedConnection : public Connection<T> {
+public:
+  BaseDelayedConnection(const std::string& name, Reactor* container, bool is_physical, Duration delay)
+      : Connection<T>(name, container, is_physical, 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
+      if constexpr (std::is_same<T, void>::value) {
+        this->schedule();
+      } else {
+        this->schedule(std::move(typed_port.get()));
+      }
+    };
+  }
+
+  void setup() noexcept override {
+    Action<T>::setup();
+
+    if constexpr (std::is_same<T, void>::value) {
+      for (auto port : this->downstream_ports()) {
+        port->set();
+      }
+    } else {
+      for (auto port : this->downstream_ports()) {
+        port->set(std::move(this->get()));
+      }
+    }
+  }
+};
+
+template <class T> class DelayedConnection : public BaseDelayedConnection<T> {
+public:
+  DelayedConnection(const std::string& name, Reactor* container, Duration delay)
+      : BaseDelayedConnection<T>(name, container, true, delay) {}
+};
+
+template <class T> class PhysicalConnection : public BaseDelayedConnection<T> {
+public:
+  PhysicalConnection(const std::string& name, Reactor* container, Duration delay)
+      : BaseDelayedConnection<T>(name, container, false, delay) {}
+};
+
+} // namespace reactor
+
+#endif // REACTOR_CPP_CONNECTION_HH

include/reactor-cpp/fwd.hh

@@ -9,6 +9,8 @@
 #ifndef REACTOR_CPP_FWD_HH
 #define REACTOR_CPP_FWD_HH
 
+#include <functional>
+
 namespace reactor {
 
 class BaseAction;
@@ -22,6 +24,8 @@ class Tag;
 template <class T> class Action;
 template <class T> class Port;
 
+using PortCallback = std::function<void(const BasePort&)>;
+
 } // namespace reactor
 
 #endif // REACTOR_CPP_FWD_HH

include/reactor-cpp/impl/port_impl.hh

@@ -32,6 +32,7 @@ template <class T> void Port<T>::set(const ImmutableValuePtr<T>& value_ptr) {
   reactor::validate(!has_inward_binding(), "set() may only be called on ports that do not have an inward "
                                            "binding!");
   reactor::validate(value_ptr != nullptr, "Ports may not be set to nullptr!");
+
   auto scheduler = environment()->scheduler();
   this->value_ptr_ = std::move(value_ptr);
   scheduler->set_port(this);

include/reactor-cpp/multiport.hh

@@ -11,39 +11,49 @@
 
 #include <algorithm>
 #include <atomic>
+#include <cstddef>
 #include <iostream>
 #include <type_traits>
 #include <vector>
 
+#include "assert.hh"
+#include "fwd.hh"
+
 namespace reactor {
 
 class BaseMultiport { // NOLINT cppcoreguidelines-special-member-functions,-warnings-as-errors
-protected:
-  std::atomic<std::size_t> size_{0};         // NOLINT
-  std::vector<std::size_t> present_ports_{}; // NOLINT
+private:
+  std::atomic<std::size_t> size_{0};
+  std::vector<std::size_t> present_ports_{};
 
-public:
-  BaseMultiport() = default;
-  ~BaseMultiport() = default;
+  // record that the port with the given index has been set
+  inline void set_present(std::size_t index);
 
-  // tells the parent multiport that this port has been set.
-  [[nodiscard]] inline auto set_present(std::size_t index) -> bool {
-    auto calculated_index = size_.fetch_add(1, std::memory_order_relaxed);
+  // reset the list of set port indexes
+  inline void reset() noexcept { size_.store(0, std::memory_order_relaxed); }
 
-    reactor_assert(calculated_index < present_ports_.size());
+  [[nodiscard]] auto get_set_callback(std::size_t index) noexcept -> PortCallback;
+  const PortCallback clean_callback_{[this]([[maybe_unused]] const BasePort& port) { this->reset(); }};
 
-    present_ports_[calculated_index] = index;
-    return true;
-  }
+  [[nodiscard]] auto get_clean_callback() const noexcept -> const PortCallback& { return clean_callback_; }
 
-  // resets parent multiport
-  inline void clear() noexcept { size_.store(0, std::memory_order_relaxed); }
+protected:
+  [[nodiscard]] inline auto present_ports() const -> const auto& { return present_ports_; }
+  [[nodiscard]] inline auto present_ports_size() const -> auto { return size_.load(); }
+
+  inline void present_ports_reserve(size_t n) { present_ports_.reserve(n); }
+
+  void register_port(BasePort& port, size_t idx);
+
+public:
+  BaseMultiport() = default;
+  ~BaseMultiport() = default;
 };
 
 template <class T, class A = std::allocator<T>>
 class Multiport : public BaseMultiport { // NOLINT cppcoreguidelines-special-member-functions
 protected:
-  std::vector<T> data_{}; // NOLINT cppcoreguidelines-non-private-member-variables-in-classes
+  std::vector<T> ports_{}; // NOLINT cppcoreguidelines-non-private-member-variables-in-classes
 
 public:
   using value_type = typename A::value_type;
@@ -56,29 +66,30 @@ public:
   ~Multiport() noexcept = default;
 
   auto operator==(const Multiport& other) const noexcept -> bool {
-    return std::equal(std::begin(data_), std::end(data_), std::begin(other.data_), std::end(other.data_));
+    return std::equal(std::begin(ports_), std::end(ports_), std::begin(other.ports_), std::end(other.ports_));
   }
   auto operator!=(const Multiport& other) const noexcept -> bool { return !(*this == other); };
-  inline auto operator[](std::size_t index) noexcept -> T& { return data_[index]; }
-  inline auto operator[](std::size_t index) const noexcept -> const T& { return data_[index]; }
+  inline auto operator[](std::size_t index) noexcept -> T& { return ports_[index]; }
+  inline auto operator[](std::size_t index) const noexcept -> const T& { return ports_[index]; }
 
-  inline auto begin() noexcept -> iterator { return data_.begin(); };
-  inline auto begin() const noexcept -> const_iterator { return data_.begin(); };
-  inline auto cbegin() const noexcept -> const_iterator { return data_.cbegin(); };
-  inline auto end() noexcept -> iterator { return data_.end(); };
-  inline auto end() const noexcept -> const_iterator { return data_.end(); };
-  inline auto cend() const noexcept -> const_iterator { return data_.cend(); };
+  inline auto begin() noexcept -> iterator { return ports_.begin(); };
+  inline auto begin() const noexcept -> const_iterator { return ports_.begin(); };
+  inline auto cbegin() const noexcept -> const_iterator { return ports_.cbegin(); };
+  inline auto end() noexcept -> iterator { return ports_.end(); };
+  inline auto end() const noexcept -> const_iterator { return ports_.end(); };
+  inline auto cend() const noexcept -> const_iterator { return ports_.cend(); };
 
-  inline auto size() const noexcept -> size_type { return data_.size(); };
-  [[nodiscard]] inline auto empty() const noexcept -> bool { return data_.empty(); };
+  inline auto size() const noexcept -> size_type { return ports_.size(); };
+  [[nodiscard]] inline auto empty() const noexcept -> bool { return ports_.empty(); };
 
   [[nodiscard]] inline auto present_indices_unsorted() const noexcept -> std::vector<std::size_t> {
-    return std::vector<std::size_t>(std::begin(present_ports_), std::begin(present_ports_) + size_.load());
+    return std::vector<std::size_t>(std::begin(present_ports()), std::begin(present_ports()) + present_ports_size());
   }
 
   [[nodiscard]] inline auto present_indices_sorted() const noexcept -> std::vector<std::size_t> {
-    std::sort(std::begin(present_ports_), std::begin(present_ports_) + size_.load());
-    return std::vector<std::size_t>(std::begin(present_ports_), std::begin(present_ports_) + size_.load());
+    std::vector<std::size_t> indices(std::begin(present_ports()), std::begin(present_ports()) + present_ports_size());
+    std::sort(std::begin(indices), std::end(indices));
+    return indices;
   }
 };
 
@@ -89,18 +100,18 @@ public:
   ~ModifableMultiport() = default;
 
   inline void reserve(std::size_t size) noexcept {
-    this->data_.reserve(size);
-    this->present_ports_.reserve(size);
+    this->ports_.reserve(size);
+    this->present_ports_reserve(size);
   }
 
   inline void push_back(const T& elem) noexcept {
-    this->data_.push_back(elem);
-    this->present_ports_.emplace_back(0);
+    this->ports_.push_back(elem);
+    this->register_port(this->ports_.back(), this->ports_.size() - 1);
   }
 
   template <class... Args> inline void emplace_back(Args&&... args) noexcept {
-    this->data_.emplace_back(args...);
-    this->present_ports_.emplace_back(0);
+    this->ports_.emplace_back(args...);
+    this->register_port(this->ports_.back(), this->ports_.size() - 1);
   }
 };
 } // namespace reactor