src/msg/msg_task.cpp

   1 /* Copyright (c) 2004-2017. 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/simix/smx_private.hpp"
   8 #include <algorithm>
   9
  10 extern "C" {
  11
  12 /** @addtogroup m_task_management
  13  *
  14  *  Since most scheduling algorithms rely on a concept of task  that can be either <em>computed</em> locally or
  15  *  <em>transferred</em> on another processor, it seems to be the right level of abstraction for our purposes.
  16  *  A <em>task</em> may then be defined by a <em>computing amount</em>, a <em>message size</em> and
  17  *  some <em>private data</em>.
  18  */
  19
  20 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_task, msg, "Logging specific to MSG (task)");
  21
  22 void s_simdata_task_t::reportMultipleUse() const
  23 {
  24   if (msg_global->debug_multiple_use){
  25     XBT_ERROR("This task is already used in there:");
  26     // TODO, backtrace
  27     XBT_ERROR("<missing backtrace>");
  28     XBT_ERROR("And you try to reuse it from here:");
  29     xbt_backtrace_display_current();
  30   } else {
  31     xbt_die("This task is still being used somewhere else. You cannot send it now. Go fix your code!"
  32              "(use --cfg=msg/debug-multiple-use:on to get the backtrace of the other process)");
  33   }
  34 }
  35
  36 /********************************* Task **************************************/
  37 /** \ingroup m_task_management
  38  * \brief Creates a new #msg_task_t.
  39  *
  40  * A constructor for #msg_task_t taking four arguments and returning the corresponding object.
  41  * \param name a name for the object. It is for user-level information and can be nullptr.
  42  * \param flop_amount a value of the processing amount (in flop) needed to process this new task.
  43  * If 0, then it cannot be executed with MSG_task_execute(). This value has to be >=0.
  44  * \param message_size a value of the amount of data (in bytes) needed to transfer this new task. If 0, then it cannot
  45  * be transfered with MSG_task_send() and MSG_task_recv(). This value has to be >=0.
  46  * \param data a pointer to any data may want to attach to the new object.  It is for user-level information and can
  47  * be nullptr. It can be retrieved with the function \ref MSG_task_get_data.
  48  * \see msg_task_t
  49  * \return The new corresponding object.
  50  */
  51 msg_task_t MSG_task_create(const char *name, double flop_amount, double message_size, void *data)
  52 {
  53   msg_task_t task        = new s_msg_task_t;
  54   simdata_task_t simdata = new s_simdata_task_t();
  55   task->simdata = simdata;
  56
  57   /* Task structure */
  58   task->name = xbt_strdup(name);
  59   task->data = data;
  60
  61   /* Simulator Data */
  62   simdata->bytes_amount = message_size;
  63   simdata->flops_amount = flop_amount;
  64
  65   TRACE_msg_task_create(task);
  66
  67   return task;
  68 }
  69
  70 /** \ingroup m_task_management
  71  * \brief Creates a new #msg_task_t (a parallel one....).
  72  *
  73  * A constructor for #msg_task_t taking six arguments and returning the corresponding object.
  74  * \param name a name for the object. It is for user-level information and can be nullptr.
  75  * \param host_nb the number of hosts implied in the parallel task.
  76  * \param host_list an array of \p host_nb msg_host_t.
  77  * \param flops_amount an array of \p host_nb doubles.
  78  *        flops_amount[i] is the total number of operations that have to be performed on host_list[i].
  79  * \param bytes_amount an array of \p host_nb* \p host_nb doubles.
  80  * \param data a pointer to any data may want to attach to the new object.
  81  *             It is for user-level information and can be nullptr.
  82  *             It can be retrieved with the function \ref MSG_task_get_data.
  83  * \see msg_task_t
  84  * \return The new corresponding object.
  85  */
  86 msg_task_t MSG_parallel_task_create(const char *name, int host_nb, const msg_host_t * host_list,
  87                                     double *flops_amount, double *bytes_amount, void *data)
  88 {
  89   // Task's flops amount is set to an arbitrary value > 0.0 to be able to distinguish, in
  90   // MSG_task_get_remaining_work_ratio(), a finished task and a task that has not started yet.
  91   msg_task_t task        = MSG_task_create(name, 1.0, 0, data);
  92   simdata_task_t simdata = task->simdata;
  93
  94   /* Simulator Data specific to parallel tasks */
  95   simdata->host_nb = host_nb;
  96   simdata->host_list             = new sg_host_t[host_nb];
  97   std::copy_n(host_list, host_nb, simdata->host_list);
  98   if (flops_amount != nullptr) {
  99     simdata->flops_parallel_amount = new double[host_nb];
 100     std::copy_n(flops_amount, host_nb, simdata->flops_parallel_amount);
 101   }
 102   if (bytes_amount != nullptr) {
 103     simdata->bytes_parallel_amount = new double[host_nb * host_nb];
 104     std::copy_n(bytes_amount, host_nb * host_nb, simdata->bytes_parallel_amount);
 105   }
 106
 107   return task;
 108 }
 109
 110 /** \ingroup m_task_management
 111  * \brief Return the user data of a #msg_task_t.
 112  *
 113  * This function checks whether \a task is a valid pointer and return the user data associated to \a task if possible.
 114  */
 115 void *MSG_task_get_data(msg_task_t task)
 116 {
 117   return (task->data);
 118 }
 119
 120 /** \ingroup m_task_management
 121  * \brief Sets the user data of a #msg_task_t.
 122  *
 123  * This function allows to associate a new pointer to the user data associated of \a task.
 124  */
 125 void MSG_task_set_data(msg_task_t task, void *data)
 126 {
 127   task->data = data;
 128 }
 129
 130 /** \ingroup m_task_management
 131  * \brief Sets a function to be called when a task has just been copied.
 132  * \param callback a callback function
 133  */
 134 void MSG_task_set_copy_callback(void (*callback) (msg_task_t task, msg_process_t sender, msg_process_t receiver)) {
 135
 136   msg_global->task_copy_callback = callback;
 137
 138   if (callback) {
 139     SIMIX_comm_set_copy_data_callback(MSG_comm_copy_data_from_SIMIX);
 140   } else {
 141     SIMIX_comm_set_copy_data_callback(SIMIX_comm_copy_pointer_callback);
 142   }
 143 }
 144
 145 /** \ingroup m_task_management
 146  * \brief Return the sender of a #msg_task_t.
 147  *
 148  * This functions returns the #msg_process_t which sent this task
 149  */
 150 msg_process_t MSG_task_get_sender(msg_task_t task)
 151 {
 152   return task->simdata->sender;
 153 }
 154
 155 /** \ingroup m_task_management
 156  * \brief Return the source of a #msg_task_t.
 157  *
 158  * This functions returns the #msg_host_t from which this task was sent
 159  */
 160 msg_host_t MSG_task_get_source(msg_task_t task)
 161 {
 162   return task->simdata->source;
 163 }
 164
 165 /** \ingroup m_task_management
 166  * \brief Return the name of a #msg_task_t.
 167  *
 168  * This functions returns the name of a #msg_task_t as specified on creation
 169  */
 170 const char *MSG_task_get_name(msg_task_t task)
 171 {
 172   return task->name;
 173 }
 174
 175 /** \ingroup m_task_management
 176  * \brief Sets the name of a #msg_task_t.
 177  *
 178  * This functions allows to associate a name to a task
 179  */
 180 void MSG_task_set_name(msg_task_t task, const char *name)
 181 {
 182   task->name = xbt_strdup(name);
 183 }
 184
 185 /** \ingroup m_task_management
 186  * \brief Destroy a #msg_task_t.
 187  *
 188  * Destructor for #msg_task_t. Note that you should free user data, if any, \b before calling this function.
 189  *
 190  * Only the process that owns the task can destroy it.
 191  * The owner changes after a successful send.
 192  * If a task is successfully sent, the receiver becomes the owner and is supposed to destroy it. The sender should not
 193  * use it anymore.
 194  * If the task failed to be sent, the sender remains the owner of the task.
 195  */
 196 msg_error_t MSG_task_destroy(msg_task_t task)
 197 {
 198   if (task->simdata->isused) {
 199     /* the task is being sent or executed: cancel it first */
 200     MSG_task_cancel(task);
 201   }
 202   TRACE_msg_task_destroy(task);
 203
 204   xbt_free(task->name);
 205
 206   /* free main structures */
 207   delete task->simdata;
 208   delete task;
 209
 210   return MSG_OK;
 211 }
 212
 213 /** \ingroup m_task_usage
 214  * \brief Cancel a #msg_task_t.
 215  * \param task the task to cancel. If it was executed or transfered, it stops the process that were working on it.
 216  */
 217 msg_error_t MSG_task_cancel(msg_task_t task)
 218 {
 219   xbt_assert((task != nullptr), "Cannot cancel a nullptr task");
 220
 221   simdata_task_t simdata = task->simdata;
 222   if (simdata->compute) {
 223     simcall_execution_cancel(simdata->compute);
 224   } else if (simdata->comm) {
 225     simcall_comm_cancel(simdata->comm);
 226   }
 227   simdata->setNotUsed();
 228   return MSG_OK;
 229 }
 230
 231 /** \ingroup m_task_management
 232  * \brief Returns a value in ]0,1[ that represent the task remaining work
 233  *    to do: starts at 1 and goes to 0. Returns 0 if not started or finished.
 234  *
 235  * It works for either parallel or sequential tasks.
 236  */
 237 double MSG_task_get_remaining_work_ratio(msg_task_t task) {
 238
 239   xbt_assert((task != nullptr), "Cannot get information from a nullptr task");
 240   if (task->simdata->compute) {
 241     // Task in progress
 242     return task->simdata->compute->remainingRatio();
 243   } else {
 244     // Task not started (flops_amount is > 0.0) or finished (flops_amount is set to 0.0)
 245     return task->simdata->flops_amount > 0.0 ? 1.0 : 0.0;
 246   }
 247 }
 248
 249 /** \ingroup m_task_management
 250  * \brief Returns the amount of flops that remain to be computed
 251  *
 252  * The returned value is initially the cost that you defined for the task, then it decreases until it reaches 0
 253  *
 254  * It works for sequential tasks, but the remaining amount of work is not a scalar value for parallel tasks.
 255  * So you will get an exception if you call this function on parallel tasks. Just don't do it.
 256  */
 257 double MSG_task_get_flops_amount(msg_task_t task) {
 258   if (task->simdata->compute != nullptr) {
 259     return task->simdata->compute->remains();
 260   } else {
 261     // Not started or already done.
 262     // - Before starting, flops_amount is initially the task cost
 263     // - After execution, flops_amount is set to 0 (until someone uses MSG_task_set_flops_amount, if any)
 264     return task->simdata->flops_amount;
 265   }
 266 }
 267
 268 /** \ingroup m_task_management
 269  * \brief set the computation amount needed to process a task #msg_task_t.
 270  *
 271  * \warning If the computation is ongoing (already started and not finished),
 272  * it is not modified by this call. Moreover, after its completion, the ongoing execution with set the flops_amount to
 273  * zero, overriding any value set during the execution.
 274  */
 275 void MSG_task_set_flops_amount(msg_task_t task, double flops_amount)
 276 {
 277   task->simdata->flops_amount = flops_amount;
 278 }
 279
 280 /** \ingroup m_task_management
 281  * \brief set the amount data attached with a task #msg_task_t.
 282  *
 283  * \warning If the transfer is ongoing (already started and not finished), it is not modified by this call.
 284  */
 285 void MSG_task_set_bytes_amount(msg_task_t task, double data_size)
 286 {
 287   task->simdata->bytes_amount = data_size;
 288 }
 289
 290 /** \ingroup m_task_management
 291  * \brief Returns the total amount received by a task #msg_task_t.
 292  *        If the communication does not exist it will return 0.
 293  *        So, if the communication has FINISHED or FAILED it returns zero.
 294  */
 295 double MSG_task_get_remaining_communication(msg_task_t task)
 296 {
 297   XBT_DEBUG("calling simcall_communication_get_remains(%p)", task->simdata->comm.get());
 298   return task->simdata->comm->remains();
 299 }
 300
 301 /** \ingroup m_task_management
 302  * \brief Returns the size of the data attached to a task #msg_task_t.
 303  */
 304 double MSG_task_get_bytes_amount(msg_task_t task)
 305 {
 306   xbt_assert((task != nullptr) && (task->simdata != nullptr), "Invalid parameter");
 307   return task->simdata->bytes_amount;
 308 }
 309
 310 /** \ingroup m_task_management
 311  * \brief Changes the priority of a computation task. This priority doesn't affect the transfer rate. A priority of 2
 312  *        will make a task receive two times more cpu power than the other ones.
 313  */
 314 void MSG_task_set_priority(msg_task_t task, double priority)
 315 {
 316   task->simdata->priority = 1 / priority;
 317   if (task->simdata->compute)
 318     simcall_execution_set_priority(task->simdata->compute, task->simdata->priority);
 319 }
 320
 321 /** \ingroup m_task_management
 322  * \brief Changes the maximum CPU utilization of a computation task.
 323  *        Unit is flops/s.
 324  *
 325  * For VMs, there is a pitfall. Please see MSG_vm_set_bound().
 326  */
 327 void MSG_task_set_bound(msg_task_t task, double bound)
 328 {
 329   if (bound < 1e-12) /* close enough to 0 without any floating precision surprise */
 330     XBT_INFO("bound == 0 means no capping (i.e., unlimited).");
 331
 332   task->simdata->bound = bound;
 333   if (task->simdata->compute)
 334     simcall_execution_set_bound(task->simdata->compute, task->simdata->bound);
 335 }
 336 }