src/s4u/s4u_Task.cpp

   1 #include <memory>
   2 #include <simgrid/Exception.hpp>
   3 #include <simgrid/s4u/Comm.hpp>
   4 #include <simgrid/s4u/Exec.hpp>
   5 #include <simgrid/s4u/Io.hpp>
   6 #include <simgrid/s4u/Task.hpp>
   7 #include <simgrid/simix.hpp>
   8
   9 #include "src/simgrid/module.hpp"
  10
  11 SIMGRID_REGISTER_PLUGIN(task, "Battery management", nullptr)
  12 /**
  13   @beginrst
  14
  15
  16 Tasks are designed to represent dataflows, i.e, graphs of Tasks.
  17 Tasks can only be instancied using either
  18 :cpp:func:`simgrid::s4u::ExecTask::init` or :cpp:func:`simgrid::s4u::CommTask::init`
  19 An ExecTask is an Execution Task. Its underlying Activity is an :ref:`Exec <API_s4u_Exec>`.
  20 A CommTask is a Communication Task. Its underlying Activity is a :ref:`Comm <API_s4u_Comm>`.
  21
  22   @endrst
  23  */
  24 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(Task, kernel, "Logging specific to the task plugin");
  25
  26 namespace simgrid::s4u {
  27
  28 Task::Task(const std::string& name) : name_(name) {}
  29
  30 /**
  31  *  @brief Return True if the Task can start a new Activity.
  32  *  @note The Task is ready if not already doing something and there is at least one execution waiting in queue.
  33  */
  34 bool Task::ready_to_run() const
  35 {
  36   return not working_ && queued_firings_ > 0;
  37 }
  38
  39 /**
  40  *  @param source The sender.
  41  *  @brief Receive a token from another Task.
  42  *  @note Check upon reception if the Task has received a token from each of its predecessors,
  43  * and in this case consumes those tokens and enqueue an execution.
  44  */
  45 void Task::receive(Task* source)
  46 {
  47   XBT_DEBUG("Task %s received a token from %s", name_.c_str(), source->name_.c_str());
  48   auto source_count = predecessors_[source];
  49   predecessors_[source]++;
  50   if (tokens_received_.size() <= queued_firings_ + source_count)
  51     tokens_received_.emplace_back();
  52   tokens_received_[queued_firings_ + source_count][source] = source->token_;
  53   bool enough_tokens                                       = true;
  54   for (auto const& [key, val] : predecessors_)
  55     if (val < 1) {
  56       enough_tokens = false;
  57       break;
  58     }
  59   if (enough_tokens) {
  60     for (auto& [key, val] : predecessors_)
  61       val--;
  62     enqueue_firings(1);
  63   }
  64 }
  65
  66 /**
  67  *  @brief Task routine when finishing an execution.
  68  *  @note Set its working status as false.
  69  * Add 1 to its count of finished executions.
  70  * Call the on_this_end func.
  71  * Fire on_end callback.
  72  * Send a token to each of its successors.
  73  * Start a new execution if possible.
  74  */
  75 void Task::complete()
  76 {
  77   xbt_assert(Actor::is_maestro());
  78   working_ = false;
  79   count_++;
  80   on_this_completion(this);
  81   on_completion(this);
  82   if (current_activity_)
  83     previous_activity_ = std::move(current_activity_);
  84   for (auto const& t : successors_)
  85     t->receive(this);
  86   if (ready_to_run())
  87     fire();
  88 }
  89
  90 /** @param n The number of firings to enqueue.
  91  *  @brief Enqueue firing.
  92  *  @note Immediatly fire an activity if possible.
  93  */
  94 void Task::enqueue_firings(int n)
  95 {
  96   simgrid::kernel::actor::simcall_answered([this, n] {
  97     queued_firings_ += n;
  98     if (ready_to_run())
  99       fire();
 100   });
 101 }
 102
 103 /** @param amount The amount to set.
 104  *  @brief Set the amout of work to do.
 105  *  @note Amount in flop for ExecTask and in bytes for CommTask.
 106  */
 107 void Task::set_amount(double amount)
 108 {
 109   simgrid::kernel::actor::simcall_answered([this, amount] { amount_ = amount; });
 110 }
 111
 112 /** @param token The token to set.
 113  *  @brief Set the token to send to successors.
 114  *  @note The token is passed to each successor after the task end, i.e., after the on_end callback.
 115  */
 116 void Task::set_token(std::shared_ptr<Token> token)
 117 {
 118   simgrid::kernel::actor::simcall_answered([this, token] { token_ = token; });
 119 }
 120
 121 /** @return Map of tokens received for the next execution.
 122  *  @note If there is no queued execution for this task the map might not exist or be partially empty.
 123  */
 124 std::shared_ptr<Token> Task::get_next_token_from(TaskPtr t)
 125 {
 126   return tokens_received_.front()[t];
 127 }
 128
 129 void Task::fire()
 130 {
 131   on_this_start(this);
 132   on_start(this);
 133   working_        = true;
 134   queued_firings_ = std::max(queued_firings_ - 1, 0);
 135   if (not tokens_received_.empty())
 136     tokens_received_.pop_front();
 137 }
 138
 139 /** @param successor The Task to add.
 140  *  @brief Add a successor to this Task.
 141  *  @note It also adds this as a predecessor of successor.
 142  */
 143 void Task::add_successor(TaskPtr successor)
 144 {
 145   simgrid::kernel::actor::simcall_answered([this, successor_p = successor.get()] {
 146     successors_.insert(successor_p);
 147     successor_p->predecessors_.try_emplace(this, 0);
 148   });
 149 }
 150
 151 /** @param successor The Task to remove.
 152  *  @brief Remove a successor from this Task.
 153  *  @note It also remove this from the predecessors of successor.
 154  */
 155 void Task::remove_successor(TaskPtr successor)
 156 {
 157   simgrid::kernel::actor::simcall_answered([this, successor_p = successor.get()] {
 158     successor_p->predecessors_.erase(this);
 159     successors_.erase(successor_p);
 160   });
 161 }
 162
 163 void Task::remove_all_successors()
 164 {
 165   simgrid::kernel::actor::simcall_answered([this] {
 166     while (not successors_.empty()) {
 167       auto* successor = *(successors_.begin());
 168       successor->predecessors_.erase(this);
 169       successors_.erase(successor);
 170     }
 171   });
 172 }
 173
 174 /**
 175  *  @brief Default constructor.
 176  */
 177 ExecTask::ExecTask(const std::string& name) : Task(name) {}
 178
 179 /** @ingroup plugin_task
 180  *  @brief Smart Constructor.
 181  */
 182 ExecTaskPtr ExecTask::init(const std::string& name)
 183 {
 184   return ExecTaskPtr(new ExecTask(name));
 185 }
 186
 187 /** @ingroup plugin_task
 188  *  @brief Smart Constructor.
 189  */
 190 ExecTaskPtr ExecTask::init(const std::string& name, double flops, Host* host)
 191 {
 192   return init(name)->set_flops(flops)->set_host(host);
 193 }
 194
 195 /**
 196  *  @brief Do one execution of the Task.
 197  *  @note Call the on_this_start() func. Set its working status as true.
 198  *  Init and start the underlying Activity.
 199  */
 200 void ExecTask::fire()
 201 {
 202   Task::fire();
 203   auto exec = Exec::init()->set_name(get_name())->set_flops_amount(get_amount())->set_host(host_);
 204   exec->start();
 205   exec->on_this_completion_cb([this](Exec const&) { this->complete(); });
 206   set_current_activity(exec);
 207 }
 208
 209 /** @ingroup plugin_task
 210  *  @param host The host to set.
 211  *  @brief Set a new host.
 212  */
 213 ExecTaskPtr ExecTask::set_host(Host* host)
 214 {
 215   kernel::actor::simcall_answered([this, host] { host_ = host; });
 216   return this;
 217 }
 218
 219 /** @ingroup plugin_task
 220  *  @param flops The amount of flops to set.
 221  */
 222 ExecTaskPtr ExecTask::set_flops(double flops)
 223 {
 224   kernel::actor::simcall_answered([this, flops] { set_amount(flops); });
 225   return this;
 226 }
 227
 228 /**
 229  *  @brief Default constructor.
 230  */
 231 CommTask::CommTask(const std::string& name) : Task(name) {}
 232
 233 /** @ingroup plugin_task
 234  *  @brief Smart constructor.
 235  */
 236 CommTaskPtr CommTask::init(const std::string& name)
 237 {
 238   return CommTaskPtr(new CommTask(name));
 239 }
 240
 241 /** @ingroup plugin_task
 242  *  @brief Smart constructor.
 243  */
 244 CommTaskPtr CommTask::init(const std::string& name, double bytes, Host* source, Host* destination)
 245 {
 246   return init(name)->set_bytes(bytes)->set_source(source)->set_destination(destination);
 247 }
 248
 249 /**
 250  *  @brief Do one execution of the Task.
 251  *  @note Call the on_this_start() func. Set its working status as true.
 252  *  Init and start the underlying Activity.
 253  */
 254 void CommTask::fire()
 255 {
 256   Task::fire();
 257   auto comm = Comm::sendto_init(source_, destination_)->set_name(get_name())->set_payload_size(get_amount());
 258   comm->start();
 259   comm->on_this_completion_cb([this](Comm const&) { this->complete(); });
 260   set_current_activity(comm);
 261 }
 262
 263 /** @ingroup plugin_task
 264  *  @param source The host to set.
 265  *  @brief Set a new source host.
 266  */
 267 CommTaskPtr CommTask::set_source(Host* source)
 268 {
 269   kernel::actor::simcall_answered([this, source] { source_ = source; });
 270   return this;
 271 }
 272
 273 /** @ingroup plugin_task
 274  *  @param destination The host to set.
 275  *  @brief Set a new destination host.
 276  */
 277 CommTaskPtr CommTask::set_destination(Host* destination)
 278 {
 279   kernel::actor::simcall_answered([this, destination] { destination_ = destination; });
 280   return this;
 281 }
 282
 283 /** @ingroup plugin_task
 284  *  @param bytes The amount of bytes to set.
 285  */
 286 CommTaskPtr CommTask::set_bytes(double bytes)
 287 {
 288   kernel::actor::simcall_answered([this, bytes] { set_amount(bytes); });
 289   return this;
 290 }
 291
 292 /**
 293  *  @brief Default constructor.
 294  */
 295 IoTask::IoTask(const std::string& name) : Task(name) {}
 296
 297 /** @ingroup plugin_task
 298  *  @brief Smart Constructor.
 299  */
 300 IoTaskPtr IoTask::init(const std::string& name)
 301 {
 302   return IoTaskPtr(new IoTask(name));
 303 }
 304
 305 /** @ingroup plugin_task
 306  *  @brief Smart Constructor.
 307  */
 308 IoTaskPtr IoTask::init(const std::string& name, double bytes, Disk* disk, Io::OpType type)
 309 {
 310   return init(name)->set_bytes(bytes)->set_disk(disk)->set_op_type(type);
 311 }
 312
 313 /** @ingroup plugin_task
 314  *  @param disk The disk to set.
 315  *  @brief Set a new disk.
 316  */
 317 IoTaskPtr IoTask::set_disk(Disk* disk)
 318 {
 319   kernel::actor::simcall_answered([this, disk] { disk_ = disk; });
 320   return this;
 321 }
 322
 323 /** @ingroup plugin_task
 324  *  @param bytes The amount of bytes to set.
 325  */
 326 IoTaskPtr IoTask::set_bytes(double bytes)
 327 {
 328   kernel::actor::simcall_answered([this, bytes] { set_amount(bytes); });
 329   return this;
 330 }
 331
 332 /** @ingroup plugin_task */
 333 IoTaskPtr IoTask::set_op_type(Io::OpType type)
 334 {
 335   kernel::actor::simcall_answered([this, type] { type_ = type; });
 336   return this;
 337 }
 338
 339 void IoTask::fire()
 340 {
 341   Task::fire();
 342   auto io = Io::init()->set_name(get_name())->set_size(get_amount())->set_disk(disk_)->set_op_type(type_);
 343   io->start();
 344   io->on_this_completion_cb([this](Io const&) { this->complete(); });
 345   set_current_activity(io);
 346 }
 347
 348 } // namespace simgrid::s4u