Merge pull request #37 from lf-lang/affiliated

Enable multiple environments and schedulers to run at the same time

Author: cmnrd

include/reactor-cpp/environment.hh

@@ -13,6 +13,7 @@
 #include <string>
 #include <vector>
 
+#include "reactor-cpp/logging.hh"
 #include "reactor-cpp/time.hh"
 #include "reactor.hh"
 #include "scheduler.hh"
@@ -30,6 +31,9 @@ public:
 
 private:
   using Dependency = std::pair<Reaction*, Reaction*>;
+
+  const std::string name_{};
+  const log::NamedLogger log_;
   const unsigned int num_workers_{default_number_worker};
   unsigned int max_reaction_index_{default_max_reaction_index};
   const bool run_forever_{default_run_forever};
@@ -39,6 +43,13 @@ private:
   std::set<Reaction*> reactions_{};
   std::vector<Dependency> dependencies_{};
 
+  /// The environment containing this environment. nullptr if this is the top environment
+  Environment* containing_environment_{nullptr};
+  /// Set of all environments contained by this environment
+  std::set<Environment*> contained_environments_{};
+  /// Pointer to the top level environment
+  Environment* top_environment_{nullptr};
+
   Scheduler scheduler_;
   Phase phase_{Phase::Construction};
   TimePoint start_time_{};
@@ -50,14 +61,14 @@ private:
 
   std::mutex shutdown_mutex_{};
 
+  auto startup(const TimePoint& start_time) -> std::thread;
+
 public:
   explicit Environment(unsigned int num_workers, bool run_forever = default_run_forever,
-                       bool fast_fwd_execution = default_fast_fwd_execution, const Duration& timeout = Duration::max())
-      : num_workers_(num_workers)
-      , run_forever_(run_forever)
-      , fast_fwd_execution_(fast_fwd_execution)
-      , scheduler_(this)
-      , timeout_(timeout) {}
+                       bool fast_fwd_execution = default_fast_fwd_execution, const Duration& timeout = Duration::max());
+  explicit Environment(const std::string& name, Environment* containing_environment);
+
+  auto name() -> const std::string& { return name_; }
 
   void register_reactor(Reactor* reactor);
   void assemble();

include/reactor-cpp/logging.hh

@@ -15,12 +15,14 @@
 #include <iostream>
 #include <memory>
 #include <mutex>
+#include <string>
+#include <utility>
 
 namespace reactor::log {
 
 template <bool enabled> class BaseLogger {};
 
-template <> class BaseLogger<true> { // NOLINT
+template <> class BaseLogger<true> {
 private:
   using Lock = std::unique_lock<std::mutex>;
 
@@ -34,6 +36,10 @@ public:
       , lock_(mutex_) {
     std::cerr << log_prefix;
   }
+  BaseLogger(const BaseLogger&) = delete;
+  auto operator=(const BaseLogger&) -> BaseLogger& = delete;
+  BaseLogger(BaseLogger&&) = delete;
+  auto operator=(BaseLogger&&) -> BaseLogger& = delete;
 
   template <class T> auto operator<<(const T& msg) -> BaseLogger& {
     std::cerr << msg; // NOLINT
@@ -43,36 +49,61 @@ public:
   ~BaseLogger() { std::cerr << std::endl; }
 };
 
-template <> class BaseLogger<false> { // NOLINT
+template <> class BaseLogger<false> {
 public:
   explicit BaseLogger([[maybe_unused]] const std::string& /*unused*/) {}
+  BaseLogger(const BaseLogger&) = delete;
+  auto operator=(const BaseLogger&) -> BaseLogger& = delete;
+  BaseLogger(BaseLogger&&) = delete;
+  auto operator=(BaseLogger&&) -> BaseLogger& = delete;
 
-  template <class T> auto operator<<(const T& /*unused*/) const -> const BaseLogger& { return *this; }
+  template <class T> auto operator<<([[maybe_unused]] const T& /*unused*/) const -> const BaseLogger& { return *this; }
 
   ~BaseLogger() = default;
 };
 
 constexpr bool debug_enabled = 4 <= REACTOR_CPP_LOG_LEVEL;
 constexpr bool info_enabled = 3 <= REACTOR_CPP_LOG_LEVEL;
-constexpr bool warning_enabled = 2 <= REACTOR_CPP_LOG_LEVEL;
+constexpr bool warn_enabled = 2 <= REACTOR_CPP_LOG_LEVEL;
 constexpr bool error_enabled = 1 <= REACTOR_CPP_LOG_LEVEL;
 
 struct Debug : BaseLogger<debug_enabled> {
   Debug()
-      : BaseLogger<debug_enabled>("[DEBUG] "){}; // NOLINT Update C++20
+      : BaseLogger<debug_enabled>("[DEBUG] "){};
 };
 struct Info : BaseLogger<info_enabled> {
   Info()
-      : BaseLogger<info_enabled>("[INFO]  "){}; // NOLINT Update C++20
+      : BaseLogger<info_enabled>("[INFO]  "){};
 };
-struct Warn : BaseLogger<warning_enabled> {
+struct Warn : BaseLogger<warn_enabled> {
   Warn()
-      : BaseLogger<warning_enabled>("[WARN]  "){}; // NOLINT Update C++20
+      : BaseLogger<warn_enabled>("[WARN]  "){};
 };
 struct Error : BaseLogger<error_enabled> {
   Error()
-      : BaseLogger<error_enabled>("[ERROR] "){}; // NOLINT Update C++20
+      : BaseLogger<error_enabled>("[ERROR] "){};
 };
+
+class NamedLogger {
+private:
+  const std::string debug_prefix_{};
+  const std::string info_prefix_{};
+  const std::string warn_prefix_{};
+  const std::string error_prefix_{};
+
+public:
+  NamedLogger(const std::string& name)
+      : debug_prefix_("[DEBUG] (" + name + ") ")
+      , info_prefix_("[INFO]  (" + name + ") ")
+      , warn_prefix_("[WARN]  (" + name + ") ")
+      , error_prefix_("[ERROR] (" + name + ") ") {}
+
+  [[nodiscard]] auto debug() const -> BaseLogger<debug_enabled> { return BaseLogger<debug_enabled>(debug_prefix_); }
+  [[nodiscard]] auto info() const -> BaseLogger<info_enabled> { return BaseLogger<info_enabled>(info_prefix_); }
+  [[nodiscard]] auto warn() const -> BaseLogger<warn_enabled> { return BaseLogger<warn_enabled>(warn_prefix_); }
+  [[nodiscard]] auto error() const -> BaseLogger<error_enabled> { return BaseLogger<error_enabled>(error_prefix_); }
+};
+
 } // namespace reactor::log
 
 #endif // REACTOR_CPP_LOGGING_HH

include/reactor-cpp/scheduler.hh

@@ -21,6 +21,7 @@
 
 #include "fwd.hh"
 #include "logical_time.hh"
+#include "reactor-cpp/logging.hh"
 #include "reactor-cpp/time.hh"
 #include "safe_vector.hh"
 #include "semaphore.hh"
@@ -32,20 +33,23 @@ class Scheduler;
 class Worker;
 
 class Worker { // NOLINT
-public:
+private:
   Scheduler& scheduler_;
   const unsigned int identity_{0};
   std::thread thread_{};
+  log::NamedLogger log_;
 
   // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
   static thread_local const Worker* current_worker;
 
   void work() const;
   void execute_reaction(Reaction* reaction) const;
 
-  Worker(Scheduler& scheduler, unsigned int identity)
+public:
+  Worker(Scheduler& scheduler, unsigned int identity, const std::string& name)
       : scheduler_{scheduler}
-      , identity_{identity} {}
+      , identity_{identity}
+      , log_(name) {}
   Worker(Worker&& worker); // NOLINT(performance-noexcept-move-constructor)
   Worker(const Worker& worker) = delete;
 
@@ -62,10 +66,12 @@ private:
   Semaphore sem_{0};
   std::ptrdiff_t waiting_workers_{0};
   const unsigned int num_workers_;
+  log::NamedLogger& log_;
 
 public:
-  explicit ReadyQueue(unsigned num_workers)
-      : num_workers_(num_workers) {}
+  explicit ReadyQueue(log::NamedLogger& log, unsigned num_workers)
+      : num_workers_(num_workers)
+      , log_(log) {}
 
   /**
    * Retrieve a ready reaction from the queue.
@@ -95,15 +101,19 @@ class Scheduler { // NOLINT
 private:
   const bool using_workers_;
   LogicalTime logical_time_{};
+  TimePoint last_observed_physical_time_{TimePoint::min()};
 
   Environment* environment_;
   std::vector<Worker> workers_{};
+  log::NamedLogger log_;
 
   std::mutex scheduling_mutex_;
   std::condition_variable cv_schedule_;
 
   std::shared_mutex mutex_event_queue_;
   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};

lib/environment.cc

@@ -11,6 +11,8 @@
 #include <algorithm>
 #include <fstream>
 #include <map>
+#include <thread>
+#include <vector>
 
 #include "reactor-cpp/action.hh"
 #include "reactor-cpp/assert.hh"
@@ -21,6 +23,28 @@
 
 namespace reactor {
 
+Environment::Environment(unsigned int num_workers, bool run_forever, bool fast_fwd_execution, const Duration& timeout)
+    : log_("Environment")
+    , num_workers_(num_workers)
+    , run_forever_(run_forever)
+    , fast_fwd_execution_(fast_fwd_execution)
+    , top_environment_(this)
+    , scheduler_(this)
+    , timeout_(timeout) {}
+
+Environment::Environment(const std::string& name, Environment* containing_environment)
+    : name_(name)
+    , log_("Environment " + name)
+    , num_workers_(containing_environment->num_workers_)
+    , run_forever_(containing_environment->run_forever_)
+    , fast_fwd_execution_(containing_environment->fast_fwd_execution_)
+    , containing_environment_(containing_environment)
+    , top_environment_(containing_environment_->top_environment_)
+    , scheduler_(this)
+    , timeout_(containing_environment->timeout()) {
+  reactor_assert(containing_environment->contained_environments_.insert(this).second);
+}
+
 void Environment::register_reactor(Reactor* reactor) {
   reactor_assert(reactor != nullptr);
   validate(this->phase() == Phase::Construction, "Reactors may only be registered during construction phase!");
@@ -36,6 +60,7 @@ void recursive_assemble(Reactor* container) { // NOLINT
 }
 
 void Environment::assemble() {
+  log_.debug() << "Assemble";
   validate(this->phase() == Phase::Construction, "assemble() may only be called during construction phase!");
   phase_ = Phase::Assembly;
   for (auto* reactor : top_level_reactors_) {
@@ -47,6 +72,11 @@ void Environment::assemble() {
     build_dependency_graph(reactor);
   }
   calculate_indexes();
+
+  // assemble all contained environments
+  for (auto* env : contained_environments_) {
+    env->assemble();
+  }
 }
 
 void Environment::build_dependency_graph(Reactor* reactor) { // NOLINT
@@ -89,7 +119,7 @@ void Environment::build_dependency_graph(Reactor* reactor) { // NOLINT
 
 void Environment::sync_shutdown() {
   {
-    std::lock_guard<std::mutex> lock(shutdown_mutex_);
+    std::lock_guard<std::mutex> lock{shutdown_mutex_};
 
     if (phase_ >= Phase::Shutdown) {
       // sync_shutdown() was already called -> abort
@@ -101,7 +131,7 @@ void Environment::sync_shutdown() {
   }
 
   // the following will only be executed once
-  log::Debug() << "Terminating the execution";
+  log_.debug() << "Terminating the execution";
 
   for (auto* reactor : top_level_reactors_) {
     reactor->shutdown();
@@ -164,7 +194,7 @@ void Environment::export_dependency_graph(const std::string& path) {
 
   dot.close();
 
-  log::Info() << "Reaction graph was written to " << path;
+  log_.info() << "Reaction graph was written to " << path;
 }
 
 void Environment::calculate_indexes() {
@@ -177,7 +207,7 @@ void Environment::calculate_indexes() {
     graph[dependencies.first].insert(dependencies.second);
   }
 
-  log::Debug() << "Reactions sorted by index:";
+  log_.debug() << "Reactions sorted by index:";
   unsigned int index = 0;
   while (!graph.empty()) {
     // find nodes with degree zero and assign index
@@ -195,7 +225,7 @@ void Environment::calculate_indexes() {
                             "/tmp/reactor_dependency_graph.dot");
     }
 
-    log::Debug dbg;
+    auto dbg = log_.debug();
     dbg << index << ": ";
     for (auto* reaction : degree_zero) {
       dbg << reaction->fqn() << ", ";
@@ -218,26 +248,39 @@ void Environment::calculate_indexes() {
 }
 
 auto Environment::startup() -> std::thread {
-  validate(this->phase() == Phase::Assembly, "startup() may only be called during assembly phase!");
+  validate(this == top_environment_, "startup() may only be called on the top environment");
+  auto start_time = get_physical_time();
+  return startup(start_time);
+}
 
-  // build the dependency graph
-  for (auto* reactor : top_level_reactors_) {
-    build_dependency_graph(reactor);
-  }
-  calculate_indexes();
+auto Environment::startup(const TimePoint& start_time) -> std::thread {
+  validate(this->phase() == Phase::Assembly, "startup() may only be called during assembly phase!");
 
-  log::Info() << "Starting the execution";
+  log_.debug() << "Starting the execution";
   phase_ = Phase::Startup;
 
-  start_time_ = get_physical_time();
+  start_time_ = start_time;
   // start up initialize all reactors
   for (auto* reactor : top_level_reactors_) {
     reactor->startup();
   }
 
   // start processing events
   phase_ = Phase::Execution;
-  return std::thread([this]() { this->scheduler_.start(); });
+
+  return std::thread([this]() {
+    std::vector<std::thread> threads;
+    // startup all contained environments recursively
+    for (auto* env : contained_environments_) {
+      threads.emplace_back(env->startup(start_time_));
+    }
+    // start the local scheduler and wait until it returns
+    this->scheduler_.start();
+    // then join all the created threads
+    for (auto& thread : threads) {
+      thread.join();
+    }
+  });
 }
 
 void Environment::dump_trigger_to_yaml(std::ofstream& yaml, const BaseAction& trigger) {
@@ -367,7 +410,7 @@ void Environment::dump_to_yaml(const std::string& path) {
     yaml << "  - to: " << iterator.second->fqn() << std::endl;
   }
 
-  log::Info() << "Program structure was dumped to " << path;
+  log_.info() << "Program structure was dumped to " << path;
 }
 
 } // namespace reactor