Merge pull request #49 from lf-lang/enclave-fix

Fix race condition in time barriers

Author: cmnrd

include/reactor-cpp/action.hh

@@ -47,10 +47,11 @@ protected:
    * 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,
+  virtual auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock,
                            const std::function<bool(void)>& abort_waiting) -> bool {
     reactor_assert(!logical_);
-    return PhysicalTimeBarrier::acquire_tag(tag, lock, cv, abort_waiting);
+    reactor_assert(lock.owns_lock());
+    return PhysicalTimeBarrier::acquire_tag(tag, lock, environment()->scheduler(), abort_waiting);
   }
 
   BaseAction(const std::string& name, Reactor* container, bool logical, Duration min_delay)

include/reactor-cpp/assert.hh

@@ -21,7 +21,7 @@ constexpr bool runtime_assertion = false;
 constexpr bool runtime_assertion = true;
 #endif
 
-#include "environment.hh"
+#include "fwd.hh"
 
 #include <cassert>
 #include <sstream>
@@ -37,7 +37,6 @@ constexpr bool runtime_assertion = true;
 #define reactor_assert(x) assert(x)
 
 namespace reactor {
-using EnvPhase = Environment::Phase;
 
 class ValidationError : public std::runtime_error {
 private:
@@ -74,31 +73,8 @@ template <typename E> constexpr auto extract_value(E enum_value) -> typename std
   return static_cast<typename std::underlying_type<E>::type>(enum_value);
 }
 
-inline void assert_phase([[maybe_unused]] const ReactorElement* ptr, [[maybe_unused]] EnvPhase phase) {
-  if constexpr (runtime_assertion) { // NOLINT
-    if (ptr->environment()->phase() != phase) {
-      auto enum_value_to_name = [](EnvPhase phase) -> std::string {
-        const std::map<EnvPhase, std::string> conversation_map = {
-            // NOLINT
-            {EnvPhase::Construction, "Construction"}, {EnvPhase::Assembly, "Assembly"},
-            {EnvPhase::Startup, "Startup"},           {EnvPhase::Execution, "Execution"},
-            {EnvPhase::Shutdown, "Shutdown"},         {EnvPhase::Deconstruction, "Deconstruction"}};
-        // in C++20 use .contains()
-        if (conversation_map.find(phase) != std::end(conversation_map)) {
-          return conversation_map.at(phase);
-        }
-        return "Unknown Phase: Value: " + std::to_string(extract_value(phase));
-      };
-#ifdef __linux__
-      print_debug_backtrace();
-#endif
+void assert_phase([[maybe_unused]] const ReactorElement* ptr, [[maybe_unused]] Phase phase);
 
-      // C++20 std::format
-      throw ValidationError("Expected Phase: " + enum_value_to_name(phase) +
-                            " Current Phase: " + enum_value_to_name(ptr->environment()->phase()));
-    }
-  }
-}
 } // namespace reactor
 
 #endif // REACTOR_CPP_ASSERT_HH

include/reactor-cpp/connection.hh

@@ -110,12 +110,14 @@ protected:
 
   EnclaveConnection(const std::string& name, Environment* enclave, const Duration& delay)
       : BaseDelayedConnection<T>(name, enclave, false, delay)
-      , log_{this->fqn()} {}
+      , log_{this->fqn()}
+      , logical_time_barrier_(enclave->scheduler()) {}
 
 public:
   EnclaveConnection(const std::string& name, Environment* enclave)
       : BaseDelayedConnection<T>(name, enclave, false, Duration::zero())
-      , log_{this->fqn()} {}
+      , log_{this->fqn()}
+      , logical_time_barrier_(enclave->scheduler()) {}
 
   inline auto upstream_set_callback() noexcept -> PortCallback override {
     return [this](const BasePort& port) {
@@ -136,8 +138,9 @@ public:
     };
   }
 
-  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
+  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock,
                           const std::function<bool(void)>& abort_waiting) -> bool override {
+    reactor_assert(lock.owns_lock());
     log_.debug() << "downstream tries to acquire tag " << tag;
 
     if (this->upstream_port() == nullptr) {
@@ -164,7 +167,7 @@ public:
     }
 
     // Wait until we receive a release_tag message from upstream
-    return logical_time_barrier_.acquire_tag(tag, lock, cv, abort_waiting);
+    return logical_time_barrier_.acquire_tag(tag, lock, abort_waiting);
   }
 
   void bind_upstream_port(Port<T>* port) override {
@@ -201,15 +204,15 @@ public:
     };
   }
 
-  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
+  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock,
                           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);
+    return EnclaveConnection<T>::acquire_tag(upstream_tag, lock, abort_waiting);
   }
 };
 
@@ -230,10 +233,10 @@ public:
     };
   }
 
-  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
+  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock,
                           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);
+    return PhysicalTimeBarrier::acquire_tag(tag, lock, this->environment()->scheduler(), abort_waiting);
   }
 
   void bind_upstream_port(Port<T>* port) override { Connection<T>::bind_upstream_port(port); }

include/reactor-cpp/environment.hh

@@ -26,10 +26,9 @@ constexpr unsigned int default_max_reaction_index = 0;
 constexpr bool default_run_forever = false;
 constexpr bool default_fast_fwd_execution = false;
 
-class Environment {
-public:
-  enum class Phase { Construction = 0, Assembly = 1, Startup = 2, Execution = 3, Shutdown = 4, Deconstruction = 5 };
+enum class Phase { Construction = 0, Assembly = 1, Startup = 2, Execution = 3, Shutdown = 4, Deconstruction = 5 };
 
+class Environment {
 private:
   using Dependency = std::pair<Reaction*, Reaction*>;
 

include/reactor-cpp/fwd.hh

@@ -16,8 +16,10 @@ namespace reactor {
 class BaseAction;
 class BasePort;
 class Environment;
+enum class Phase;
 class Reaction;
 class Reactor;
+class ReactorElement;
 class Scheduler;
 class Tag;
 

include/reactor-cpp/scheduler.hh

@@ -19,12 +19,14 @@
 #include <thread>
 #include <vector>
 
+#include "assert.hh"
 #include "fwd.hh"
 #include "logical_time.hh"
 #include "reactor-cpp/logging.hh"
 #include "reactor-cpp/time.hh"
 #include "safe_vector.hh"
 #include "semaphore.hh"
+#include "time.hh"
 
 namespace reactor {
 
@@ -176,9 +178,19 @@ public:
   auto schedule_async_at(BaseAction* action, const Tag& tag) -> bool;
   auto schedule_empty_async_at(const Tag& tag) -> bool;
 
+  auto inline lock() noexcept -> std::unique_lock<std::mutex> {
+    return std::unique_lock<std::mutex>(scheduling_mutex_);
+  }
   void inline notify() noexcept { cv_schedule_.notify_one(); }
-
-  auto inline lock() noexcept -> auto { return std::unique_lock<std::mutex>(scheduling_mutex_); }
+  void inline wait(std::unique_lock<std::mutex>& lock, const std::function<bool(void)>& predicate) noexcept {
+    reactor_assert(lock.owns_lock());
+    cv_schedule_.wait(lock, predicate);
+  };
+  auto inline wait_until(std::unique_lock<std::mutex>& lock, TimePoint time_point,
+                         const std::function<bool(void)>& predicate) noexcept -> bool {
+    reactor_assert(lock.owns_lock());
+    return cv_schedule_.wait_until(lock, time_point, predicate);
+  };
 
   void set_port(BasePort* port);
 

include/reactor-cpp/time_barrier.hh

@@ -11,11 +11,9 @@
 
 #include "fwd.hh"
 #include "logical_time.hh"
+#include "scheduler.hh"
 #include "time.hh"
-#include <atomic>
-#include <condition_variable>
 #include <functional>
-#include <mutex>
 
 namespace reactor {
 
@@ -38,36 +36,55 @@ public:
     return tag.time_point() < physical_time;
   }
 
-  static inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
-                                 const std::function<bool(void)>& abort_waiting) {
+  static inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, Scheduler* scheduler,
+                                 const std::function<bool(void)>& abort_waiting) -> bool {
     if (try_acquire_tag(tag)) {
       return true;
     }
-    return !cv.wait_until(lock, tag.time_point(), abort_waiting);
+    return !scheduler->wait_until(lock, tag.time_point(), abort_waiting);
   }
 };
 
 class LogicalTimeBarrier {
-  std::mutex mutex_;
+private:
   LogicalTime released_time_;
+  Scheduler* scheduler_;
 
+  /*
+   * Note that we use the main scheduler lock to protect our `released_time_`
+   * variable. We cannot use a local mutex here that only protects our local
+   * state. This is, because we also need to use a lock to wait on a condition
+   * variable. This lock must be the same lock as the one that is used when
+   * updating the wait condition. (See "An atomic predicate" in
+   * https://www.modernescpp.com/index.php/c-core-guidelines-be-aware-of-the-traps-of-condition-variables)
+   *
+   * Here we have two conditions in which we want to stop the wait:
+   *   1. When the tag that we are waiting for was released.
+   *   2. When the given function `abort_waiting` returns true. In a typical usage
+   *      scenario, this would be the case when a new event was scheduled.
+   *
+   * We need to make sure that both conditions are updated while holding the
+   * lock that we use for waiting on the condition variable. Our only option for
+   * this is to use the global scheduler lock.
+   */
 public:
+  LogicalTimeBarrier(Scheduler* scheduler)
+      : scheduler_(scheduler) {}
+
   inline void release_tag(const LogicalTime& tag) {
-    std::lock_guard lock(mutex_);
+    auto lock = scheduler_->lock();
     released_time_.advance_to(tag);
   }
 
-  inline auto try_acquire_tag(const Tag& tag) {
-    std::lock_guard lock(mutex_);
-    return tag <= released_time_;
-  }
+  // The caller must hold a lock on the scheduler mutex
+  inline auto try_acquire_tag(const Tag& tag) { return tag <= released_time_; }
 
-  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
+  inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock,
                           const std::function<bool(void)>& abort_waiting) -> bool {
     if (try_acquire_tag(tag)) {
       return true;
     }
-    cv.wait(lock, [this, &tag, &abort_waiting]() { return try_acquire_tag(tag) || abort_waiting(); });
+    scheduler_->wait(lock, [this, &tag, &abort_waiting]() { return try_acquire_tag(tag) || abort_waiting(); });
     return !abort_waiting();
   }
 };

lib/action.cc

@@ -18,7 +18,7 @@ namespace reactor {
 void BaseAction::register_trigger(Reaction* reaction) {
   reactor_assert(reaction != nullptr);
   reactor_assert(this->environment() == reaction->environment());
-  assert_phase(this, Environment::Phase::Assembly);
+  assert_phase(this, Phase::Assembly);
   validate(this->container() == reaction->container(),
            "Action triggers must belong to the same reactor as the triggered "
            "reaction");
@@ -29,7 +29,7 @@ void BaseAction::register_trigger(Reaction* reaction) {
 void BaseAction::register_scheduler(Reaction* reaction) {
   reactor_assert(reaction != nullptr);
   reactor_assert(this->environment() == reaction->environment());
-  assert_phase(this, Environment::Phase::Assembly);
+  assert_phase(this, Phase::Assembly);
   // the reaction must belong to the same reactor as this action
   validate(this->container() == reaction->container(), "Scheduable actions must belong to the same reactor as the "
                                                        "triggered reaction");

lib/assert.cc

@@ -7,6 +7,7 @@
  */
 
 #include "reactor-cpp/assert.hh"
+#include "reactor-cpp/environment.hh"
 
 namespace reactor {
 
@@ -16,4 +17,30 @@ auto ValidationError::build_message(const std::string_view msg) noexcept -> std:
   return string_stream.str();
 }
 
+void assert_phase([[maybe_unused]] const ReactorElement* ptr, [[maybe_unused]] Phase phase) {
+  if constexpr (runtime_assertion) { // NOLINT
+    if (ptr->environment()->phase() != phase) {
+      auto enum_value_to_name = [](Phase phase) -> std::string {
+        const std::map<Phase, std::string> conversation_map = {
+            // NOLINT
+            {Phase::Construction, "Construction"}, {Phase::Assembly, "Assembly"},
+            {Phase::Startup, "Startup"},           {Phase::Execution, "Execution"},
+            {Phase::Shutdown, "Shutdown"},         {Phase::Deconstruction, "Deconstruction"}};
+        // in C++20 use .contains()
+        if (conversation_map.find(phase) != std::end(conversation_map)) {
+          return conversation_map.at(phase);
+        }
+        return "Unknown Phase: Value: " + std::to_string(extract_value(phase));
+      };
+#ifdef __linux__
+      print_debug_backtrace();
+#endif
+
+      // C++20 std::format
+      throw ValidationError("Expected Phase: " + enum_value_to_name(phase) +
+                            " Current Phase: " + enum_value_to_name(ptr->environment()->phase()));
+    }
+  }
+}
+
 } // namespace reactor

lib/port.cc

@@ -20,7 +20,7 @@ void BasePort::base_bind_to(BasePort* port) {
   reactor_assert(this->environment() == port->environment());
   validate(!port->has_inward_binding(), "Ports may only be connected once");
   validate(!port->has_anti_dependencies(), "Ports with anti dependencies may not be connected to other ports");
-  assert_phase(this, Environment::Phase::Assembly);
+  assert_phase(this, Phase::Assembly);
   if (this->is_input() && port->is_input()) {
     validate(this->container() == port->container()->container(),
              "An input port A may only be bound to another input port B if B is contained by a reactor that in turn is "
@@ -52,7 +52,7 @@ void BasePort::register_dependency(Reaction* reaction, bool is_trigger) noexcept
   reactor_assert(reaction != nullptr);
   reactor_assert(this->environment() == reaction->environment());
   validate(!this->has_outward_bindings(), "Dependencies may no be declared on ports with an outward binding!");
-  assert_phase(this, Environment::Phase::Assembly);
+  assert_phase(this, Phase::Assembly);
 
   if (this->is_input()) {
     validate(this->container() == reaction->container(), "Dependent input ports must belong to the same reactor as the "
@@ -74,7 +74,7 @@ void BasePort::register_antidependency(Reaction* reaction) noexcept {
   reactor_assert(reaction != nullptr);
   reactor_assert(this->environment() == reaction->environment());
   validate(!this->has_inward_binding(), "Antidependencies may no be declared on ports with an inward binding!");
-  assert_phase(this, Environment::Phase::Assembly);
+  assert_phase(this, Phase::Assembly);
 
   if (this->is_output()) {
     validate(this->container() == reaction->container(),