changing scheduler so mutations are executed sequentially

Author: tanneberger

CMakeLists.txt

@@ -9,8 +9,8 @@ endif()
 
 project(${LIB_TARGET} LANGUAGES CXX VERSION 0.0.1)
 
-# require C++17
-set(CMAKE_CXX_STANDARD 17)
+# require C++20
+set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
 set(CMAKE_CXX_EXTENSIONS OFF)
 

examples/CMakeLists.txt

@@ -1,10 +1,4 @@
-include_directories(
-  "${PROJECT_SOURCE_DIR}/include"
-  )
+include_directories("${PROJECT_SOURCE_DIR}/include")
 
-add_custom_target(examples)
-add_subdirectory(count)
-add_subdirectory(ports)
-add_subdirectory(hello)
-add_subdirectory(power_train)
-add_subdirectory(multiport_mutation)
+    add_custom_target(examples) add_subdirectory(count) add_subdirectory(ports) add_subdirectory(hello)
+        add_subdirectory(power_train) add_subdirectory(multiport_mutation)

examples/count/CMakeLists.txt

@@ -1,3 +1,2 @@
-add_executable(count EXCLUDE_FROM_ALL main.cc)
-target_link_libraries(count reactor-cpp)
-add_dependencies(examples count)
+add_executable(count EXCLUDE_FROM_ALL main.cc) target_link_libraries(count reactor - cpp)
+    add_dependencies(examples count)

examples/hello/CMakeLists.txt

@@ -1,3 +1,2 @@
-add_executable(hello EXCLUDE_FROM_ALL main.cc)
-target_link_libraries(hello reactor-cpp)
-add_dependencies(examples hello)
+add_executable(hello EXCLUDE_FROM_ALL main.cc) target_link_libraries(hello reactor - cpp)
+    add_dependencies(examples hello)

examples/multiport_mutation/CMakeLists.txt

@@ -1,3 +1,2 @@
-add_executable(mutation_multiports main.cc)
-target_link_libraries(mutation_multiports reactor-cpp)
-add_dependencies(examples mutation_multiports)
+add_executable(mutation_multiports main.cc) target_link_libraries(mutation_multiports reactor - cpp)
+    add_dependencies(examples mutation_multiports)

examples/multiport_mutation/consumer.hh

@@ -17,8 +17,8 @@ using namespace std::chrono_literals;
 
 class Consumer final : public Reactor { // NOLINT
   class Inner : public Scope {
-    Inner(Reactor* reactor, std::size_t index)
-        : Scope(reactor)
+    Inner(Reaction* reaction, std::size_t index)
+        : Scope(reaction)
         , index_(index) {}
     std::size_t index_ = 0;
 
@@ -30,12 +30,12 @@ class Consumer final : public Reactor { // NOLINT
   };
 
   Inner _lf_inner;
-  Reaction handle{"handle", 1, this, [this]() { _lf_inner.reaction_1(this->in); }};
+  Reaction handle{"handle", 1, false, this, [this]() { _lf_inner.reaction_1(this->in); }};
 
 public:
   Consumer(const std::string& name, Environment* env, std::size_t index)
       : Reactor(name, env)
-      , _lf_inner(this, index) {
+      , _lf_inner(&handle, index) {
     std::cout << "creating instance of consumer" << '\n';
   }
   ~Consumer() override { std::cout << "Consumer Object is deleted" << '\n'; };

examples/multiport_mutation/load_balancer.hh

@@ -17,29 +17,34 @@ using namespace std::chrono_literals;
 
 class LoadBalancer final : public Reactor { // NOLINT
   class Inner : public MutableScope {
-    explicit Inner(Reactor* reactor)
-        : MutableScope(reactor) {}
+    explicit Inner(Reaction* reaction)
+        : MutableScope(reaction) {}
 
     // reaction bodies
     static void reaction_1(const Input<unsigned>& inbound, Output<unsigned>& scale_bank,
                            Multiport<Output<unsigned>>& outbound) {
-      if (std::rand() % 15 == 0) {            // NOLINT
+      if (std::rand() % 15 == 0) { // NOLINT
+
+        std::cout << "triggering mutation" << std::endl;
         scale_bank.set(std::rand() % 20 + 1); // NOLINT
       }
       const unsigned outbound_port = std::rand() % outbound.size(); // NOLINT
+
+      std::cout << "forwarding to: " << outbound_port << std::endl;
       outbound[outbound_port].set(inbound.get());
     }
 
     friend LoadBalancer;
   };
 
   Inner _lf_inner;
-  Reaction process{"process", 1, this, [this]() { Inner::reaction_1(this->inbound, this->scale_bank, this->out); }};
+  Reaction process{"process", 1, false, this,
+                   [this]() { Inner::reaction_1(this->inbound, this->scale_bank, this->out); }};
 
 public:
   LoadBalancer(const std::string& name, Environment* env)
       : Reactor(name, env)
-      , _lf_inner(this) {
+      , _lf_inner(&process) {
     out.reserve(4);
     for (size_t _lf_idx = 0; _lf_idx < 4; _lf_idx++) {
       out.create_new_port();

examples/multiport_mutation/main.cc

@@ -22,17 +22,21 @@ class Deployment final : public Reactor { // NOLINT
   std::unique_ptr<LoadBalancer> load_balancer_;
   std::vector<std::unique_ptr<Consumer>> consumers_;
 
-  Reaction scale_bank{"scale_bank", 1, this,
-                      [this]() { this->_inner.reaction_1(this->scale, this->consumers_, load_balancer_->out); }};
+  Reaction scale_bank{"scale_bank", 1, true, this, [this]() {
+                        this->_inner.reaction_1(this->scale, this->consumers_, load_balancer_->out);
+                        //
+                      }};
 
+  class InnerNonMutable : public Scope {};
   class Inner : public MutableScope {
     int state = 0;
 
   public:
-    explicit Inner(Reactor* reactor)
-        : MutableScope(reactor) {}
+    explicit Inner(Reaction* reaction)
+        : MutableScope(reaction) {}
     void reaction_1(const Input<unsigned>& scale, std::vector<std::unique_ptr<Consumer>>& reactor_bank,
                     ModifableMultiport<Output<unsigned>>& load_balancer) {
+      std::cout << "mutation reaction is being executed" << std::endl;
       std::size_t new_size = *scale.get();
 
       std::function lambda = [](Environment* env, std::size_t index) {
@@ -41,11 +45,11 @@ class Deployment final : public Reactor { // NOLINT
       };
 
       std::function get_input_port = [](const std::unique_ptr<Consumer>& consumer) { return &consumer->in; };
+
       const auto rescale = std::make_shared<ResizeMultiportToBank<unsigned, Consumer>>(
-          &load_balancer, &reactor_bank, get_input_port, lambda, new_size);
+          reaction_, &load_balancer, &reactor_bank, get_input_port, lambda, new_size);
 
       add_to_transaction(rescale);
-
       commit_transaction(true);
     }
 
@@ -57,7 +61,7 @@ class Deployment final : public Reactor { // NOLINT
 public:
   Deployment(const std::string& name, Environment* env)
       : Reactor(name, env)
-      , _inner(this)
+      , _inner(&scale_bank)
       , producer_(std::make_unique<Producer>("producer", environment()))
       , load_balancer_(std::make_unique<LoadBalancer>("load_balancer", environment())) {
     std::cout << "creating instance of deployment" << '\n';

examples/multiport_mutation/multiport_to_bank.hh

@@ -33,54 +33,57 @@ template <class PortType, class ReactorType> class ResizeMultiportToBank : publi
   std::size_t new_size_ = 0;
 
 public:
-  ResizeMultiportToBank(ModifableMultiport<Output<PortType>>* multiport,
+  ResizeMultiportToBank(Reaction* reaction, ModifableMultiport<Output<PortType>>* multiport,
                         std::vector<std::unique_ptr<ReactorType>>* bank,
                         std::function<Input<PortType>*(const std::unique_ptr<ReactorType>&)> get_input_port,
                         std::function<std::unique_ptr<ReactorType>(Environment* env, std::size_t index)> create_lambda,
                         std::size_t new_size)
-      : multiport_(multiport)
+      : Mutation(reaction)
+      , multiport_(multiport)
       , bank_(bank)
       , get_input_port_(get_input_port)
       , create_lambda_(create_lambda)
       , new_size_(new_size) {}
 
   ~ResizeMultiportToBank() = default;
-  auto run() -> MutationResult {
+  auto run() -> MutationResult override {
     if (multiport_->size() != bank_->size()) {
       return NotMatchingBankSize;
     }
     auto old_size = multiport_->size();
 
     if (new_size_ > old_size) {
-      auto change_multiport_size = std::make_shared<MutationChangeOutputMultiportSize<unsigned>>(multiport_, new_size_);
+      auto change_multiport_size =
+          std::make_shared<MutationChangeOutputMultiportSize<unsigned>>(reaction_, multiport_, new_size_);
 
       change_multiport_size->run();
 
       auto change_bank_size = std::make_shared<MutationChangeBankSize<std::unique_ptr<ReactorType>>>(
-          bank_, (*bank_)[0]->environment(), new_size_, create_lambda_);
+          reaction_, bank_, new_size_, create_lambda_);
 
       change_bank_size->run();
 
       for (auto i = old_size; i < new_size_; i++) {
         auto add_conn = std::make_shared<MutationAddConnection<Output<PortType>, Input<PortType>>>(
-            &(*multiport_)[i], get_input_port_((*bank_)[i]), (*bank_)[0]->environment(), true);
+            reaction_, &(*multiport_)[i], get_input_port_((*bank_)[i]), true);
 
         add_conn->run();
       }
     } else if (new_size_ < old_size) {
       for (auto i = old_size - 1; i >= new_size_; i--) {
         auto add_conn = std::make_shared<MutationAddConnection<Output<PortType>, Input<PortType>>>(
-            &(*multiport_)[i], get_input_port_((*bank_)[i]), (*bank_)[0]->environment(), false);
+            reaction_, &(*multiport_)[i], get_input_port_((*bank_)[i]), false);
 
         add_conn->run();
       }
 
-      auto change_multiport_size = std::make_shared<MutationChangeOutputMultiportSize<unsigned>>(multiport_, new_size_);
+      auto change_multiport_size =
+          std::make_shared<MutationChangeOutputMultiportSize<unsigned>>(reaction_, multiport_, new_size_);
 
       change_multiport_size->run();
 
       auto change_bank_size = std::make_shared<MutationChangeBankSize<std::unique_ptr<ReactorType>>>(
-          bank_, (*bank_)[0]->environment(), new_size_, create_lambda_);
+          reaction_, bank_, new_size_, create_lambda_);
 
       change_bank_size->run();
     }

examples/multiport_mutation/producer.hh

@@ -17,18 +17,15 @@ using namespace std::chrono_literals;
 class Producer final : public Reactor { // NOLINT
 private:
   Timer timer{"timer", this, 1s, 1s};
-  Reaction r_timer{"r_timer", 1, this, [this]() { _lf_inner.reaction_1(this->value); }};
+  Reaction r_timer{"r_timer", 1, false, this, [this]() { _lf_inner.reaction_1(this->value); }};
 
   class Inner : public Scope {
     unsigned int counter_ = 0;
 
-    void reaction_1([[maybe_unused]] Output<unsigned>& out) {
-      // std::cout << "producing value:" << counter_ << "\n";
-      out.set(counter_++);
-    }
+    void reaction_1([[maybe_unused]] Output<unsigned>& out) { out.set(counter_++); }
 
-    explicit Inner(Reactor* reactor)
-        : Scope(reactor) {}
+    explicit Inner(Reaction* reaction)
+        : Scope(reaction) {}
 
     friend Producer;
   };
@@ -38,7 +35,7 @@ private:
 public:
   Producer(const std::string& name, Environment* env)
       : Reactor(name, env)
-      , _lf_inner(this) {
+      , _lf_inner(&r_timer) {
     std::cout << "creating instance of producer\n";
   }
   Producer() = delete;