first step at introducing a scheduling policy

This makes the worker implementation plolicy based

Author: cmnrd

include/reactor-cpp/impl/scheduler_impl.hh

@@ -0,0 +1,58 @@
+/*
+ * Copyright (C) 2022 TU Dresden
+ * All rights reserved.
+ *
+ * Authors:
+ *   Christian Menard
+ */
+
+#ifndef REACTOR_CPP_IMPL_SCHEDULER_IMPL_HH
+#define REACTOR_CPP_IMPL_SCHEDULER_IMPL_HH
+
+#include "reactor-cpp/trace.hh"
+#include <reactor-cpp/logging.hh>
+#include <reactor-cpp/reaction.hh>
+
+namespace reactor {
+
+template <class SchedulingPolicy>
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+thread_local const Worker<SchedulingPolicy>* Worker<SchedulingPolicy>::current_worker = nullptr;
+
+template <class SchedulingPolicy>
+Worker<SchedulingPolicy>::Worker(Worker<SchedulingPolicy>&& worker) // NOLINT(performance-noexcept-move-constructor)
+    : policy_{worker.policy_}
+    , identity_{worker.identity_} {
+  // Need to provide the move constructor in order to organize workers in a
+  // std::vector. However, moving is not save if the thread is already running,
+  // thus we throw an exception here if the worker is moved but the
+  // internal thread is already running.
+
+  if (worker.thread_.joinable()) {
+    throw std::runtime_error{"Running workers cannot be moved!"};
+  }
+}
+
+template <class SchedulingPolicy> void Worker<SchedulingPolicy>::work() const {
+  // initialize the current worker thread local variable
+  current_worker = this;
+
+  log::Debug() << "(Worker " << identity_ << ") Starting";
+
+  policy_.worker_function(*this);
+
+  log::Debug() << "(Worker " << identity_ << ") terminates";
+}
+
+template <class SchedulingPolicy> void Worker<SchedulingPolicy>::execute_reaction(Reaction* reaction) const {
+  log::Debug() << "(Worker " << identity_ << ") "
+               << "execute reaction " << reaction->fqn();
+
+  tracepoint(reactor_cpp, reaction_execution_starts, id, reaction->fqn(), scheduler.logical_time());
+  reaction->trigger();
+  tracepoint(reactor_cpp, reaction_execution_finishes, id, reaction->fqn(), scheduler.logical_time());
+}
+
+} // namespace reactor
+
+#endif // REACTOR_CPP_IMPL_SCHEDULER_IMPL_HH

include/reactor-cpp/scheduler.hh

@@ -10,6 +10,7 @@
 #define REACTOR_CPP_SCHEDULER_HH
 
 #include <condition_variable>
+#include <cstddef>
 #include <functional>
 #include <future>
 #include <map>
@@ -26,12 +27,11 @@ namespace reactor {
 
 // forward declarations
 class Scheduler;
-class Worker;
 
-class Worker { // NOLINT
-public:
-  Scheduler& scheduler_;
-  const unsigned int identity_{0};
+template <class SchedulingPolicy> class Worker {
+private:
+  SchedulingPolicy& policy_;
+  const std::size_t identity_{0};
   std::thread thread_{};
 
   // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
@@ -40,20 +40,44 @@ public:
   void work() const;
   void execute_reaction(Reaction* reaction) const;
 
-  Worker(Scheduler& scheduler, unsigned int identity)
-      : scheduler_{scheduler}
+public:
+  Worker(SchedulingPolicy& policy, std::size_t identity)
+      : policy_{policy}
       , identity_{identity} {}
   Worker(Worker&& worker); // NOLINT(performance-noexcept-move-constructor)
   Worker(const Worker& worker) = delete;
+  ~Worker() = default;
+
+  auto operator=(const Worker& worker) -> Worker& = delete;
+  auto operator=(Worker&& worker) -> Worker& = delete;
 
-  void start_thread() { thread_ = std::thread(&Worker::work, this); }
-  void join_thread() { thread_.join(); }
+  void start() { thread_ = std::thread(&Worker::work, this); }
+  void join() { thread_.join(); }
+
+  [[nodiscard]] auto id() const -> std::size_t { return identity_; }
 
   static auto current_worker_id() -> unsigned { return current_worker->identity_; }
+
+  friend SchedulingPolicy;
+};
+
+class DefaultSchedulingPolicy {
+  Scheduler& scheduler_;
+  std::size_t identity_counter{0};
+
+public:
+  DefaultSchedulingPolicy(Scheduler& scheduler)
+      : scheduler_(scheduler) {}
+
+  void worker_function(const Worker<DefaultSchedulingPolicy>& worker) const;
+
+  auto create_worker() -> Worker<DefaultSchedulingPolicy> { return {*this, identity_counter++}; }
 };
 
 class ReadyQueue {
 private:
+  using Worker = Worker<DefaultSchedulingPolicy>;
+
   std::vector<Reaction*> queue_{};
   std::atomic<std::ptrdiff_t> size_{0};
   Semaphore sem_{0};
@@ -89,6 +113,10 @@ using EventMap = std::map<BaseAction*, std::function<void(void)>>;
 
 class Scheduler { // NOLINT
 private:
+  using Worker = Worker<DefaultSchedulingPolicy>;
+
+  DefaultSchedulingPolicy policy_;
+
   const bool using_workers_;
   LogicalTime logical_time_{};
 
@@ -137,9 +165,12 @@ public:
   void start();
   void stop();
 
-  friend Worker;
+  // FIXME: this needs to be removed in the final version, as we cannot make all policies friends...
+  friend DefaultSchedulingPolicy;
 };
 
 } // namespace reactor
 
+#include "impl/scheduler_impl.hh"
+
 #endif // REACTOR_CPP_SCHEDULER_HH

lib/scheduler.cc

@@ -20,29 +20,8 @@
 
 namespace reactor {
 
-// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
-thread_local const Worker* Worker::current_worker = nullptr;
-
-Worker::Worker(Worker&& work) // NOLINT(performance-noexcept-move-constructor)
-    : scheduler_{work.scheduler_}
-    , identity_{work.identity_} {
-  // Need to provide the move constructor in order to organize workers in a
-  // std::vector. However, moving is not save if the thread is already running,
-  // thus we throw an exception here if the worker is moved but the
-  // internal thread is already running.
-
-  if (work.thread_.joinable()) {
-    throw std::runtime_error{"Running workers cannot be moved!"};
-  }
-}
-
-void Worker::work() const {
-  // initialize the current worker thread local variable
-  current_worker = this;
-
-  log::Debug() << "(Worker " << this->identity_ << ") Starting";
-
-  if (identity_ == 0) {
+void DefaultSchedulingPolicy::worker_function(const Worker<DefaultSchedulingPolicy>& worker) const {
+  if (worker.id() == 0) {
     log::Debug() << "(Worker 0) do the initial scheduling";
     scheduler_.schedule();
   }
@@ -57,7 +36,7 @@ void Worker::work() const {
     }
 
     // execute the reaction
-    execute_reaction(reaction);
+    worker.execute_reaction(reaction);
 
     // was this the very last reaction?
     if (scheduler_.reactions_to_process_.fetch_sub(1, std::memory_order_acq_rel) == 1) {
@@ -67,17 +46,6 @@ void Worker::work() const {
     }
     // continue otherwise
   }
-
-  log::Debug() << "(Worker " << identity_ << ") terminates";
-}
-
-void Worker::execute_reaction(Reaction* reaction) const {
-  log::Debug() << "(Worker " << identity_ << ") "
-               << "execute reaction " << reaction->fqn();
-
-  tracepoint(reactor_cpp, reaction_execution_starts, id, reaction->fqn(), scheduler.logical_time());
-  reaction->trigger();
-  tracepoint(reactor_cpp, reaction_execution_finishes, id, reaction->fqn(), scheduler.logical_time());
 }
 
 void Scheduler::schedule() noexcept {
@@ -212,13 +180,13 @@ void Scheduler::start() {
   // be moved.
   workers_.reserve(num_workers);
   for (unsigned i = 0; i < num_workers; i++) {
-    workers_.emplace_back(*this, i);
-    workers_.back().start_thread();
+    workers_.emplace_back(policy_.create_worker());
+    workers_.back().start();
   }
 
   // join all worker threads
   for (auto& worker : workers_) {
-    worker.join_thread();
+    worker.join();
   }
 }
 
@@ -336,7 +304,8 @@ void Scheduler::next() { // NOLINT
 }
 
 Scheduler::Scheduler(Environment* env)
-    : using_workers_(env->num_workers() > 1)
+    : policy_(*this)
+    , using_workers_(env->num_workers() > 1)
     , environment_(env)
     , ready_queue_(env->num_workers()) {}