src/plugins/task.cpp

   1 #include <simgrid/Exception.hpp>
   2 #include <simgrid/plugins/task.hpp>
   3 #include <simgrid/s4u/Comm.hpp>
   4 #include <simgrid/s4u/Exec.hpp>
   5 #include <simgrid/s4u/Io.hpp>
   6 #include <simgrid/simix.hpp>
   7
   8 #include "src/simgrid/module.hpp"
   9
  10 SIMGRID_REGISTER_PLUGIN(task, "Battery management", nullptr)
  11 /** @defgroup plugin_task plugin_task Plugin Task
  12
  13   @beginrst
  14
  15 This is the task plugin, enabling management of Tasks.
  16 To activate this plugin, first call :cpp:func:`Task::init`.
  17
  18 Tasks are designed to represent dataflows, i.e, graphs of Tasks.
  19 Tasks can only be instancied using either
  20 :cpp:func:`simgrid::plugins::ExecTask::init` or :cpp:func:`simgrid::plugins::CommTask::init`
  21 An ExecTask is an Execution Task. Its underlying Activity is an :ref:`Exec <API_s4u_Exec>`.
  22 A CommTask is a Communication Task. Its underlying Activity is a :ref:`Comm <API_s4u_Comm>`.
  23
  24   @endrst
  25  */
  26 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(Task, kernel, "Logging specific to the task plugin");
  27
  28 namespace simgrid::plugins {
  29
  30 xbt::signal<void(Task*)> Task::on_start;
  31 xbt::signal<void(Task*)> Task::on_end;
  32
  33 Task::Task(const std::string& name) : name_(name) {}
  34
  35 /**
  36  *  @param predecessor The Task to add.
  37  *  @brief Add a predecessor to this Task.
  38  */
  39 void Task::add_predecessor(Task* predecessor)
  40 {
  41   if (predecessors_.find(predecessor) == predecessors_.end())
  42     simgrid::kernel::actor::simcall_answered([this, predecessor] { predecessors_[predecessor] = 0; });
  43 }
  44
  45 /**
  46  *  @param predecessor The Task to remove.
  47  *  @brief Remove a predecessor from this Task.
  48  */
  49 void Task::remove_predecessor(Task* predecessor)
  50 {
  51   simgrid::kernel::actor::simcall_answered([this, predecessor] { predecessors_.erase(predecessor); });
  52 }
  53
  54 /**
  55  *  @brief Return True if the Task can start a new Activity.
  56  *  @note The Task is ready if not already doing something and there is at least one execution waiting in queue.
  57  */
  58 bool Task::ready_to_run() const
  59 {
  60   return not working_ && queued_execs_ > 0;
  61 }
  62
  63 /**
  64  *  @param source The sender.
  65  *  @brief Receive a token from another Task.
  66  *  @note Check upon reception if the Task has received a token from each of its predecessors,
  67  * and in this case consumes those tokens and enqueue an execution.
  68  */
  69 void Task::receive(Task* source)
  70 {
  71   XBT_DEBUG("Task %s received a token from %s", name_.c_str(), source->name_.c_str());
  72   auto it = predecessors_.find(source);
  73   simgrid::kernel::actor::simcall_answered([this, it] {
  74     it->second++;
  75     bool enough_tokens = true;
  76     for (auto const& [key, val] : predecessors_)
  77       if (val < 1) {
  78         enough_tokens = false;
  79         break;
  80       }
  81     if (enough_tokens) {
  82       for (auto& [key, val] : predecessors_)
  83         val--;
  84       enqueue_execs(1);
  85     }
  86   });
  87 }
  88
  89 /**
  90  *  @brief Task routine when finishing an execution.
  91  *  @note Set its working status as false.
  92  * Add 1 to its count of finished executions.
  93  * Call the on_this_end func.
  94  * Fire on_end callback.
  95  * Send a token to each of its successors.
  96  * Start a new execution if possible.
  97  */
  98 void Task::complete()
  99 {
 100   simgrid::kernel::actor::simcall_answered([this] {
 101     working_ = false;
 102     count_++;
 103   });
 104   for (auto const& end_func : end_func_handlers_)
 105     end_func(this);
 106   Task::on_end(this);
 107   for (auto const& t : successors_)
 108     t->receive(this);
 109   if (ready_to_run())
 110     fire();
 111 }
 112
 113 /** @ingroup plugin_task
 114  *  @brief Init the Task plugin.
 115  *  @note Add a completion callback to all Activities to call Task::complete().
 116  */
 117 void Task::init()
 118 {
 119   if (Task::inited_)
 120     return;
 121   Task::inited_                           = true;
 122   ExtendedAttributeActivity::EXTENSION_ID = simgrid::s4u::Activity::extension_create<ExtendedAttributeActivity>();
 123   simgrid::s4u::Exec::on_completion_cb(
 124       [](simgrid::s4u::Exec const& exec) { exec.extension<ExtendedAttributeActivity>()->task_->complete(); });
 125   simgrid::s4u::Comm::on_completion_cb(
 126       [](simgrid::s4u::Comm const& comm) { comm.extension<ExtendedAttributeActivity>()->task_->complete(); });
 127   simgrid::s4u::Io::on_completion_cb(
 128       [](simgrid::s4u::Io const& io) { io.extension<ExtendedAttributeActivity>()->task_->complete(); });
 129 }
 130
 131 /** @ingroup plugin_task
 132  *  @param n The number of executions to enqueue.
 133  *  @brief Enqueue executions.
 134  *  @note Immediatly starts an execution if possible.
 135  */
 136 void Task::enqueue_execs(int n)
 137 {
 138   simgrid::kernel::actor::simcall_answered([this, n] {
 139     queued_execs_ += n;
 140     if (ready_to_run())
 141       fire();
 142   });
 143 }
 144
 145 /** @ingroup plugin_task
 146  *  @param amount The amount to set.
 147  *  @brief Set the amout of work to do.
 148  *  @note Amount in flop for ExecTask and in bytes for CommTask.
 149  */
 150 void Task::set_amount(double amount)
 151 {
 152   simgrid::kernel::actor::simcall_answered([this, amount] { amount_ = amount; });
 153 }
 154
 155 /** @ingroup plugin_task
 156  *  @param successor The Task to add.
 157  *  @brief Add a successor to this Task.
 158  *  @note It also adds this as a predecessor of successor.
 159  */
 160 void Task::add_successor(TaskPtr successor)
 161 {
 162   simgrid::kernel::actor::simcall_answered([this, successor] { successors_.insert(successor.get()); });
 163   successor->add_predecessor(this);
 164 }
 165
 166 /** @ingroup plugin_task
 167  *  @param successor The Task to remove.
 168  *  @brief Remove a successor from this Task.
 169  *  @note It also remove this from the predecessors of successor.
 170  */
 171 void Task::remove_successor(TaskPtr successor)
 172 {
 173   simgrid::kernel::actor::simcall_answered([this, successor] { successors_.erase(successor.get()); });
 174   successor->remove_predecessor(this);
 175 }
 176
 177 void Task::remove_all_successors()
 178 {
 179   simgrid::kernel::actor::simcall_answered([this] {
 180     while (not successors_.empty()) {
 181       auto* successor = *(successors_.begin());
 182       successor->predecessors_.erase(this);
 183       successors_.erase(successor);
 184     }
 185   });
 186 }
 187
 188 /** @ingroup plugin_task
 189  *  @param func The function to set.
 190  *  @brief Set a function to be called before each execution.
 191  *  @note The function is called before the underlying Activity starts.
 192  */
 193 void Task::on_this_start(const std::function<void(Task*)>& func)
 194 {
 195   simgrid::kernel::actor::simcall_answered([this, &func] { start_func_handlers_.push_back(func); });
 196 }
 197
 198 /** @ingroup plugin_task
 199  *  @param func The function to set.
 200  *  @brief Set a function to be called after each execution.
 201  *  @note The function is called after the underlying Activity ends, but before sending tokens to successors.
 202  */
 203 void Task::on_this_end(const std::function<void(Task*)>& func)
 204 {
 205   simgrid::kernel::actor::simcall_answered([this, &func] { end_func_handlers_.push_back(func); });
 206 }
 207
 208 /** @ingroup plugin_task
 209  *  @brief Return the number of completed executions.
 210  */
 211 int Task::get_count() const
 212 {
 213   return count_;
 214 }
 215
 216 /**
 217  *  @brief Default constructor.
 218  */
 219 ExecTask::ExecTask(const std::string& name) : Task(name) {}
 220
 221 /** @ingroup plugin_task
 222  *  @brief Smart Constructor.
 223  */
 224 ExecTaskPtr ExecTask::init(const std::string& name)
 225 {
 226   return ExecTaskPtr(new ExecTask(name));
 227 }
 228
 229 /** @ingroup plugin_task
 230  *  @brief Smart Constructor.
 231  */
 232 ExecTaskPtr ExecTask::init(const std::string& name, double flops, s4u::Host* host)
 233 {
 234   return init(name)->set_flops(flops)->set_host(host);
 235 }
 236
 237 /**
 238  *  @brief Do one execution of the Task.
 239  *  @note Call the on_this_start() func. Set its working status as true.
 240  *  Init and start the underlying Activity.
 241  */
 242 void ExecTask::fire()
 243 {
 244   for (auto const& start_func : start_func_handlers_)
 245     start_func(this);
 246   Task::on_start(this);
 247   kernel::actor::simcall_answered([this] {
 248     working_      = true;
 249     queued_execs_ = std::max(queued_execs_ - 1, 0);
 250   });
 251   s4u::ExecPtr exec = s4u::Exec::init();
 252   exec->set_name(name_);
 253   exec->set_flops_amount(amount_);
 254   exec->set_host(host_);
 255   exec->start();
 256   exec->extension_set(new ExtendedAttributeActivity());
 257   exec->extension<ExtendedAttributeActivity>()->task_ = this;
 258   kernel::actor::simcall_answered([this, exec] { current_activity_ = exec; });
 259 }
 260
 261 /** @ingroup plugin_task
 262  *  @param host The host to set.
 263  *  @brief Set a new host.
 264  */
 265 ExecTaskPtr ExecTask::set_host(s4u::Host* host)
 266 {
 267   kernel::actor::simcall_answered([this, host] { host_ = host; });
 268   return this;
 269 }
 270
 271 /** @ingroup plugin_task
 272  *  @param flops The amount of flops to set.
 273  */
 274 ExecTaskPtr ExecTask::set_flops(double flops)
 275 {
 276   kernel::actor::simcall_answered([this, flops] { amount_ = flops; });
 277   return this;
 278 }
 279
 280 /**
 281  *  @brief Default constructor.
 282  */
 283 CommTask::CommTask(const std::string& name) : Task(name) {}
 284
 285 /** @ingroup plugin_task
 286  *  @brief Smart constructor.
 287  */
 288 CommTaskPtr CommTask::init(const std::string& name)
 289 {
 290   return CommTaskPtr(new CommTask(name));
 291 }
 292
 293 /** @ingroup plugin_task
 294  *  @brief Smart constructor.
 295  */
 296 CommTaskPtr CommTask::init(const std::string& name, double bytes, s4u::Host* source, s4u::Host* destination)
 297 {
 298   return init(name)->set_bytes(bytes)->set_source(source)->set_destination(destination);
 299 }
 300
 301 /**
 302  *  @brief Do one execution of the Task.
 303  *  @note Call the on_this_start() func. Set its working status as true.
 304  *  Init and start the underlying Activity.
 305  */
 306 void CommTask::fire()
 307 {
 308   for (auto const& start_func : start_func_handlers_)
 309     start_func(this);
 310   Task::on_start(this);
 311   kernel::actor::simcall_answered([this] {
 312     working_      = true;
 313     queued_execs_ = std::max(queued_execs_ - 1, 0);
 314   });
 315   s4u::CommPtr comm = s4u::Comm::sendto_init(source_, destination_);
 316   comm->set_name(name_);
 317   comm->set_payload_size(amount_);
 318   comm->start();
 319   comm->extension_set(new ExtendedAttributeActivity());
 320   comm->extension<ExtendedAttributeActivity>()->task_ = this;
 321   kernel::actor::simcall_answered([this, comm] { current_activity_ = comm; });
 322 }
 323
 324 /** @ingroup plugin_task
 325  *  @param source The host to set.
 326  *  @brief Set a new source host.
 327  */
 328 CommTaskPtr CommTask::set_source(s4u::Host* source)
 329 {
 330   kernel::actor::simcall_answered([this, source] { source_ = source; });
 331   return this;
 332 }
 333
 334 /** @ingroup plugin_task
 335  *  @param destination The host to set.
 336  *  @brief Set a new destination host.
 337  */
 338 CommTaskPtr CommTask::set_destination(s4u::Host* destination)
 339 {
 340   kernel::actor::simcall_answered([this, destination] { destination_ = destination; });
 341   return this;
 342 }
 343
 344 /** @ingroup plugin_task
 345  *  @param bytes The amount of bytes to set.
 346  */
 347 CommTaskPtr CommTask::set_bytes(double bytes)
 348 {
 349   kernel::actor::simcall_answered([this, bytes] { amount_ = bytes; });
 350   return this;
 351 }
 352
 353 /**
 354  *  @brief Default constructor.
 355  */
 356 IoTask::IoTask(const std::string& name) : Task(name) {}
 357
 358 /** @ingroup plugin_task
 359  *  @brief Smart Constructor.
 360  */
 361 IoTaskPtr IoTask::init(const std::string& name)
 362 {
 363   return IoTaskPtr(new IoTask(name));
 364 }
 365
 366 /** @ingroup plugin_task
 367  *  @brief Smart Constructor.
 368  */
 369 IoTaskPtr IoTask::init(const std::string& name, double bytes, s4u::Disk* disk, s4u::Io::OpType type)
 370 {
 371   return init(name)->set_bytes(bytes)->set_disk(disk)->set_op_type(type);
 372 }
 373
 374 /** @ingroup plugin_task
 375  *  @param disk The disk to set.
 376  *  @brief Set a new disk.
 377  */
 378 IoTaskPtr IoTask::set_disk(s4u::Disk* disk)
 379 {
 380   kernel::actor::simcall_answered([this, disk] { disk_ = disk; });
 381   return this;
 382 }
 383
 384 /** @ingroup plugin_task
 385  *  @param bytes The amount of bytes to set.
 386  */
 387 IoTaskPtr IoTask::set_bytes(double bytes)
 388 {
 389   kernel::actor::simcall_answered([this, bytes] { amount_ = bytes; });
 390   return this;
 391 }
 392
 393 /** @ingroup plugin_task */
 394 IoTaskPtr IoTask::set_op_type(s4u::Io::OpType type)
 395 {
 396   kernel::actor::simcall_answered([this, type] { type_ = type; });
 397   return this;
 398 }
 399
 400 void IoTask::fire()
 401 {
 402   for (auto const& start_func : start_func_handlers_)
 403     start_func(this);
 404   Task::on_start(this);
 405   kernel::actor::simcall_answered([this] {
 406     working_      = true;
 407     queued_execs_ = std::max(queued_execs_ - 1, 0);
 408   });
 409   s4u::IoPtr io = s4u::Io::init();
 410   io->set_name(name_);
 411   io->set_size(amount_);
 412   io->set_disk(disk_);
 413   io->set_op_type(type_);
 414   io->start();
 415   io->extension_set(new ExtendedAttributeActivity());
 416   io->extension<ExtendedAttributeActivity>()->task_ = this;
 417   kernel::actor::simcall_answered([this, io] { current_activity_ = io; });
 418 }
 419
 420 } // namespace simgrid::plugins
 421
 422 simgrid::xbt::Extension<simgrid::s4u::Activity, simgrid::plugins::ExtendedAttributeActivity>
 423     simgrid::plugins::ExtendedAttributeActivity::EXTENSION_ID;
 424 bool simgrid::plugins::Task::inited_ = false;