mutation fixes

OmerMajNition · tanneberger · commit e088e960441d · 2024-12-13T06:16:58.000+01:00
Signed-off-by: tanneberger &lt;github@tanneberger.me&gt;
diff --git a/examples/multiport_mutation/consumer.hh b/examples/multiport_mutation/consumer.hh
@@ -18,8 +18,8 @@ private:
 
     [[maybe_unused]] const Inner& __lf_inner = *this;
 
-    void reaction_1([[maybe_unused]] const Input<unsigned>& scale) {
-      std::cout << "consumer: " << index_ << " received value!" << std::endl;
+    void reaction_1([[maybe_unused]] const Input<unsigned>& in) {
+      std::cout << "consumer: " << index_ << " received value:" << *in.get() << std::endl;
     }
 
     friend Consumer;
diff --git a/examples/multiport_mutation/load_balancer.hh b/examples/multiport_mutation/load_balancer.hh
@@ -24,9 +24,9 @@ private:
       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());
+      unsigned sel = rand() % outbound.size();
+      std::cout << "Sending out to:" << sel << std::endl;
+      outbound[sel].set(inbound.get());
     }
 
     void reaction_2(ModifableMultiport<Output<unsigned>>&outbound, [[maybe_unused]] const LogicalAction<unsigned>& scale, Output<unsigned>& scale_bank) {
@@ -74,6 +74,7 @@ public:
     }
     process.declare_trigger(&inbound);
     scale.declare_trigger(&scale_action);
+    scale.declare_antidependency(&scale_bank);
   }
 };
 
diff --git a/examples/multiport_mutation/main.cc b/examples/multiport_mutation/main.cc
@@ -47,12 +47,12 @@ class Inner: public MutableScope {
         }
       } 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_};
+        for (auto i = 0; i < new_size; i++) {
+            std::cout << "add connection: " << i << std::endl;
+            MutationAddConnection<Output<unsigned>, Input<unsigned>> add_conn{&load_balancer[i], &reactor_bank[i].get()->in, reactor_};
             add_to_transaction(&add_conn);
-            commit_transaction();
         }
+        commit_transaction(true);
       }
 
       std::cout << "new bank size:" << reactor_bank.size() << std::endl;
@@ -90,7 +90,7 @@ class Inner: public MutableScope {
 
 auto main() -> int {
   //srand(time(nullptr));
-  Environment env{4};
+  Environment env{4, true};
   auto deployment = std::make_unique<Deployment>("c1", &env);
   env.optimize();
   env.assemble();
diff --git a/examples/multiport_mutation/producer.hh b/examples/multiport_mutation/producer.hh
@@ -16,10 +16,11 @@ private:
   Reaction r_timer{"r_timer", 1, this, [this]() { __lf_inner.reaction_1(this->value);}};
 
   class Inner: public Scope {
+    unsigned itr = 0;
     [[maybe_unused]] const Inner& __lf_inner = *this;
     void reaction_1([[maybe_unused]] Output<unsigned>& out) {
-      std::cout << "producing value" << std::endl;
-      out.set(42);
+      std::cout << "producing value:" << itr << std::endl;
+      out.set(itr++);
     }
     explicit Inner(Reactor* reactor) : Scope(reactor) {}
 
diff --git a/include/reactor-cpp/environment.hh b/include/reactor-cpp/environment.hh
@@ -84,6 +84,7 @@ public:
   //TODO: fix visebility
   void calculate_indexes();
   void build_dependency_graph(Reactor* reactor);
+  void clear_dependency_graph();
 
 
   explicit Environment(unsigned int num_workers, bool fast_fwd_execution = default_fast_fwd_execution,
diff --git a/include/reactor-cpp/graph.hh b/include/reactor-cpp/graph.hh
@@ -54,7 +54,14 @@ public:
       std::vector<std::pair<P, E>> edges{std::make_pair(properties, destination)};
       graph_[source] = edges;
     } else {
-      graph_[source].emplace_back(properties, destination);
+        auto &edges = graph_[source];
+        auto duplicate = std::find_if(edges.begin(), edges.end(),
+                                       [&](const std::pair<P, E>& edge) {
+                                        return edge.first == properties && edge.second == destination;
+                                       });
+        if (duplicate == edges.end()) {
+            graph_[source].emplace_back(properties, destination);
+        }
     }
   }
 
diff --git a/include/reactor-cpp/reaction.hh b/include/reactor-cpp/reaction.hh
@@ -27,7 +27,6 @@ private:
   std::set<BasePort*> dependencies_;
 
   const int priority_ = -1;
-  const bool mutation_;
   unsigned int index_ = -1;
 
   std::function<void(void)> body_{nullptr};
@@ -53,6 +52,7 @@ public:
   [[nodiscard]] auto port_triggers() const noexcept -> const auto& { return port_trigger_; }
 
   [[maybe_unused]] [[nodiscard]] auto antidependencies() const noexcept -> const auto& { return antidependencies_; }
+  [[maybe_unused]] void clear_antidependencies() noexcept { antidependencies_.clear(); }
 
   [[nodiscard]] auto dependencies() const noexcept -> const auto& { return dependencies_; }
 
diff --git a/include/reactor-cpp/scopes.hh b/include/reactor-cpp/scopes.hh
@@ -43,7 +43,7 @@ public:
   explicit MutableScope(Reactor* reactor) : Scope(reactor), transaction_(reactor), reactor_(reactor), env_(reactor->environment()) {}
   ~MutableScope() = default;
 
-  void commit_transaction();
+  void commit_transaction(bool recalculate = false);
   void add_to_transaction(Mutation* mutation);
 
 };
diff --git a/include/reactor-cpp/transaction.hh b/include/reactor-cpp/transaction.hh
@@ -21,7 +21,6 @@ class Environment;
 
 class Transaction {
 private:
-  Reactor* parent_ = nullptr;
   Environment* environment_ = nullptr;
   std::vector<Mutation*> mutations_{};
 
@@ -30,7 +29,7 @@ public:
   ~Transaction() = default;
 
   void push_back(Mutation* mutation);
-  auto execute() -> MutationResult;
+  auto execute(bool recalculate = false) -> MutationResult;
 };
 }
 #endif // REACTOR_CPP_TRANSACTION_HH
diff --git a/lib/environment.cc b/lib/environment.cc
@@ -144,6 +144,11 @@ void Environment::assemble() { // NOLINT(readability-function-cognitive-complexi
   }
 }
 
+void Environment::clear_dependency_graph() {
+    dependencies_.clear();
+    reactions_.clear();
+}
+
 void Environment::build_dependency_graph(Reactor* reactor) {
   // obtain dependencies from each contained reactor
   for (auto* sub_reactor : reactor->reactors()) {
@@ -157,23 +162,14 @@ void Environment::build_dependency_graph(Reactor* reactor) {
     validate(result.second, "priorities must be unique for all reactions_ of the same reactor");
   }
 
-  dependencies_.clear(); //TODO: fix
-
   // connect all reactions_ this reaction depends on
   for (auto* reaction : reactor->reactions()) {
-    std::set<BasePort*> dependencies = reaction->dependencies();
-    for (auto* dependency : dependencies) {
-        if (dependency <= (BasePort*)0x100) {
-          std::cout << "FUCK" << std::endl;
-        }
-      while (dependency->has_inward_binding()) {
-        if (dependency <= (BasePort*)0x100) {
-          std::cout << "FUCK" << std::endl;
-        }
-        dependency = dependency->inward_binding();
+    for (auto* dependency : reaction->dependencies()) {
+      auto* source = dependency;
+      while (source->has_inward_binding()) {
+        source = source->inward_binding();
       }
-      std::cout << "anti deps of: " << dependency->fqn() << "(" << dependency << ")" << std::endl;
-      for (auto* antidependency : dependency->anti_dependencies()) {
+      for (auto* antidependency : source->anti_dependencies()) {
         dependencies_.emplace_back(reaction, antidependency);
       }
     }
diff --git a/lib/mutation/bank.cc b/lib/mutation/bank.cc
@@ -21,9 +21,9 @@ void reactor::MutationChangeBankSize<T>::change_size(std::size_t new_size) {
 
     for (auto i = 0; i < current_size - new_size; i++) {
       //TODO: consider saving the ports here here
-      Reactor* last = (*bank_->end()).get();
-      reactor_->environment()->remove_top_level_reactor(last);
+      std::unique_ptr<Reactor> lastElement = std::move(bank_->back());
       bank_->pop_back();
+      reactor_->environment()->remove_top_level_reactor(lastElement.get());
     }
   } else {
     // upscale
diff --git a/lib/mutation/connection.cc b/lib/mutation/connection.cc
@@ -17,12 +17,6 @@ template <class A, class B> auto reactor::MutationAddConnection<A, B>::run() ->
   source_->add_outward_binding(sink_);
   std::cout << "from: " << source_->fqn() << "(" << source_ << ")"
                        << " --> to: " << sink_->fqn() << "(" << sink_ << ")" << std::endl;
-
-  for (const auto* reactor : reactor_->environment()->top_level_reactors()) {
-    reactor_->environment()->build_dependency_graph((Reactor*)reactor);
-  }
-
-  reactor_->environment()->calculate_indexes();
   return Success;
 }
 
diff --git a/lib/mutation/multiport.cc b/lib/mutation/multiport.cc
@@ -11,7 +11,6 @@ reactor::MutationChangeOutputMultiportSize<T>::MutationChangeOutputMultiportSize
 
 template<class T>
 void reactor::MutationChangeOutputMultiportSize<T>::change_size(std::size_t new_size) {
-  multiport_->reserve(new_size);
   auto current_size = multiport_->size();
   std::cout << "scaling from: " << current_size << " to " << new_size << std::endl;
 
@@ -21,23 +20,34 @@ void reactor::MutationChangeOutputMultiportSize<T>::change_size(std::size_t new_
     for (auto i = 0; i < current_size - new_size; i++) {
       //TODO: consider saving the ports here here
 
-      std::string port_name_ = multiport_->name() + "_" + std::to_string(current_size + i - 1);
+      std::string port_name_ = multiport_->name() + "_" + std::to_string(current_size - i - 1);
       multiport_->pop_back();
       auto base_port = Output<T>{port_name_, reactor_};
       reactor_->remove_inputs(&base_port);
     }
+
+    for (auto* anti_dep : anti_dependencies_) {
+        anti_dep->clear_antidependencies();
+        for (auto i = 0; i < new_size; i++) {
+            anti_dep->declare_antidependency(&multiport_->operator[](i));
+        }
+    }
   } else {
     // upscale
 
     for (auto i = 0; i < new_size - current_size; i++) {
       std::string port_name_ = multiport_->name() + "_" + std::to_string(current_size + i);
       multiport_->emplace_back(port_name_, reactor_);
-      BasePort* port = &(multiport_->operator[](current_size + i));
-      for (auto* anti_dep : anti_dependencies_) {
-        anti_dep->declare_antidependency(port);
-      }
     }
 
+    for (auto* anti_dep : anti_dependencies_) {
+        anti_dep->clear_antidependencies();
+        for (auto i = 0; i < new_size; i++) {
+            anti_dep->declare_antidependency(&multiport_->operator[](i));
+        }
+    }
+
+
   }
 }
 template<class T>
diff --git a/lib/reaction.cc b/lib/reaction.cc
@@ -20,7 +20,6 @@ namespace reactor {
 Reaction::Reaction(const std::string& name, int priority, Reactor* container, std::function<void(void)> body)
     : ReactorElement(name, ReactorElement::Type::Reaction, container)
     , priority_(priority)
-    , mutation_(false)
     , body_(std::move(std::move(body)))
 {
   reactor_assert(priority != 0);
diff --git a/lib/reactor.cc b/lib/reactor.cc
@@ -47,6 +47,7 @@ void Reactor::register_input(BasePort* port) {
 }
 
 void Reactor::register_output(BasePort* port) {
+  (void)port;
   reactor_assert(port != nullptr);
   reactor::validate(this->environment()->phase() == Phase::Construction || this->environment()->phase() == Phase::Mutation,
                     "Ports can only be registered during construction phase!");
diff --git a/lib/scopes.cc b/lib/scopes.cc
@@ -6,6 +6,7 @@ void reactor::MutableScope::add_to_transaction(Mutation* mutation) {
   transaction_.push_back(mutation);
 }
 
-void reactor::MutableScope::commit_transaction() {
-  transaction_.execute();
+void reactor::MutableScope::commit_transaction(bool recalculate) {
+  (void)recalculate;
+  transaction_.execute(recalculate);
 }
diff --git a/lib/transaction.cc b/lib/transaction.cc
@@ -4,16 +4,25 @@
 #include "reactor-cpp/environment.hh"
 #include "reactor-cpp/reactor.hh"
 
- reactor::Transaction::Transaction(Reactor* parent)  : parent_(parent), environment_(parent->environment()) {  }
+ reactor::Transaction::Transaction(Reactor* parent) : environment_(parent->environment()) {  }
 
 
-auto reactor::Transaction::execute() -> MutationResult {
+auto reactor::Transaction::execute(bool recalculate) -> MutationResult {
 
    this->environment_->start_mutation();
    for (auto *mutation : mutations_) {
      mutation->run();
    }
 
+   if (recalculate) {
+    this->environment_->clear_dependency_graph();
+      for (const auto* reactor : this->environment_->top_level_reactors()) {
+        this->environment_->build_dependency_graph((Reactor*)reactor);
+      }
+
+      this->environment_->calculate_indexes();
+   }
+
    this->environment_->stop_mutation();
    mutations_.clear();
     return Success;

Original file line number	Diff line number	Diff line change
`@@ -18,8 +18,8 @@ private:`
`18`	`18`
`19`	`19`	`[[maybe_unused]] const Inner& __lf_inner = *this;`
`20`	`20`
`21`		`- void reaction_1([[maybe_unused]] const Input<unsigned>& scale) {`
`22`		`- std::cout << "consumer: " << index_ << " received value!" << std::endl;`
	`21`	`+ void reaction_1([[maybe_unused]] const Input<unsigned>& in) {`
	`22`	`+ std::cout << "consumer: " << index_ << " received value:" << *in.get() << std::endl;`
`23`	`23`	`}`
`24`	`24`
`25`	`25`	`friend Consumer;`
Original file line number	Diff line number	Diff line change
`@@ -24,9 +24,9 @@ private:`
`24`	`24`	`if (rand() % 30 == 0) {`
`25`	`25`	`scale_action.schedule(rand() % 20 + 1);`
`26`	`26`	`}`
`27`		`-`
`28`		`- std::cout << "multiport size: " << outbound.size() << std::endl;`
`29`		`- outbound[rand() % outbound.size()].set(inbound.get());`
	`27`	`+ unsigned sel = rand() % outbound.size();`
	`28`	`+ std::cout << "Sending out to:" << sel << std::endl;`
	`29`	`+ outbound[sel].set(inbound.get());`
`30`	`30`	`}`
`31`	`31`
`32`	`32`	`void reaction_2(ModifableMultiport<Output<unsigned>>&outbound, [[maybe_unused]] const LogicalAction<unsigned>& scale, Output<unsigned>& scale_bank) {`
`@@ -74,6 +74,7 @@ public:`
`74`	`74`	`}`
`75`	`75`	`process.declare_trigger(&inbound);`
`76`	`76`	`scale.declare_trigger(&scale_action);`
	`77`	`+ scale.declare_antidependency(&scale_bank);`
`77`	`78`	`}`
`78`	`79`	`};`
`79`	`80`