src/msg/msg_task.cpp

   1 /* Copyright (c) 2004-2022. The SimGrid Team. All rights reserved.          */
   2
   3 /* This program is free software; you can redistribute it and/or modify it
   4  * under the terms of the license (GNU LGPL) which comes with this package. */
   5
   6 #include "msg_private.hpp"
   7 #include "src/instr/instr_private.hpp"
   8 #include <simgrid/s4u/Comm.hpp>
   9 #include <simgrid/s4u/Exec.hpp>
  10 #include <simgrid/s4u/Host.hpp>
  11 #include <simgrid/s4u/Mailbox.hpp>
  12
  13 #include <algorithm>
  14 #include <vector>
  15
  16 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_task, msg, "Logging specific to MSG (task)");
  17
  18 namespace simgrid {
  19
  20 template class xbt::Extendable<msg::Task>;
  21
  22 namespace msg {
  23
  24 Task::Task(const std::string& name, double flops_amount, double bytes_amount, void* data)
  25     : name_(name), flops_amount(flops_amount), bytes_amount(bytes_amount)
  26 {
  27   static std::atomic_ullong counter{0};
  28   id_ = counter++;
  29   set_data(data);
  30   MC_ignore_heap(&id_, sizeof id_);
  31 }
  32
  33 Task::Task(const std::string& name, std::vector<s4u::Host*>&& hosts, std::vector<double>&& flops_amount,
  34            std::vector<double>&& bytes_amount, void* data)
  35     : Task(name, 1.0, 0, data)
  36 {
  37   parallel_             = true;
  38   hosts_                = std::move(hosts);
  39   flops_parallel_amount = std::move(flops_amount);
  40   bytes_parallel_amount = std::move(bytes_amount);
  41 }
  42
  43 Task* Task::create(const std::string& name, double flops_amount, double bytes_amount, void* data)
  44 {
  45   return new Task(name, flops_amount, bytes_amount, data);
  46 }
  47
  48 Task* Task::create_parallel(const std::string& name, int host_nb, const msg_host_t* host_list, double* flops_amount,
  49                             double* bytes_amount, void* data)
  50 {
  51   std::vector<s4u::Host*> hosts(host_list, host_list + host_nb);
  52   std::vector<double> flops;
  53   std::vector<double> bytes;
  54   if (flops_amount != nullptr)
  55     flops = std::vector<double>(flops_amount, flops_amount + host_nb);
  56   if (bytes_amount != nullptr)
  57     bytes = std::vector<double>(bytes_amount, bytes_amount + host_nb * host_nb);
  58
  59   return new Task(name, std::move(hosts), std::move(flops), std::move(bytes), data);
  60 }
  61
  62 msg_error_t Task::execute()
  63 {
  64   /* checking for infinite values */
  65   xbt_assert(std::isfinite(flops_amount), "flops_amount is not finite!");
  66
  67   msg_error_t status = MSG_OK;
  68   if (flops_amount <= 0.0)
  69     return MSG_OK;
  70
  71   try {
  72     set_used();
  73     if (parallel_)
  74       compute = s4u::this_actor::exec_init(hosts_, flops_parallel_amount, bytes_parallel_amount);
  75     else
  76       compute = s4u::this_actor::exec_init(flops_amount);
  77
  78     compute->set_name(name_)
  79         ->set_tracing_category(tracing_category_)
  80         ->set_priority(1 / priority_)
  81         ->set_bound(bound_)
  82         ->wait_for(timeout_);
  83
  84     set_not_used();
  85     XBT_DEBUG("Execution task '%s' finished", get_cname());
  86   } catch (const HostFailureException&) {
  87     status = MSG_HOST_FAILURE;
  88   } catch (const TimeoutException&) {
  89     status = MSG_TIMEOUT;
  90   } catch (const CancelException&) {
  91     status = MSG_TASK_CANCELED;
  92   }
  93
  94   /* action ended, set comm and compute = nullptr, the actions is already destroyed in the main function */
  95   flops_amount = 0.0;
  96   comm         = nullptr;
  97   compute      = nullptr;
  98
  99   return status;
 100 }
 101
 102 s4u::CommPtr Task::send_async(const std::string& alias, void_f_pvoid_t cleanup, bool detached)
 103 {
 104   if (TRACE_actor_is_enabled()) {
 105     auto* process_container       = instr::Container::by_name(instr_pid(*MSG_process_self()));
 106     std::string key               = "p" + std::to_string(get_id());
 107     instr::Container::get_root()->get_link("ACTOR_LINK")->start_event(process_container, "SR", key);
 108   }
 109
 110   /* Prepare the task to send */
 111   set_used();
 112   this->comm = nullptr;
 113   msg_global->sent_msg++;
 114
 115   s4u::CommPtr s4u_comm = s4u::Mailbox::by_name(alias)->put_init(this, bytes_amount)->set_rate(get_rate());
 116   if (TRACE_is_enabled() && has_tracing_category())
 117     s4u_comm->set_tracing_category(tracing_category_);
 118
 119   comm                  = s4u_comm;
 120
 121   if (detached)
 122     comm->detach(cleanup);
 123   else
 124     comm->start();
 125
 126   return comm;
 127 }
 128
 129 msg_error_t Task::send(const std::string& alias, double timeout)
 130 {
 131   msg_error_t ret = MSG_OK;
 132   /* Try to send it */
 133   try {
 134     comm = nullptr; // needed, otherwise MC gets confused.
 135     s4u::CommPtr s4u_comm = send_async(alias, nullptr, false);
 136     comm                  = s4u_comm;
 137     comm->wait_for(timeout);
 138   } catch (const TimeoutException&) {
 139     ret = MSG_TIMEOUT;
 140   } catch (const CancelException&) {
 141     ret = MSG_HOST_FAILURE;
 142   } catch (const NetworkFailureException&) {
 143     ret = MSG_TRANSFER_FAILURE;
 144     /* If the send failed, it is not used anymore */
 145     set_not_used();
 146   }
 147
 148   return ret;
 149 }
 150 void Task::cancel()
 151 {
 152   if (compute) {
 153     compute->cancel();
 154   } else if (comm) {
 155     comm->cancel();
 156   }
 157   set_not_used();
 158 }
 159
 160 void Task::set_priority(double priority)
 161 {
 162   xbt_assert(std::isfinite(1.0 / priority), "priority is not finite!");
 163   priority_ = 1.0 / priority;
 164 }
 165
 166 s4u::Actor* Task::get_sender() const
 167 {
 168   return comm ? comm->get_sender() : nullptr;
 169 }
 170
 171 s4u::Host* Task::get_source() const
 172 {
 173   return comm ? comm->get_sender()->get_host() : nullptr;
 174 }
 175
 176 void Task::set_used()
 177 {
 178   if (is_used_)
 179     report_multiple_use();
 180   is_used_ = true;
 181 }
 182
 183 void Task::report_multiple_use() const
 184 {
 185   if (MSG_Global_t::debug_multiple_use) {
 186     XBT_ERROR("This task is already used in there:");
 187     // TODO, backtrace
 188     XBT_ERROR("<missing backtrace>");
 189     XBT_ERROR("And you try to reuse it from here:");
 190     xbt_backtrace_display_current();
 191   } else {
 192     xbt_die("This task is still being used somewhere else. You cannot send it now. Go fix your code!"
 193              "(use --cfg=msg/debug-multiple-use:on to get the backtrace of the other process)");
 194   }
 195 }
 196 } // namespace msg
 197 } // namespace simgrid
 198
 199 /********************************* Task **************************************/
 200 /** @brief Creates a new task
 201  *
 202  * A constructor for msg_task_t taking four arguments.
 203  *
 204  * @param name a name for the object. It is for user-level information and can be nullptr.
 205  * @param flop_amount a value of the processing amount (in flop) needed to process this new task.
 206  * If 0, then it cannot be executed with MSG_task_execute(). This value has to be >=0.
 207  * @param message_size a value of the amount of data (in bytes) needed to transfer this new task. If 0, then it cannot
 208  * be transferred with MSG_task_send() and MSG_task_recv(). This value has to be >=0.
 209  * @param data a pointer to any data may want to attach to the new object.  It is for user-level information and can
 210  * be nullptr. It can be retrieved with the function MSG_task_get_data().
 211  * @return The new corresponding object.
 212  */
 213 msg_task_t MSG_task_create(const char *name, double flop_amount, double message_size, void *data)
 214 {
 215   return simgrid::msg::Task::create(name ? name : "", flop_amount, message_size, data);
 216 }
 217
 218 /** @brief Creates a new parallel task
 219  *
 220  * A constructor for #msg_task_t taking six arguments.
 221  *
 222  * @beginrst
 223  * See :ref:`simgrid::s4u::this_actor::parallel_execute() <API_s4u_parallel_execute>` for
 224  * the exact semantic of the parameters.
 225  * @endrst
 226  *
 227  * @param name a name for the object. It is for user-level information and can be nullptr.
 228  * @param host_nb the number of hosts implied in the parallel task.
 229  * @param host_list an array of @p host_nb msg_host_t.
 230  * @param flops_amount an array of @p host_nb doubles.
 231  *        flops_amount[i] is the total number of operations that have to be performed on host_list[i].
 232  * @param bytes_amount an array of @p host_nb* @p host_nb doubles.
 233  * @param data a pointer to any data may want to attach to the new object.
 234  *             It is for user-level information and can be nullptr.
 235  *             It can be retrieved with the function MSG_task_get_data().
 236  */
 237 msg_task_t MSG_parallel_task_create(const char *name, int host_nb, const msg_host_t * host_list,
 238                                     double *flops_amount, double *bytes_amount, void *data)
 239 {
 240   // Task's flops amount is set to an arbitrary value > 0.0 to be able to distinguish, in
 241   // MSG_task_get_remaining_work_ratio(), a finished task and a task that has not started yet.
 242   return simgrid::msg::Task::create_parallel(name ? name : "", host_nb, host_list, flops_amount, bytes_amount, data);
 243 }
 244
 245 /** @brief Return the user data of the given task */
 246 void* MSG_task_get_data(const_msg_task_t task)
 247 {
 248   return task->get_data<void>();
 249 }
 250
 251 /** @brief Sets the user data of a given task */
 252 void MSG_task_set_data(msg_task_t task, void *data)
 253 {
 254   task->set_data(data);
 255 }
 256
 257 /** @brief Returns the sender of the given task */
 258 msg_process_t MSG_task_get_sender(const_msg_task_t task)
 259 {
 260   return task->get_sender();
 261 }
 262
 263 /** @brief Returns the source (the sender's host) of the given task */
 264 msg_host_t MSG_task_get_source(const_msg_task_t task)
 265 {
 266   return task->get_source();
 267 }
 268
 269 /** @brief Returns the name of the given task. */
 270 const char* MSG_task_get_name(const_msg_task_t task)
 271 {
 272   return task->get_cname();
 273 }
 274
 275 /** @brief Sets the name of the given task. */
 276 void MSG_task_set_name(msg_task_t task, const char *name)
 277 {
 278   task->set_name(name);
 279 }
 280
 281 /**
 282  * @brief Executes a task and waits for its termination.
 283  *
 284  * This function is used for describing the behavior of a process. It takes only one parameter.
 285  * @param task a #msg_task_t to execute on the location on which the process is running.
 286  * @return #MSG_OK if the task was successfully completed, #MSG_TASK_CANCELED or #MSG_HOST_FAILURE otherwise
 287  */
 288 msg_error_t MSG_task_execute(msg_task_t task)
 289 {
 290   return task->execute();
 291 }
 292
 293 /**
 294  * @brief Executes a parallel task and waits for its termination.
 295  *
 296  * @param task a #msg_task_t to execute on the location on which the process is running.
 297  *
 298  * @return #MSG_OK if the task was successfully completed, #MSG_TASK_CANCELED or #MSG_HOST_FAILURE otherwise
 299  */
 300 msg_error_t MSG_parallel_task_execute(msg_task_t task)
 301 {
 302   return task->execute();
 303 }
 304
 305 msg_error_t MSG_parallel_task_execute_with_timeout(msg_task_t task, double timeout)
 306 {
 307   task->set_timeout(timeout);
 308   return task->execute();
 309 }
 310
 311 /**
 312  * @brief Sends a task on a mailbox.
 313  *
 314  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication.
 315  *
 316  * @param task a #msg_task_t to send on another location.
 317  * @param alias name of the mailbox to sent the task to
 318  * @return the msg_comm_t communication created
 319  */
 320 msg_comm_t MSG_task_isend(msg_task_t task, const char* alias)
 321 {
 322   return new simgrid::msg::Comm(task, nullptr, task->send_async(alias, nullptr, false));
 323 }
 324
 325 /**
 326  * @brief Sends a task on a mailbox with a maximum rate
 327  *
 328  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication. The maxrate
 329  * parameter allows the application to limit the bandwidth utilization of network links when sending the task.
 330  *
 331  * @param task a #msg_task_t to send on another location.
 332  * @param alias name of the mailbox to sent the task to
 333  * @param maxrate the maximum communication rate for sending this task (byte/sec).
 334  * @return the msg_comm_t communication created
 335  */
 336 msg_comm_t MSG_task_isend_bounded(msg_task_t task, const char* alias, double maxrate)
 337 {
 338   task->set_rate(maxrate);
 339   return new simgrid::msg::Comm(task, nullptr, task->send_async(alias, nullptr, false));
 340 }
 341
 342 /**
 343  * @brief Sends a task on a mailbox.
 344  *
 345  * This is a non blocking detached send function.
 346  * Think of it as a best effort send. Keep in mind that the third parameter is only called if the communication fails.
 347  * If the communication does work, it is responsibility of the receiver code to free anything related to the task, as
 348  * usual. More details on this can be obtained on
 349  * <a href="http://lists.gforge.inria.fr/pipermail/simgrid-user/2011-November/002649.html">this thread</a>
 350  * in the SimGrid-user mailing list archive.
 351  *
 352  * @param task a #msg_task_t to send on another location.
 353  * @param alias name of the mailbox to sent the task to
 354  * @param cleanup a function to destroy the task if the communication fails, e.g. MSG_task_destroy
 355  * (if nullptr, no function will be called)
 356  */
 357 void MSG_task_dsend(msg_task_t task, const char* alias, void_f_pvoid_t cleanup)
 358 {
 359   task->send_async(alias, cleanup, true);
 360 }
 361
 362 /**
 363  * @brief Sends a task on a mailbox with a maximal rate.
 364  *
 365  * This is a non blocking detached send function.
 366  * Think of it as a best effort send. Keep in mind that the third parameter is only called if the communication fails.
 367  * If the communication does work, it is responsibility of the receiver code to free anything related to the task, as
 368  * usual. More details on this can be obtained on
 369  * <a href="http://lists.gforge.inria.fr/pipermail/simgrid-user/2011-November/002649.html">this thread</a>
 370  * in the SimGrid-user mailing list archive.
 371  *
 372  * The rate parameter can be used to send a task with a limited bandwidth (smaller than the physical available value).
 373  * Use MSG_task_dsend() if you don't limit the rate (or pass -1 as a rate value do disable this feature).
 374  *
 375  * @param task a #msg_task_t to send on another location.
 376  * @param alias name of the mailbox to sent the task to
 377  * @param cleanup a function to destroy the task if the communication fails, e.g. MSG_task_destroy (if nullptr, no
 378  *        function will be called)
 379  * @param maxrate the maximum communication rate for sending this task (byte/sec)
 380  *
 381  */
 382 void MSG_task_dsend_bounded(msg_task_t task, const char* alias, void_f_pvoid_t cleanup, double maxrate)
 383 {
 384   task->set_rate(maxrate);
 385   task->send_async(alias, cleanup, true);
 386 }
 387 /**
 388  * @brief Sends a task to a mailbox
 389  *
 390  * This is a blocking function, the execution flow will be blocked until the task is sent (and received on the other
 391  * side if #MSG_task_receive is used).
 392  * See #MSG_task_isend for sending tasks asynchronously.
 393  *
 394  * @param task the task to be sent
 395  * @param alias the mailbox name to where the task is sent
 396  *
 397  * @return Returns #MSG_OK if the task was successfully sent,
 398  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
 399  */
 400 msg_error_t MSG_task_send(msg_task_t task, const char* alias)
 401 {
 402   XBT_DEBUG("MSG_task_send: Trying to send a message on mailbox '%s'", alias);
 403   return task->send(alias, -1);
 404 }
 405
 406 /**
 407  * @brief Sends a task to a mailbox with a maximum rate
 408  *
 409  * This is a blocking function, the execution flow will be blocked until the task is sent. The maxrate parameter allows
 410  * the application to limit the bandwidth utilization of network links when sending the task.
 411  *
 412  * The maxrate parameter can be used to send a task with a limited bandwidth (smaller than the physical available
 413  * value). Use MSG_task_send() if you don't limit the rate (or pass -1 as a rate value do disable this feature).
 414  *
 415  * @param task the task to be sent
 416  * @param alias the mailbox name to where the task is sent
 417  * @param maxrate the maximum communication rate for sending this task (byte/sec)
 418  *
 419  * @return Returns #MSG_OK if the task was successfully sent,
 420  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
 421  */
 422 msg_error_t MSG_task_send_bounded(msg_task_t task, const char* alias, double maxrate)
 423 {
 424   task->set_rate(maxrate);
 425   return task->send(alias, -1);
 426 }
 427
 428 /**
 429  * @brief Sends a task to a mailbox with a timeout
 430  *
 431  * This is a blocking function, the execution flow will be blocked until the task is sent or the timeout is achieved.
 432  *
 433  * @param task the task to be sent
 434  * @param alias the mailbox name to where the task is sent
 435  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_send)
 436  *
 437  * @return Returns #MSG_OK if the task was successfully sent,
 438  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
 439  */
 440 msg_error_t MSG_task_send_with_timeout(msg_task_t task, const char* alias, double timeout)
 441 {
 442   return task->send(alias, timeout);
 443 }
 444
 445 /**
 446  * @brief Sends a task to a mailbox with a timeout and with a maximum rate
 447  *
 448  * This is a blocking function, the execution flow will be blocked until the task is sent or the timeout is achieved.
 449  *
 450  * The maxrate parameter can be used to send a task with a limited bandwidth (smaller than the physical available
 451  * value). Use MSG_task_send_with_timeout() if you don't limit the rate (or pass -1 as a rate value do disable this
 452  * feature).
 453  *
 454  * @param task the task to be sent
 455  * @param alias the mailbox name to where the task is sent
 456  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_send)
 457  * @param maxrate the maximum communication rate for sending this task (byte/sec)
 458  *
 459  * @return Returns #MSG_OK if the task was successfully sent,
 460  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
 461  */
 462 msg_error_t MSG_task_send_with_timeout_bounded(msg_task_t task, const char* alias, double timeout, double maxrate)
 463 {
 464   task->set_rate(maxrate);
 465   return task->send(alias, timeout);
 466 }
 467
 468 /**
 469  * @brief Receives a task from a mailbox.
 470  *
 471  * This is a blocking function, the execution flow will be blocked until the task is received. See #MSG_task_irecv
 472  * for receiving tasks asynchronously.
 473  *
 474  * @param task a memory location for storing a #msg_task_t.
 475  * @param alias name of the mailbox to receive the task from
 476  *
 477  * @return Returns
 478  * #MSG_OK if the task was successfully received,
 479  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
 480  */
 481 msg_error_t MSG_task_receive(msg_task_t * task, const char *alias)
 482 {
 483   return MSG_task_receive_with_timeout(task, alias, -1);
 484 }
 485
 486 /**
 487  * @brief Receives a task from a mailbox at a given rate.
 488  *
 489  * @param task a memory location for storing a #msg_task_t.
 490  * @param alias name of the mailbox to receive the task from
 491  * @param rate limit the reception to rate bandwidth (byte/sec)
 492  *
 493  * The rate parameter can be used to receive a task with a limited bandwidth (smaller than the physical available
 494  * value). Use MSG_task_receive() if you don't limit the rate (or pass -1 as a rate value do disable this feature).
 495  *
 496  * @return Returns
 497  * #MSG_OK if the task was successfully received,
 498  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
 499  */
 500 msg_error_t MSG_task_receive_bounded(msg_task_t* task, const char* alias, double rate)
 501 {
 502   return MSG_task_receive_with_timeout_bounded(task, alias, -1, rate);
 503 }
 504
 505 /**
 506  * @brief Receives a task from a mailbox with a given timeout.
 507  *
 508  * This is a blocking function with a timeout, the execution flow will be blocked until the task is received or the
 509  * timeout is achieved. See #MSG_task_irecv for receiving tasks asynchronously.  You can provide a -1 timeout
 510  * to obtain an infinite timeout.
 511  *
 512  * @param task a memory location for storing a #msg_task_t.
 513  * @param alias name of the mailbox to receive the task from
 514  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_receive)
 515  *
 516  * @return Returns
 517  * #MSG_OK if the task was successfully received,
 518  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
 519  */
 520 msg_error_t MSG_task_receive_with_timeout(msg_task_t* task, const char* alias, double timeout)
 521 {
 522   return MSG_task_receive_with_timeout_bounded(task, alias, timeout, -1);
 523 }
 524
 525 /**
 526  * @brief Receives a task from a mailbox with a given timeout and at a given rate.
 527  *
 528  * @param task a memory location for storing a #msg_task_t.
 529  * @param alias name of the mailbox to receive the task from
 530  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_receive)
 531  * @param rate limit the reception to rate bandwidth (byte/sec)
 532  *
 533  * The rate parameter can be used to send a task with a limited
 534  * bandwidth (smaller than the physical available value). Use
 535  * MSG_task_receive() if you don't limit the rate (or pass -1 as a
 536  * rate value do disable this feature).
 537  *
 538  * @return Returns
 539  * #MSG_OK if the task was successfully received,
 540  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
 541  */
 542 msg_error_t MSG_task_receive_with_timeout_bounded(msg_task_t* task, const char* alias, double timeout, double rate)
 543 {
 544   XBT_DEBUG("MSG_task_receive_with_timeout_bounded: Trying to receive a message on mailbox '%s'", alias);
 545   msg_error_t ret = MSG_OK;
 546
 547   /* Sanity check */
 548   xbt_assert(task, "Null pointer for the task storage");
 549
 550   if (*task)
 551     XBT_WARN("Asked to write the received task in a non empty struct -- proceeding.");
 552
 553   /* Try to receive it by calling SIMIX network layer */
 554   try {
 555     void* payload;
 556     simgrid::s4u::Mailbox::by_name(alias)
 557         ->get_init()
 558         ->set_dst_data(&payload, sizeof(msg_task_t*))
 559         ->set_rate(rate)
 560         ->wait_for(timeout);
 561     *task = static_cast<msg_task_t>(payload);
 562     XBT_DEBUG("Got task %s from %s", (*task)->get_cname(), alias);
 563     (*task)->set_not_used();
 564   } catch (const simgrid::HostFailureException&) {
 565     ret = MSG_HOST_FAILURE;
 566   } catch (const simgrid::TimeoutException&) {
 567     ret = MSG_TIMEOUT;
 568   } catch (const simgrid::CancelException&) {
 569     ret = MSG_TASK_CANCELED;
 570   } catch (const simgrid::NetworkFailureException&) {
 571     ret = MSG_TRANSFER_FAILURE;
 572   }
 573
 574   if (TRACE_actor_is_enabled() && ret != MSG_HOST_FAILURE && ret != MSG_TRANSFER_FAILURE && ret != MSG_TIMEOUT) {
 575     auto* process_container = simgrid::instr::Container::by_name(instr_pid(*MSG_process_self()));
 576
 577     std::string key = "p" + std::to_string((*task)->get_id());
 578     simgrid::instr::Container::get_root()->get_link("ACTOR_LINK")->end_event(process_container, "SR", key);
 579   }
 580   return ret;
 581 }
 582
 583 /**
 584  * @brief Starts listening for receiving a task from an asynchronous communication.
 585  *
 586  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication.
 587  *
 588  * @param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
 589  * @param name of the mailbox to receive the task on
 590  * @return the msg_comm_t communication created
 591  */
 592 msg_comm_t MSG_task_irecv(msg_task_t* task, const char* name)
 593 {
 594   return MSG_task_irecv_bounded(task, name, -1.0);
 595 }
 596
 597 /**
 598  * @brief Starts listening for receiving a task from an asynchronous communication at a given rate.
 599  *
 600  * The rate parameter can be used to receive a task with a limited
 601  * bandwidth (smaller than the physical available value). Use
 602  * MSG_task_irecv() if you don't limit the rate (or pass -1 as a rate
 603  * value do disable this feature).
 604  *
 605  * @param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
 606  * @param name of the mailbox to receive the task on
 607  * @param rate limit the bandwidth to the given rate (byte/sec)
 608  * @return the msg_comm_t communication created
 609  */
 610 msg_comm_t MSG_task_irecv_bounded(msg_task_t* task, const char* name, double rate)
 611 {
 612   /* FIXME: these functions are not traceable */
 613   /* Sanity check */
 614   xbt_assert(task, "Null pointer for the task storage");
 615
 616   if (*task)
 617     XBT_CRITICAL("MSG_task_irecv() was asked to write in a non empty task struct.");
 618
 619   /* Try to receive it by calling SIMIX network layer */
 620   simgrid::s4u::CommPtr comm = simgrid::s4u::Mailbox::by_name(name)
 621                                    ->get_init()
 622                                    ->set_dst_data((void**)task, sizeof(msg_task_t*))
 623                                    ->set_rate(rate)
 624                                    ->start();
 625
 626   return new simgrid::msg::Comm(nullptr, task, comm);
 627 }
 628
 629 /**
 630  * @brief Look if there is a communication on a mailbox and return the PID of the sender process.
 631  *
 632  * @param alias the name of the mailbox to be considered
 633  *
 634  * @return Returns the PID of sender process (or -1 if there is no communication in the mailbox)
 635  *
 636  */
 637 int MSG_task_listen_from(const char* alias)
 638 {
 639   return simgrid::s4u::Mailbox::by_name(alias)->listen_from();
 640 }
 641
 642 /** @brief Destroys the given task.
 643  *
 644  * You should free user data, if any, @b before calling this destructor.
 645  *
 646  * Only the process that owns the task can destroy it.
 647  * The owner changes after a successful send.
 648  * If a task is successfully sent, the receiver becomes the owner and is supposed to destroy it. The sender should not
 649  * use it anymore.
 650  * If the task failed to be sent, the sender remains the owner of the task.
 651  */
 652 msg_error_t MSG_task_destroy(msg_task_t task)
 653 {
 654   if (task->is_used()) {
 655     /* the task is being sent or executed: cancel it first */
 656     task->cancel();
 657   }
 658
 659   /* free main structures */
 660   delete task;
 661
 662   return MSG_OK;
 663 }
 664
 665 /** @brief Cancel the given task
 666  *
 667  * If it was currently executed or transferred, the working process is stopped.
 668  */
 669 msg_error_t MSG_task_cancel(msg_task_t task)
 670 {
 671   xbt_assert((task != nullptr), "Cannot cancel a nullptr task");
 672   task->cancel();
 673   return MSG_OK;
 674 }
 675
 676 /** @brief Returns a value in ]0,1[ that represent the task remaining work
 677  *    to do: starts at 1 and goes to 0. Returns 0 if not started or finished.
 678  *
 679  * It works for either parallel or sequential tasks.
 680  */
 681 double MSG_task_get_remaining_work_ratio(const_msg_task_t task)
 682 {
 683   xbt_assert((task != nullptr), "Cannot get information from a nullptr task");
 684   if (task->compute) {
 685     // Task in progress
 686     return task->compute->get_remaining_ratio();
 687   } else {
 688     // Task not started (flops_amount is > 0.0) or finished (flops_amount is set to 0.0)
 689     return task->flops_amount > 0.0 ? 1.0 : 0.0;
 690   }
 691 }
 692
 693 /** @brief Returns the amount of flops that remain to be computed
 694  *
 695  * The returned value is initially the cost that you defined for the task, then it decreases until it reaches 0
 696  *
 697  * It works for sequential tasks, but the remaining amount of work is not a scalar value for parallel tasks.
 698  * So you will get an exception if you call this function on parallel tasks. Just don't do it.
 699  */
 700 double MSG_task_get_flops_amount(const_msg_task_t task)
 701 {
 702   if (task->compute != nullptr && task->compute->get_state() == simgrid::s4u::Activity::State::STARTED) {
 703     return task->compute->get_remaining();
 704   } else {
 705     // Not started or already done.
 706     // - Before starting, flops_amount is initially the task cost
 707     // - After execution, flops_amount is set to 0 (until someone uses MSG_task_set_flops_amount, if any)
 708     return task->flops_amount;
 709   }
 710 }
 711
 712 /** @brief set the computation amount needed to process the given task.
 713  *
 714  * @warning If the computation is ongoing (already started and not finished),
 715  * it is not modified by this call. Moreover, after its completion, the ongoing execution with set the flops_amount to
 716  * zero, overriding any value set during the execution.
 717  */
 718 void MSG_task_set_flops_amount(msg_task_t task, double flops_amount)
 719 {
 720   task->flops_amount = flops_amount;
 721 }
 722
 723 /** @brief set the amount data attached with the given task.
 724  *
 725  * @warning If the transfer is ongoing (already started and not finished), it is not modified by this call.
 726  */
 727 void MSG_task_set_bytes_amount(msg_task_t task, double data_size)
 728 {
 729   task->bytes_amount = data_size;
 730 }
 731
 732 /** @brief Returns the total amount received by the given task
 733  *
 734  *  If the communication does not exist it will return 0.
 735  *  So, if the communication has FINISHED or FAILED it returns zero.
 736  */
 737 double MSG_task_get_remaining_communication(const_msg_task_t task)
 738 {
 739   XBT_DEBUG("calling s4u::Comm::get_remaining (%p)", task->comm.get());
 740   return task->comm->get_remaining();
 741 }
 742
 743 /** @brief Returns the size of the data attached to the given task. */
 744 double MSG_task_get_bytes_amount(const_msg_task_t task)
 745 {
 746   xbt_assert(task != nullptr, "Invalid parameter");
 747   return task->bytes_amount;
 748 }
 749
 750 /** @brief Changes the priority of a computation task.
 751  *
 752  * This priority doesn't affect the transfer rate. A priority of 2
 753  * will make a task receive two times more cpu power than regular tasks.
 754  */
 755 void MSG_task_set_priority(msg_task_t task, double priority)
 756 {
 757   task->set_priority(priority);
 758 }
 759
 760 /** @brief Changes the maximum CPU utilization of a computation task (in flops/s).
 761  *
 762  * For VMs, there is a pitfall. Please see MSG_vm_set_bound().
 763  */
 764 void MSG_task_set_bound(msg_task_t task, double bound)
 765 {
 766   if (bound < 1e-12) /* close enough to 0 without any floating precision surprise */
 767     XBT_INFO("bound == 0 means no capping (i.e., unlimited).");
 768   task->set_bound(bound);
 769 }
 770
 771 /**
 772  * @brief Sets the tracing category of a task.
 773  *
 774  * This function should be called after the creation of a MSG task, to define the category of that task. The
 775  * first parameter task must contain a task that was =created with the function MSG_task_create(). The second
 776  * parameter category must contain a category that was previously declared with the function #TRACE_category
 777  * (or with #TRACE_category_with_color).
 778  *
 779  * @beginrst
 780  * See :ref:`outcome_vizu` for details on how to trace the (categorized) resource utilization.
 781  * @endrst
 782  *
 783  * @param task the task that is going to be categorized
 784  * @param category the name of the category to be associated to the task
 785  */
 786 void MSG_task_set_category(msg_task_t task, const char* category)
 787 {
 788   xbt_assert(not task->has_tracing_category(), "Task %p(%s) already has a category (%s).", task, task->get_cname(),
 789              task->get_tracing_category().c_str());
 790
 791   // if user provides a nullptr category, task is no longer traced
 792   if (category == nullptr) {
 793     task->set_tracing_category("");
 794     XBT_DEBUG("MSG task %p(%s), category removed", task, task->get_cname());
 795   } else {
 796     // set task category
 797     task->set_tracing_category(category);
 798     XBT_DEBUG("MSG task %p(%s), category %s", task, task->get_cname(), task->get_tracing_category().c_str());
 799   }
 800 }
 801
 802 /**
 803  * @brief Gets the current tracing category of a task. (@see MSG_task_set_category)
 804  * @param task the task to be considered
 805  * @return Returns the name of the tracing category of the given task, "" otherwise
 806  */
 807 const char* MSG_task_get_category(const_msg_task_t task)
 808 {
 809   return task->get_tracing_category().c_str();
 810 }