factor out the event queue in its own class

Author: cmnrd

include/reactor-cpp/scheduler.hh

@@ -99,6 +99,33 @@ public:
 using ActionList = SafeVector<BaseAction*>;
 using ActionListPtr = std::unique_ptr<ActionList>;
 
+class EventQueue {
+private:
+  std::shared_mutex mutex_;
+  std::map<Tag, ActionListPtr> event_queue_;
+  /// stores the actions triggered at the current tag
+  ActionListPtr triggered_actions_{nullptr};
+
+  std::vector<ActionListPtr> action_list_pool_;
+  static constexpr std::size_t action_list_pool_increment_{10};
+
+  void fill_action_list_pool();
+
+public:
+  EventQueue() { fill_action_list_pool(); }
+
+  [[nodiscard]] auto empty() const -> bool { return event_queue_.empty(); }
+  [[nodiscard]] auto next_tag() const -> Tag;
+
+  auto insert_event_at(const Tag& tag) -> const ActionListPtr&;
+
+  // should only be called while holding the scheduler mutex
+  auto extract_next_event() -> ActionListPtr&&;
+
+  // should only be called while holding the scheduler mutex
+  void return_action_list(ActionListPtr&& action_list);
+};
+
 class Scheduler { // NOLINT
 private:
   const bool using_workers_;
@@ -111,16 +138,10 @@ private:
   std::mutex scheduling_mutex_;
   std::condition_variable cv_schedule_;
 
-  std::shared_mutex mutex_event_queue_;
-  std::map<Tag, ActionListPtr> event_queue_;
+  EventQueue event_queue_;
   /// stores the actions triggered at the current tag
   ActionListPtr triggered_actions_{nullptr};
 
-  std::vector<ActionListPtr> action_list_pool_;
-  static constexpr std::size_t action_list_pool_increment_{10};
-  void fill_action_list_pool();
-  auto insert_event_at(const Tag& tag) -> const ActionListPtr&;
-
   std::vector<std::vector<BasePort*>> set_ports_;
   std::vector<std::vector<Reaction*>> triggered_reactions_;
   std::vector<std::vector<Reaction*>> reaction_queue_;

lib/scheduler.cc

@@ -149,6 +149,49 @@ void ReadyQueue::fill_up(std::vector<Reaction*>& ready_reactions) {
   }
 }
 
+void EventQueue::fill_action_list_pool() {
+  for (std::size_t i{0}; i < action_list_pool_increment_; i++) {
+    action_list_pool_.emplace_back(std::make_unique<ActionList>());
+  }
+}
+
+auto EventQueue::next_tag() const -> Tag {
+  reactor_assert(!event_queue_.empty());
+  return event_queue_.begin()->first;
+}
+
+auto EventQueue::extract_next_event() -> ActionListPtr&& {
+  reactor_assert(!event_queue_.empty());
+  return std::move(event_queue_.extract(event_queue_.begin()).mapped());
+}
+
+auto EventQueue::insert_event_at(const Tag& tag) -> const ActionListPtr& {
+  auto shared_lock = std::shared_lock<std::shared_mutex>(mutex_);
+
+  auto event_it = event_queue_.find(tag);
+  if (event_it == event_queue_.end()) {
+    shared_lock.unlock();
+    {
+      auto unique_lock = std::unique_lock<std::shared_mutex>(mutex_);
+      if (action_list_pool_.empty()) {
+        fill_action_list_pool();
+      }
+      const auto& result = event_queue_.try_emplace(tag, std::move(action_list_pool_.back()));
+      if (result.second) {
+        action_list_pool_.pop_back();
+      }
+      return result.first->second;
+    }
+  } else {
+    return event_it->second;
+  }
+}
+
+void EventQueue::return_action_list(ActionListPtr&& action_list) {
+  reactor_assert(action_list != nullptr);
+  action_list_pool_.emplace_back(std::forward<ActionListPtr>(action_list));
+}
+
 void Scheduler::terminate_all_workers() {
   log_.debug() << "Send termination signal to all workers";
   auto num_workers = environment_->num_workers();
@@ -260,8 +303,9 @@ void Scheduler::next() { // NOLINT
 
     while (triggered_actions_ == nullptr || triggered_actions_->empty()) {
       if (triggered_actions_ != nullptr) {
-        action_list_pool_.emplace_back(std::move(triggered_actions_));
+        event_queue_.return_action_list(std::move(triggered_actions_));
       }
+      reactor_assert(triggered_actions_ == nullptr);
 
       // shutdown if there are no more events in the queue
       if (event_queue_.empty() && !stop_) {
@@ -278,25 +322,23 @@ void Scheduler::next() { // NOLINT
         continue_execution_ = false;
         log_.debug() << "Shutting down the scheduler";
         Tag t_next = Tag::from_logical_time(logical_time_).delay();
-        if (!event_queue_.empty() && t_next == event_queue_.begin()->first) {
+        if (!event_queue_.empty() && t_next == event_queue_.next_tag()) {
           log_.debug() << "Schedule the last round of reactions including all "
                           "termination reactions";
-          triggered_actions_ = std::move(event_queue_.begin()->second);
-          event_queue_.erase(event_queue_.begin());
+          triggered_actions_ = std::move(event_queue_.extract_next_event());
           log_.debug() << "advance logical time to tag " << t_next;
           logical_time_.advance_to(t_next);
         } else {
           return;
         }
       } else {
-        // find the next tag
-        auto t_next = event_queue_.begin()->first;
+        auto t_next = event_queue_.next_tag();
         log_.debug() << "try to advance logical time to tag " << t_next;
 
         // 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_.begin()->first; });
+              t_next, lock, cv_schedule_, [&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;
@@ -307,10 +349,10 @@ void Scheduler::next() { // NOLINT
         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_.begin()->first; });
+                                                  [&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_.begin()->first) {
+          if (!inner_result || t_next != event_queue_.next_tag()) {
             result = false;
             break;
           }
@@ -320,12 +362,11 @@ void Scheduler::next() { // NOLINT
           continue;
         }
 
-        // Retrieve all events with tag equal to current logical time from the
-        // queue.
+        // Retrieve all triggered actions at the next tag.
         // We do not need to lock mutex_event_queue_ here, as the lock on
         // scheduling_mutex_ already ensures that no one can write to the event
         // queue.
-        triggered_actions_ = std::move(event_queue_.extract(event_queue_.begin()).mapped());
+        triggered_actions_ = std::move(event_queue_.extract_next_event());
 
         // advance logical time
         log_.debug() << "advance logical time to tag " << t_next;
@@ -359,47 +400,17 @@ Scheduler::Scheduler(Environment* env)
     : using_workers_(env->num_workers() > 1)
     , environment_(env)
     , log_("Scheduler " + env->name())
-    , ready_queue_(log_, env->num_workers()) {
-  fill_action_list_pool();
-}
+    , ready_queue_(log_, env->num_workers()) {}
 
 Scheduler::~Scheduler() = default;
 
-void Scheduler::fill_action_list_pool() {
-  for (std::size_t i{0}; i < action_list_pool_increment_; i++) {
-    action_list_pool_.emplace_back(std::make_unique<ActionList>());
-  }
-}
-
-auto Scheduler::insert_event_at(const Tag& tag) -> const ActionListPtr& {
-  auto shared_lock = std::shared_lock<std::shared_mutex>(mutex_event_queue_);
-
-  auto event_it = event_queue_.find(tag);
-  if (event_it == event_queue_.end()) {
-    shared_lock.unlock();
-    {
-      auto unique_lock = std::unique_lock<std::shared_mutex>(mutex_event_queue_);
-      if (action_list_pool_.empty()) {
-        fill_action_list_pool();
-      }
-      const auto& result = event_queue_.try_emplace(tag, std::move(action_list_pool_.back()));
-      if (result.second) {
-        action_list_pool_.pop_back();
-      }
-      return result.first->second;
-    }
-  } else {
-    return event_it->second;
-  }
-}
-
 void Scheduler::schedule_sync(BaseAction* action, const Tag& tag) {
   log_.debug() << "Schedule action " << action->fqn() << (action->is_logical() ? " synchronously " : " asynchronously ")
                << " with tag " << tag;
   reactor_assert(logical_time_ < tag);
   tracepoint(reactor_cpp, schedule_action, action->container()->fqn(), action->name(), tag);
 
-  const auto& action_list = insert_event_at(tag);
+  const auto& action_list = event_queue_.insert_event_at(tag);
   action_list->push_back(action);
 }
 
@@ -436,7 +447,7 @@ auto Scheduler::schedule_empty_async_at(const Tag& tag) -> bool {
       // processed.
       return tag == logical_time_;
     }
-    insert_event_at(tag);
+    event_queue_.insert_event_at(tag);
   }
   notify();
   return true;