Merge pull request #44 from lf-lang/enclave-connections

Enclave connections and enclave coordination

Author: cmnrd

include/reactor-cpp/action.hh

@@ -9,12 +9,15 @@
 #ifndef REACTOR_CPP_ACTION_HH
 #define REACTOR_CPP_ACTION_HH
 
+#include "assert.hh"
 #include "environment.hh"
 #include "fwd.hh"
 #include "logical_time.hh"
 #include "reactor.hh"
+#include "time_barrier.hh"
 #include "value_ptr.hh"
 
+#include <condition_variable>
 #include <map>
 #include <mutex>
 
@@ -35,6 +38,21 @@ protected:
   virtual void setup() noexcept { present_ = true; }
   virtual void cleanup() noexcept { present_ = false; }
 
+  /**
+   * Use the given condition variable and lock to wait until the given tag it
+   * safe to process. The waiting is interrupted when the condition variable is
+   * notified (or has a spurious wakeup) and a call to the given `abort_waiting`
+   * function returns true. or until the condition variable is notified.
+   *
+   * Returns false if the wait was interrupted and true otherwise. True
+   * indicates that the tag is safe to process.
+   */
+  virtual auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
+                           const std::function<bool(void)>& abort_waiting) -> bool {
+    reactor_assert(!logical_);
+    return PhysicalTimeBarrier::acquire_tag(tag, lock, cv, abort_waiting);
+  }
+
   BaseAction(const std::string& name, Reactor* container, bool logical, Duration min_delay)
       : ReactorElement(name, ReactorElement::Type::Action, container)
       , min_delay_(min_delay)
@@ -94,6 +112,7 @@ public:
   void shutdown() final {}
 
   template <class Dur = Duration> void schedule(const ImmutableValuePtr<T>& value_ptr, Dur delay = Dur::zero());
+  auto schedule_at(const ImmutableValuePtr<T>& value_ptr, const Tag& tag) -> bool;
 
   template <class Dur = Duration> void schedule(MutableValuePtr<T>&& value_ptr, Dur delay = Dur::zero()) {
     schedule(ImmutableValuePtr<T>(std::forward<MutableValuePtr<T>>(value_ptr)), delay);
@@ -122,6 +141,7 @@ protected:
 
 public:
   template <class Dur = Duration> void schedule(Dur delay = Dur::zero());
+  auto schedule_at(const Tag& tag) -> bool;
 
   void startup() final {}
   void shutdown() final {}

include/reactor-cpp/assert.hh

@@ -67,16 +67,16 @@ constexpr inline void validate([[maybe_unused]] bool condition, [[maybe_unused]]
   }
 }
 
-// Use plain assert on Windows and when assertions are enabled.
-// Otherwise, the locations of assertion errors are not properly reported.
-#if defined(_WIN32) && !defined(NDEBUG)
-#define reactor_assert(condition) assert(condition)
+// assert macro that avoids unused variable warnings
+#ifdef NDEBUG
+// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
+#define reactor_assert(x)                                                                                              \
+  do {                                                                                                                 \
+    (void)sizeof(x);                                                                                                   \
+  } while (0)
 #else
-constexpr inline void reactor_assert([[maybe_unused]] bool condition) {
-  if constexpr (runtime_assertion) { // NOLINT
-    assert(condition);               // NOLINT
-  }
-}
+// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
+#define reactor_assert(x) assert(x)
 #endif
 
 template <typename E> constexpr auto extract_value(E enum_value) -> typename std::underlying_type<E>::type {

include/reactor-cpp/connection.hh

@@ -13,10 +13,13 @@
 #include "action.hh"
 #include "assert.hh"
 #include "environment.hh"
+#include "fwd.hh"
+#include "logical_time.hh"
 #include "port.hh"
 #include "reaction.hh"
 #include "reactor.hh"
 #include "time.hh"
+#include "time_barrier.hh"
 
 namespace reactor {
 
@@ -28,26 +31,32 @@ private:
 protected:
   Connection(const std::string& name, Reactor* container, bool is_logical, Duration min_delay)
       : Action<T>(name, container, is_logical, min_delay) {}
+  Connection(const std::string& name, Environment* environment, bool is_logical, Duration min_delay)
+      : Action<T>(name, environment, is_logical, min_delay) {}
 
   [[nodiscard]] auto downstream_ports() -> auto& { return downstream_ports_; }
   [[nodiscard]] auto downstream_ports() const -> const auto& { return downstream_ports_; }
+  [[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:
-  void bind_upstream_port(Port<T>* port) {
+  virtual 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); };
+  virtual void bind_downstream_port(Port<T>* port) { reactor_assert(this->downstream_ports_.insert(port).second); };
 };
 
 template <class T> class BaseDelayedConnection : public Connection<T> {
-public:
+protected:
   BaseDelayedConnection(const std::string& name, Reactor* container, bool is_logical, Duration delay)
       : Connection<T>(name, container, is_logical, delay) {}
+  BaseDelayedConnection(const std::string& name, Environment* environment, bool is_logical, Duration delay)
+      : Connection<T>(name, environment, is_logical, delay) {}
 
   inline auto upstream_set_callback() noexcept -> PortCallback override {
     return [this](const BasePort& port) {
@@ -61,6 +70,7 @@ public:
     };
   }
 
+public:
   void setup() noexcept override {
     Action<T>::setup();
 
@@ -88,6 +98,144 @@ public:
       : BaseDelayedConnection<T>(name, container, false, delay) {}
 };
 
+template <class T> class EnclaveConnection : public BaseDelayedConnection<T> {
+private:
+  LogicalTimeBarrier logical_time_barrier_;
+
+protected:
+  log::NamedLogger log_; // NOLINT
+
+  EnclaveConnection(const std::string& name, Environment* enclave, const Duration& delay)
+      : BaseDelayedConnection<T>(name, enclave, false, delay)
+      , log_{this->fqn()} {}
+
+public:
+  EnclaveConnection(const std::string& name, Environment* enclave)
+      : BaseDelayedConnection<T>(name, enclave, false, Duration::zero())
+      , 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();
+      // 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);
+      } else {
+        result = this->schedule_at(std::move(typed_port.get()), tag);
+      }
+      reactor_assert(result);
+    };
+  }
+
+  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
+                          const std::function<bool(void)>& abort_waiting) -> bool override {
+    log_.debug() << "downstream tries to acquire tag " << tag;
+
+    if (this->upstream_port() == nullptr) {
+      return true;
+    }
+
+    if (logical_time_barrier_.try_acquire_tag(tag)) {
+      return true;
+    }
+
+    // Insert an empty event into the upstream event queue. This ensures that we
+    // will get notified and woken up as soon as the tag becomes safe to process.
+    // It is important to unlock the mutex here. Otherwise we could enter a deadlock as
+    // scheduling the upstream event also requires holding the upstream mutex.
+    lock.unlock();
+    bool result = this->upstream_port()->environment()->scheduler()->schedule_empty_async_at(tag);
+    lock.lock();
+
+    // If inserting the empty event was not successful, then this is because the upstream
+    // scheduler already processes a later event. In this case, it is safe to assume that
+    // the tag is acquired.
+    if (!result) {
+      return true;
+    }
+
+    // Wait until we receive a release_tag message from upstream
+    return logical_time_barrier_.acquire_tag(tag, lock, cv, abort_waiting);
+  }
+
+  void bind_upstream_port(Port<T>* port) override {
+    Connection<T>::bind_upstream_port(port);
+    port->environment()->scheduler()->register_release_tag_callback([this](const LogicalTime& tag) {
+      logical_time_barrier_.release_tag(tag);
+      log_.debug() << "upstream released tag " << tag;
+      this->environment()->scheduler()->notify();
+    });
+  }
+};
+
+template <class T> class DelayedEnclaveConnection : public EnclaveConnection<T> {
+public:
+  DelayedEnclaveConnection(const std::string& name, Environment* enclave, Duration delay)
+      : 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();
+      // 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);
+      } else {
+        result = this->schedule_at(std::move(typed_port.get()), tag);
+      }
+      reactor_assert(result);
+    };
+  }
+
+  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
+                          const std::function<bool(void)>& abort_waiting) -> bool override {
+    // Since this is a delayed connection, we can go back in time and need to
+    // acquire the latest upstream tag that can create an event at the given
+    // tag. We also need to consider that given a delay d and a tag g=(t, n),
+    // for any value of n, g + d = (t, 0). Hence, we need to quire a tag with
+    // the highest possible microstep value.
+    auto upstream_tag = tag.subtract(this->min_delay());
+    return EnclaveConnection<T>::acquire_tag(upstream_tag, lock, cv, abort_waiting);
+  }
+};
+
+template <class T> class PhysicalEnclaveConnection : public EnclaveConnection<T> {
+public:
+  PhysicalEnclaveConnection(const std::string& name, Environment* enclave)
+      : EnclaveConnection<T>(name, enclave) {}
+
+  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()));
+      }
+    };
+  }
+
+  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
+                          const std::function<bool(void)>& abort_waiting) -> bool override {
+    this->log_.debug() << "downstream tries to acquire tag " << tag;
+    return PhysicalTimeBarrier::acquire_tag(tag, lock, cv, abort_waiting);
+  }
+
+  void bind_upstream_port(Port<T>* port) override { Connection<T>::bind_upstream_port(port); }
+};
+
 } // namespace reactor
 
 #endif // REACTOR_CPP_CONNECTION_HH

include/reactor-cpp/environment.hh

@@ -55,7 +55,7 @@ private:
 
   Scheduler scheduler_;
   Phase phase_{Phase::Construction};
-  TimePoint start_time_{};
+  Tag start_tag_{};
 
   const Duration timeout_{};
 
@@ -96,7 +96,7 @@ public:
   auto scheduler() noexcept -> Scheduler* { return &scheduler_; }
 
   [[nodiscard]] auto logical_time() const noexcept -> const LogicalTime& { return scheduler_.logical_time(); }
-  [[nodiscard]] auto start_time() const noexcept -> const TimePoint& { return start_time_; }
+  [[nodiscard]] auto start_tag() const noexcept -> const Tag& { return start_tag_; }
   [[nodiscard]] auto timeout() const noexcept -> const Duration& { return timeout_; }
 
   static auto physical_time() noexcept -> TimePoint { return get_physical_time(); }
@@ -105,6 +105,8 @@ public:
   [[nodiscard]] auto fast_fwd_execution() const noexcept -> bool { return fast_fwd_execution_; }
   [[nodiscard]] auto run_forever() const noexcept -> bool { return run_forever_; }
   [[nodiscard]] auto max_reaction_index() const noexcept -> unsigned int { return max_reaction_index_; }
+
+  friend Scheduler;
 };
 } // namespace reactor
 

include/reactor-cpp/impl/action_impl.hh

@@ -20,41 +20,68 @@ template <class T> template <class Dur> void Action<T>::schedule(const Immutable
   Duration time_delay = std::chrono::duration_cast<Duration>(delay);
   reactor::validate(time_delay >= Duration::zero(), "Schedule cannot be called with a negative delay!");
   reactor::validate(value_ptr != nullptr, "Actions may not be scheduled with a nullptr value!");
+
   auto* scheduler = environment()->scheduler();
   if (is_logical()) {
     time_delay += this->min_delay();
     auto tag = Tag::from_logical_time(scheduler->logical_time()).delay(time_delay);
-    events_[tag] = value_ptr;
+
     scheduler->schedule_sync(this, tag);
+    events_[tag] = value_ptr;
   } else {
-    {
-      std::lock_guard<std::mutex> lock(mutex_events_);
-      // We must call schedule_async while holding the mutex, because otherwise
-      // the scheduler could already start processing the event that we schedule
-      // and call setup() on this action before we insert the value in events_.
-      // Holding both the local mutex mutex_events_ and them scheduler mutes (in
-      // schedule async) should not lead to a deadlock as the scheduler will
-      // only hold one of the two mutexes at once.
-      auto tag = scheduler->schedule_async(this, time_delay);
-      events_[tag] = value_ptr;
-    }
+    std::lock_guard<std::mutex> lock{mutex_events_};
+    // We must call schedule_async while holding the mutex, because otherwise
+    // the scheduler could already start processing the event that we schedule
+    // and call setup() on this action before we insert the value in events_.
+    // Holding both the local mutex mutex_events_ and the scheduler mutex (in
+    // schedule async) should not lead to a deadlock as the scheduler will
+    // only hold one of the two mutexes at once.
+    auto tag = scheduler->schedule_async(this, time_delay);
+    events_[tag] = value_ptr;
   }
 }
 
 template <class Dur> void Action<void>::schedule(Dur delay) {
-  auto time_delay = std::chrono::duration_cast<Duration>(delay);
+  Duration time_delay = std::chrono::duration_cast<Duration>(delay);
   reactor::validate(time_delay >= Duration::zero(), "Schedule cannot be called with a negative delay!");
   auto* scheduler = environment()->scheduler();
   if (is_logical()) {
     time_delay += this->min_delay();
     auto tag = Tag::from_logical_time(scheduler->logical_time()).delay(time_delay);
     scheduler->schedule_sync(this, tag);
   } else {
-    // physical action
     scheduler->schedule_async(this, time_delay);
   }
 }
 
+template <class T> auto Action<T>::schedule_at(const ImmutableValuePtr<T>& value_ptr, const Tag& tag) -> bool {
+  reactor::validate(value_ptr != nullptr, "Actions may not be scheduled with a nullptr value!");
+
+  auto* scheduler = environment()->scheduler();
+  if (is_logical()) {
+    if (tag <= scheduler->logical_time()) {
+      return false;
+    }
+
+    scheduler->schedule_sync(this, tag);
+    events_[tag] = value_ptr;
+  } else {
+    std::lock_guard<std::mutex> lock{mutex_events_};
+    // We must call schedule_async while holding the mutex, because otherwise
+    // the scheduler could already start processing the event that we schedule
+    // and call setup() on this action before we insert the value in events_.
+    // Holding both the local mutex mutex_events_ and the scheduler mutex (in
+    // schedule async) should not lead to a deadlock as the scheduler will
+    // only hold one of the two mutexes at once.
+    bool result = scheduler->schedule_async_at(this, tag);
+    if (result) {
+      events_[tag] = value_ptr;
+    }
+    return result;
+  }
+  return true;
+}
+
 template <class T> void Action<T>::setup() noexcept {
   BaseAction::setup();
   if (value_ptr_ == nullptr) { // only do this once, even if the action was triggered multiple times