init mutations

Author: tanneberger

examples/CMakeLists.txt

@@ -7,3 +7,5 @@ add_subdirectory(count)
 add_subdirectory(ports)
 add_subdirectory(hello)
 add_subdirectory(power_train)
+add_subdirectory(multiport_mutation)
+add_subdirectory(unit_tests_mutations)

examples/count/main.cc

@@ -12,8 +12,8 @@ class Count : public Reactor {
   LogicalAction<int> counter{"counter", this};
 
   // reactions_
-  Reaction r_init{"r_init", 1, this, [this]() { init(); }};
-  Reaction r_counter{"r_counter", 2, this, [this]() { print_count(); }};
+  Reaction r_init{"r_init", 1, false, this, [this]() { this->init(); }};
+  Reaction r_counter{"r_counter", 2, false, this, [this]() { this->print_count(); }};
 
 public:
   explicit Count(Environment* env)

examples/multiport_mutation/CMakeLists.txt

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

examples/multiport_mutation/consumer.hh

@@ -0,0 +1,46 @@
+//
+// Created by tanneberger on 11/17/24.
+//
+
+#ifndef CONSUMER_HH
+#define CONSUMER_HH
+
+#include <reactor-cpp/reactor-cpp.hh>
+
+using namespace reactor;
+using namespace std::chrono_literals;
+
+class Consumer : public Reactor {
+private:
+  class Inner: public Scope {
+    Inner(Reactor* reactor, std::size_t index) : Scope(reactor), index_(index) {}
+    std::size_t index_;
+
+    [[maybe_unused]] const Inner& __lf_inner = *this;
+
+    void reaction_1([[maybe_unused]] const Input<unsigned>& scale) {
+      std::cout << "consumer: " << index_ << " received value!" << std::endl;
+    }
+
+    friend Consumer;
+  };
+
+  Inner __lf_inner;
+  Reaction handle{"handle", 1, 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) {
+    std::cout << "creating instance of consumer" << std::endl;
+  }
+  ~Consumer() override {
+    std::cout << "Consumer Object is deleted" << std::endl;
+  };
+
+  Input<unsigned> in{"in", this};
+
+  void assemble() override {
+    handle.declare_trigger(&in);
+  }
+};
+
+
+#endif //CONSUMER_HH

examples/multiport_mutation/load_balancer.hh

@@ -0,0 +1,82 @@
+//
+// Created by tanneberger on 11/17/24.
+//
+
+#ifndef LOAD_BALANCER_HH
+#define LOAD_BALANCER_HH
+
+#include <reactor-cpp/reactor-cpp.hh>
+
+#include "../../lib/mutation/multiport.cc"
+#include "reactor-cpp/mutations/multiport.hh"
+
+using namespace reactor;
+using namespace std::chrono_literals;
+
+class LoadBalancer : public Reactor {
+private:
+  class Inner: public MutableScope {
+    Inner(Reactor* reactor) : MutableScope(reactor) {}
+    [[maybe_unused]] const Inner& __lf_inner = *this;
+
+    // reaction bodies
+    void reaction_1(const Input<unsigned>& inbound, LogicalAction<unsigned>& scale_action, Multiport<Output<unsigned>>& outbound) {
+      if (rand() % 30 == 0) {
+        scale_action.schedule(rand() % 20 + 1);
+      }
+
+      std::cout << "multiport size: " << outbound.size() << std::endl;
+      outbound[rand() % outbound.size()].set(inbound.get());
+    }
+
+    void reaction_2(ModifableMultiport<Output<unsigned>>&outbound, [[maybe_unused]] const LogicalAction<unsigned>& scale, Output<unsigned>& scale_bank) {
+      ModifableMultiport<Output<unsigned>>* temp = &outbound;
+      std::size_t new_size = *scale.get();
+
+      auto antideps = (outbound[0]).anti_dependencies();
+
+      MutationChangeOutputMultiportSize change_size{temp, this->reactor_, antideps, new_size};
+
+      add_to_transaction(&change_size);
+
+      commit_transaction();
+
+      scale_bank.set(new_size);
+    }
+
+    friend LoadBalancer;
+  };
+
+  Inner __lf_inner;
+  Reaction process{"process", 2, this, [this]() { __lf_inner.reaction_1(this->inbound, this->scale_action, this->out); }};
+  Reaction scale{"scale", 1, this, [this]() { __lf_inner.reaction_2(this->out, this->scale_action, this->scale_bank); }};
+
+public:
+  LoadBalancer(const std::string& name, Environment* env)
+      : Reactor(name, env), __lf_inner(this) {
+    std::cout << "creating instance of load balancer" << std::endl;
+    out.reserve(4);
+    for (size_t _lf_idx = 0; _lf_idx < 4; _lf_idx++) {
+      std::string _lf_port_name = out.name() +  "_" + std::to_string(_lf_idx);
+      out.emplace_back(_lf_port_name, this);
+    }
+  }
+
+  LogicalAction<unsigned> scale_action{"scale", this, 1us};
+  ModifableMultiport<Output<unsigned>> out{"out"};
+  Input<unsigned> inbound{"inbound", this}; // NOLINT
+  Output<unsigned> scale_bank{"scale_bank", this};
+
+  void assemble() override {
+    std::cout << "assemble LoadBalancer\n";
+    for (auto& __lf_port : out) {
+      process.declare_antidependency(&__lf_port);
+    }
+    process.declare_trigger(&inbound);
+    scale.declare_trigger(&scale_action);
+  }
+};
+
+
+
+#endif //LOAD_BALANCER_HH

examples/multiport_mutation/main.cc

@@ -0,0 +1,100 @@
+#include <iostream>
+
+#include <reactor-cpp/mutations/bank.hh>
+#include <reactor-cpp/mutations/connection.hh>
+
+#include "./consumer.hh"
+#include "./load_balancer.hh"
+#include "./producer.hh"
+#include "../../lib/mutation/bank.cc"
+#include "../../lib/mutation/connection.cc"
+#include <reactor-cpp/reactor-cpp.hh>
+
+class Deployment : public Reactor {
+  std::unique_ptr<Producer> producer_;
+  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);}};
+
+public:
+
+class Inner: public MutableScope {
+    int state = 0;
+    [[maybe_unused]] const Inner& __lf_inner = *this;
+public:
+
+    Inner(Reactor* reactor) : MutableScope(reactor) {}
+    void reaction_1(const Input<unsigned>& scale, std::vector<std::unique_ptr<Consumer>>& reactor_bank, ModifableMultiport<Output<unsigned>>& load_balancer) {
+      std::size_t new_size = *scale.get();
+      std::size_t old_size = reactor_bank.size();
+
+      std::function<std::unique_ptr<Consumer>(Reactor*, std::size_t)> lambda = [](Reactor* reactor, std::size_t index) {
+        std::string __lf_inst_name = "consumer_" + std::to_string(index);
+        return std::make_unique<Consumer>(__lf_inst_name, reactor->environment(), index);
+      };
+      MutationChangeBankSize change_size{&reactor_bank, this->reactor_, new_size, lambda};
+
+      add_to_transaction(&change_size);
+
+      // old topology
+      commit_transaction();
+      // new topology
+
+      if (old_size > new_size) {
+
+        for (auto i = 0; i < old_size - new_size; i++) {
+        }
+      } else {
+        std::cout << "load_balancer size:" << load_balancer.size() << " bank size: " << reactor_bank.size() << std::endl;
+        for (auto i = 0; i < new_size - old_size; i++) {
+            std::cout << "add connection: " << i + old_size << std::endl;
+            MutationAddConnection<Output<unsigned>, Input<unsigned>> add_conn{&load_balancer[i + old_size], &reactor_bank[i + old_size].get()->in, reactor_};
+            add_to_transaction(&add_conn);
+            commit_transaction();
+        }
+      }
+
+      std::cout << "new bank size:" << reactor_bank.size() << std::endl;
+    }
+
+    friend LoadBalancer;
+  };
+
+  Inner __inner;
+
+  Deployment(const std::string& name, Environment* env) : Reactor(name, env), __inner(this),
+  producer_(std::make_unique<Producer>("producer", environment())),
+  load_balancer_(std::make_unique<LoadBalancer>("load_balancer", environment())) {
+    std::cout << "creating instance of deployment" << std::endl;
+    consumers_.reserve(4);
+    for (size_t __lf_idx = 0; __lf_idx < 4; __lf_idx++) {
+      std::string __lf_inst_name = "consumer_" + std::to_string(__lf_idx);
+      consumers_.push_back(std::make_unique<Consumer>(__lf_inst_name, environment(), __lf_idx));
+    }
+  }
+  ~Deployment() override = default;
+
+  Input<unsigned> scale{"scale", this};
+
+  void assemble() override {
+    for (size_t __lf_idx = 0; __lf_idx < 4; __lf_idx++) {
+      environment()->draw_connection(load_balancer_->out[__lf_idx], consumers_[__lf_idx]->in, ConnectionProperties{});
+      environment()->draw_connection(producer_->value, load_balancer_->inbound, ConnectionProperties{});
+    }
+    environment()->draw_connection(load_balancer_->scale_bank, scale, ConnectionProperties{});
+    scale_bank.declare_trigger(&this->scale);
+  }
+};
+
+
+auto main() -> int {
+  //srand(time(nullptr));
+  Environment env{4};
+  auto deployment = std::make_unique<Deployment>("c1", &env);
+  env.optimize();
+  env.assemble();
+  auto thread = env.startup();
+  thread.join();
+  return 0;
+}

examples/multiport_mutation/producer.hh

@@ -0,0 +1,45 @@
+//
+// Created by tanneberger on 11/17/24.
+//
+
+#ifndef PRODUCER_HH
+#define PRODUCER_HH
+
+#include <reactor-cpp/reactor-cpp.hh>
+
+using namespace reactor;
+using namespace std::chrono_literals;
+
+class Producer : public Reactor {
+private:
+  Timer timer{"timer", this, 1s, 1s};
+  Reaction r_timer{"r_timer", 1, this, [this]() { __lf_inner.reaction_1(this->value);}};
+
+  class Inner: public Scope {
+    [[maybe_unused]] const Inner& __lf_inner = *this;
+    void reaction_1([[maybe_unused]] Output<unsigned>& out) {
+      std::cout << "producing value" << std::endl;
+      out.set(42);
+    }
+    explicit Inner(Reactor* reactor) : Scope(reactor) {}
+
+    friend Producer;
+  };
+
+  Inner __lf_inner;
+public:
+  Producer(const std::string& name, Environment* env) : Reactor(name, env), __lf_inner(this) {
+    std::cout << "creating instance of producer" << std::endl;
+  }
+  Producer() = delete;
+  ~Producer() override = default;
+
+  Output<unsigned> value{"value", this};
+
+  void assemble() override {
+    r_timer.declare_trigger(&timer);
+    r_timer.declare_antidependency(&value);
+  }
+};
+
+#endif //PRODUCER_HH

examples/ports/main.cc

@@ -51,7 +51,7 @@ class Counter : public Reactor {
 
 class Printer : public Reactor {
 private:
-  Reaction r_value{"r_value", 1, this, [this]() { on_value(); }};
+  Reaction r_value{"r_value", 1, this, [this]() { this->on_value(); }};
 
 public:
   Input<int> value{"value", this};     // NOLINT
@@ -69,7 +69,7 @@ class Printer : public Reactor {
 
 class Adder : public Reactor {
 private:
-  Reaction r_add{"r_add", 1, this, [this]() { add(); }};
+  Reaction r_add{"r_add", 1, this, [this]() { this->add(); }};
 
 public:
   Input<int> i1{"i1", this};    // NOLINT
@@ -85,7 +85,7 @@ class Adder : public Reactor {
     r_add.declare_antidependency(&sum);
   }
 
-  void add(Adder* self) {
+  void add() {
     if (i1.is_present() && i2.is_present()) {
       sum.set(*i1.get() + *i2.get());
       std::cout << "setting sum\n";

examples/unit_tests_mutations/CMakeLists.txt

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