src/msg/msg_task.cpp

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