add the possibility to increase the parallelism degree of Tasks

[simgrid.git] / src / s4u / s4u_Task.cpp

diff --git a/src/s4u/s4u_Task.cpp b/src/s4u/s4u_Task.cpp
index 9c9fbe9..d899e1a 100644 (file)
--- a/src/s4u/s4u_Task.cpp
+++ b/src/s4u/s4u_Task.cpp
@@ -1,9 +1,9 @@
 #include <memory>
 #include <simgrid/Exception.hpp>
-#include <simgrid/s4u/Task.hpp>
 #include <simgrid/s4u/Comm.hpp>
 #include <simgrid/s4u/Exec.hpp>
 #include <simgrid/s4u/Io.hpp>
+#include <simgrid/s4u/Task.hpp>
 #include <simgrid/simix.hpp>
 
 #include "src/simgrid/module.hpp"
@@ -33,7 +33,7 @@ Task::Task(const std::string& name) : name_(name) {}
  */
 bool Task::ready_to_run() const
 {
-  return not working_ && queued_execs_ > 0;
+  return running_instances_ < parallelism_degree_ && queued_firings_ > 0;
 }
 
 /**
@@ -45,11 +45,12 @@ bool Task::ready_to_run() const
 void Task::receive(Task* source)
 {
   XBT_DEBUG("Task %s received a token from %s", name_.c_str(), source->name_.c_str());
-  auto source_count = predecessors_[source]++;
-  if (tokens_received_.size() <= queued_execs_ + source_count)
-    tokens_received_.push_back({});
-  tokens_received_[queued_execs_ + source_count][source] = source->token_;
-  bool enough_tokens = true;
+  auto source_count = predecessors_[source];
+  predecessors_[source]++;
+  if (tokens_received_.size() <= queued_firings_ + source_count)
+    tokens_received_.emplace_back();
+  tokens_received_[queued_firings_ + source_count][source] = source->token_;
+  bool enough_tokens                                       = true;
   for (auto const& [key, val] : predecessors_)
     if (val < 1) {
       enough_tokens = false;
@@ -58,7 +59,7 @@ void Task::receive(Task* source)
   if (enough_tokens) {
     for (auto& [key, val] : predecessors_)
       val--;
-    enqueue_execs(1);
+    enqueue_firings(1);
   }
 }
 
@@ -74,31 +75,52 @@ void Task::receive(Task* source)
 void Task::complete()
 {
   xbt_assert(Actor::is_maestro());
-  working_ = false;
+  running_instances_--;
   count_++;
   on_this_completion(this);
   on_completion(this);
-  if (current_activity_)
-    previous_activity_ = std::move(current_activity_);
   for (auto const& t : successors_)
     t->receive(this);
   if (ready_to_run())
     fire();
 }
 
-/** @param n The number of executions to enqueue.
- *  @brief Enqueue executions.
- *  @note Immediatly starts an execution if possible.
+/** @param n The new parallelism degree of the Task.
+ *  @brief Set the parallelism degree of the Task.
+ *  @note When increasing the degree the function starts new instances.
+ *        When decreasing the degree the function does NOT stop running instances.
  */
-void Task::enqueue_execs(int n)
+void Task::set_parallelism_degree(int n)
 {
+  xbt_assert(n > 0, "Parallelism degree of Tasks must be above 0.");
   simgrid::kernel::actor::simcall_answered([this, n] {
-    queued_execs_ += n;
-    if (ready_to_run())
+    parallelism_degree_ = n;
+    while (ready_to_run())
       fire();
   });
 }
 
+/** @param n The number of firings to enqueue.
+ *  @brief Enqueue firing.
+ *  @note Immediatly fire an activity if possible.
+ */
+void Task::enqueue_firings(int n)
+{
+  simgrid::kernel::actor::simcall_answered([this, n] {
+    queued_firings_ += n;
+    while (ready_to_run())
+      fire();
+  });
+}
+
+/** @param name The new name to set.
+ *  @brief Set the name of the Task.
+ */
+void Task::set_name(std::string name)
+{
+  name_ = name;
+}
+
 /** @param amount The amount to set.
  *  @brief Set the amout of work to do.
  *  @note Amount in flop for ExecTask and in bytes for CommTask.
@@ -125,12 +147,16 @@ std::shared_ptr<Token> Task::get_next_token_from(TaskPtr t)
   return tokens_received_.front()[t];
 }
 
-void Task::fire() {
+void Task::fire()
+{
+  if ((int)current_activities_.size() > parallelism_degree_) {
+    current_activities_.pop_front();
+  }
   on_this_start(this);
   on_start(this);
-  working_ = true;
-  queued_execs_ = std::max(queued_execs_ - 1, 0);
-  if (tokens_received_.size() > 0)
+  running_instances_++;
+  queued_firings_ = std::max(queued_firings_ - 1, 0);
+  if (not tokens_received_.empty())
     tokens_received_.pop_front();
 }
 
@@ -192,7 +218,7 @@ ExecTaskPtr ExecTask::init(const std::string& name, double flops, Host* host)
 
 /**
  *  @brief Do one execution of the Task.
- *  @note Call the on_this_start() func. Set its working status as true.
+ *  @note Call the on_this_start() func.
  *  Init and start the underlying Activity.
  */
 void ExecTask::fire()
@@ -200,8 +226,8 @@ void ExecTask::fire()
   Task::fire();
   auto exec = Exec::init()->set_name(get_name())->set_flops_amount(get_amount())->set_host(host_);
   exec->start();
-  exec->on_this_completion_cb([this](Exec const&) { this->complete(); });
-  set_current_activity(exec);
+  exec->on_this_completion_cb([this](Exec const&) { complete(); });
+  store_activity(exec);
 }
 
 /** @ingroup plugin_task
@@ -246,7 +272,7 @@ CommTaskPtr CommTask::init(const std::string& name, double bytes, Host* source,
 
 /**
  *  @brief Do one execution of the Task.
- *  @note Call the on_this_start() func. Set its working status as true.
+ *  @note Call the on_this_start() func.
  *  Init and start the underlying Activity.
  */
 void CommTask::fire()
@@ -254,8 +280,8 @@ void CommTask::fire()
   Task::fire();
   auto comm = Comm::sendto_init(source_, destination_)->set_name(get_name())->set_payload_size(get_amount());
   comm->start();
-  comm->on_this_completion_cb([this](Comm const&) { this->complete(); });
-  set_current_activity(comm);
+  comm->on_this_completion_cb([this](Comm const&) { complete(); });
+  store_activity(comm);
 }
 
 /** @ingroup plugin_task
@@ -339,8 +365,8 @@ void IoTask::fire()
   Task::fire();
   auto io = Io::init()->set_name(get_name())->set_size(get_amount())->set_disk(disk_)->set_op_type(type_);
   io->start();
-  io->on_this_completion_cb([this](Io const&) { this->complete(); });
-  set_current_activity(io);
+  io->on_this_completion_cb([this](Io const&) { complete(); });
+  store_activity(io);
 }
 
 } // namespace simgrid::s4u