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/connection.hh

@@ -20,6 +20,7 @@
 #include "reactor.hh"
 #include "time.hh"
 #include "time_barrier.hh"
+#include <unistd.h>
 
 namespace reactor {
 
@@ -138,6 +139,7 @@ public:
 
   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 {
+    reactor_assert(lock.owns_lock());
     log_.debug() << "downstream tries to acquire tag " << tag;
 
     if (this->upstream_port() == nullptr) {
@@ -164,7 +166,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 +203,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 +232,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->environmet()->scheduler(), abort_waiting);
   }
 
   void bind_upstream_port(Port<T>* port) override { Connection<T>::bind_upstream_port(port); }

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,38 @@ 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_;
 
 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/scheduler.cc

@@ -368,7 +368,7 @@ void Scheduler::next() { // NOLINT
         // synchronize with physical time if not in fast forward mode
         if (!environment_->fast_fwd_execution()) {
           bool result = PhysicalTimeBarrier::acquire_tag(
-              t_next, lock, cv_schedule_, [&t_next, this]() { return t_next != event_queue_.next_tag(); });
+              t_next, lock, this, [&t_next, this]() { return t_next != event_queue_.next_tag(); });
           // If acquire tag returns false, then a new event was inserted into the queue and we need to start over
           if (!result) {
             continue;
@@ -378,8 +378,8 @@ void Scheduler::next() { // NOLINT
         // Wait until all input actions mark the tag as safe to process.
         bool result{true};
         for (auto* action : environment_->input_actions_) {
-          bool inner_result = action->acquire_tag(t_next, lock, cv_schedule_,
-                                                  [&t_next, this]() { return t_next != event_queue_.next_tag(); });
+          bool inner_result =
+              action->acquire_tag(t_next, lock, [&t_next, this]() { return t_next != event_queue_.next_tag(); });
           // If the wait was aborted or if the next tag changed in the meantime,
           // we need to break from the loop and continue with the main loop.
           if (!inner_result || t_next != event_queue_.next_tag()) {