src/msg/msg_task.cpp

   1 /* Copyright (c) 2004-2015. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include "msg_private.h"
   8 #include "src/simix/smx_private.h"
   9 #include "xbt/sysdep.h"
  10 #include "xbt/log.h"
  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 /********************************* Task **************************************/
  23 /** \ingroup m_task_management
  24  * \brief Creates a new #msg_task_t.
  25  *
  26  * A constructor for #msg_task_t taking four arguments and returning the corresponding object.
  27  * \param name a name for the object. It is for user-level information and can be NULL.
  28  * \param flop_amount a value of the processing amount (in flop) needed to process this new task.
  29  * If 0, then it cannot be executed with MSG_task_execute(). This value has to be >=0.
  30  * \param message_size a value of the amount of data (in bytes) needed to transfer this new task. If 0, then it cannot
  31  * be transfered with MSG_task_send() and MSG_task_recv(). This value has to be >=0.
  32  * \param data a pointer to any data may want to attach to the new object.  It is for user-level information and can
  33  * be NULL. It can be retrieved with the function \ref MSG_task_get_data.
  34  * \see msg_task_t
  35  * \return The new corresponding object.
  36  */
  37 msg_task_t MSG_task_create(const char *name, double flop_amount, double message_size, void *data)
  38 {
  39   msg_task_t task = xbt_new(s_msg_task_t, 1);
  40   simdata_task_t simdata = xbt_new(s_simdata_task_t, 1);
  41   task->simdata = simdata;
  42
  43   /* Task structure */
  44   task->name = xbt_strdup(name);
  45   task->data = data;
  46
  47   /* Simulator Data */
  48   simdata->compute = NULL;
  49   simdata->comm = NULL;
  50   simdata->bytes_amount = message_size;
  51   simdata->flops_amount = flop_amount;
  52   simdata->sender = NULL;
  53   simdata->receiver = NULL;
  54   simdata->source = NULL;
  55   simdata->priority = 1.0;
  56   simdata->bound = 0;
  57   simdata->affinity_mask_db = xbt_dict_new_homogeneous(NULL);
  58   simdata->rate = -1.0;
  59   simdata->isused = 0;
  60
  61   simdata->host_nb = 0;
  62   simdata->host_list = NULL;
  63   simdata->flops_parallel_amount = NULL;
  64   simdata->bytes_parallel_amount = NULL;
  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 NULL.
  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 NULL.
  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   msg_task_t task = MSG_task_create(name, 0, 0, data);
  90   simdata_task_t simdata = task->simdata;
  91   int i;
  92
  93   /* Simulator Data specific to parallel tasks */
  94   simdata->host_nb = host_nb;
  95   simdata->host_list = xbt_new0(sg_host_t, host_nb);
  96   simdata->flops_parallel_amount = flops_amount;
  97   simdata->bytes_parallel_amount = bytes_amount;
  98
  99   for (i = 0; i < host_nb; i++)
 100     simdata->host_list[i] = host_list[i];
 101
 102   return task;
 103 }
 104
 105 /** \ingroup m_task_management
 106  * \brief Return the user data of a #msg_task_t.
 107  *
 108  * This function checks whether \a task is a valid pointer and return the user data associated to \a task if possible.
 109  */
 110 void *MSG_task_get_data(msg_task_t task)
 111 {
 112   xbt_assert((task != NULL), "Invalid parameter");
 113   return (task->data);
 114 }
 115
 116 /** \ingroup m_task_management
 117  * \brief Sets the user data of a #msg_task_t.
 118  *
 119  * This function allows to associate a new pointer to the user data associated of \a task.
 120  */
 121 void MSG_task_set_data(msg_task_t task, void *data)
 122 {
 123   xbt_assert((task != NULL), "Invalid parameter");
 124   task->data = data;
 125 }
 126
 127 /** \ingroup m_task_management
 128  * \brief Sets a function to be called when a task has just been copied.
 129  * \param callback a callback function
 130  */
 131 void MSG_task_set_copy_callback(void (*callback) (msg_task_t task, msg_process_t sender, msg_process_t receiver)) {
 132
 133   msg_global->task_copy_callback = callback;
 134
 135   if (callback) {
 136     SIMIX_comm_set_copy_data_callback(MSG_comm_copy_data_from_SIMIX);
 137   } else {
 138     SIMIX_comm_set_copy_data_callback(SIMIX_comm_copy_pointer_callback);
 139   }
 140 }
 141
 142 /** \ingroup m_task_management
 143  * \brief Return the sender of a #msg_task_t.
 144  *
 145  * This functions returns the #msg_process_t which sent this task
 146  */
 147 msg_process_t MSG_task_get_sender(msg_task_t task)
 148 {
 149   xbt_assert(task, "Invalid parameters");
 150   return ((simdata_task_t) task->simdata)->sender;
 151 }
 152
 153 /** \ingroup m_task_management
 154  * \brief Return the source of a #msg_task_t.
 155  *
 156  * This functions returns the #msg_host_t from which this task was sent
 157  */
 158 msg_host_t MSG_task_get_source(msg_task_t task)
 159 {
 160   xbt_assert(task, "Invalid parameters");
 161   return ((simdata_task_t) task->simdata)->source;
 162 }
 163
 164 /** \ingroup m_task_management
 165  * \brief Return the name of a #msg_task_t.
 166  *
 167  * This functions returns the name of a #msg_task_t as specified on creation
 168  */
 169 const char *MSG_task_get_name(msg_task_t task)
 170 {
 171   xbt_assert(task, "Invalid parameters");
 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   xbt_assert(task, "Invalid parameters");
 183   task->name = xbt_strdup(name);
 184 }
 185
 186 /** \ingroup m_task_management
 187  * \brief Destroy a #msg_task_t.
 188  *
 189  * Destructor for #msg_task_t. Note that you should free user data, if any, \b before calling this function.
 190  *
 191  * Only the process that owns the task can destroy it.
 192  * The owner changes after a successful send.
 193  * If a task is successfully sent, the receiver becomes the owner and is supposed to destroy it. The sender should not
 194  * use it anymore.
 195  * If the task failed to be sent, the sender remains the owner of the task.
 196  */
 197 msg_error_t MSG_task_destroy(msg_task_t task)
 198 {
 199   smx_synchro_t action = NULL;
 200   xbt_assert((task != NULL), "Invalid parameter");
 201
 202   if (task->simdata->isused) {
 203     /* the task is being sent or executed: cancel it first */
 204     MSG_task_cancel(task);
 205   }
 206   TRACE_msg_task_destroy(task);
 207
 208   xbt_free(task->name);
 209
 210   action = task->simdata->compute;
 211   if (action)
 212     simcall_execution_destroy(action);
 213
 214   /* parallel tasks only */
 215   xbt_free(task->simdata->host_list);
 216
 217   xbt_dict_free(&task->simdata->affinity_mask_db);
 218
 219   /* free main structures */
 220   xbt_free(task->simdata);
 221   xbt_free(task);
 222
 223   return MSG_OK;
 224 }
 225
 226 /** \ingroup m_task_usage
 227  * \brief Cancel a #msg_task_t.
 228  * \param task the task to cancel. If it was executed or transfered, it stops the process that were working on it.
 229  */
 230 msg_error_t MSG_task_cancel(msg_task_t task)
 231 {
 232   xbt_assert((task != NULL), "Cannot cancel a NULL task");
 233
 234   if (task->simdata->compute) {
 235     simcall_execution_cancel(task->simdata->compute);
 236   }
 237   else if (task->simdata->comm) {
 238     simdata_task_t simdata = task->simdata;
 239     simcall_comm_cancel(simdata->comm);
 240     if (msg_global->debug_multiple_use && simdata->isused!=0)
 241       xbt_ex_free(*(xbt_ex_t*)simdata->isused);
 242     simdata->isused = 0;
 243   }
 244   return MSG_OK;
 245 }
 246
 247 /** \ingroup m_task_management
 248  * \brief Returns the remaining amount of flops needed to execute a task #msg_task_t.
 249  *
 250  * Once a task has been processed, this amount is set to 0. If you want, you can reset this value with
 251  * #MSG_task_set_flops_amount before restarting the task.
 252  */
 253 double MSG_task_get_flops_amount(msg_task_t task) {
 254   if (task->simdata->compute) {
 255     return simcall_execution_get_remains(task->simdata->compute);
 256   } else {
 257     return task->simdata->flops_amount;
 258   }
 259 }
 260
 261 /** \ingroup m_task_management
 262  * \brief set the computation amount needed to process a task #msg_task_t.
 263  *
 264  * \warning If the computation is ongoing (already started and not finished),
 265  * it is not modified by this call. Moreover, after its completion, the ongoing execution with set the flops_amount to
 266  * zero, overriding any value set during the execution.
 267  */
 268 void MSG_task_set_flops_amount(msg_task_t task, double flops_amount)
 269 {
 270   task->simdata->flops_amount = flops_amount;
 271 }
 272
 273 /** \ingroup m_task_management
 274  * \brief set the amount data attached with a task #msg_task_t.
 275  *
 276  * \warning If the transfer is ongoing (already started and not finished), it is not modified by this call.
 277  */
 278 void MSG_task_set_bytes_amount(msg_task_t task, double data_size)
 279 {
 280   task->simdata->bytes_amount = data_size;
 281 }
 282
 283 /** \ingroup m_task_management
 284  * \brief Returns the total amount received by a task #msg_task_t.
 285  *        If the communication does not exist it will return 0.
 286  *        So, if the communication has FINISHED or FAILED it returns zero.
 287  */
 288 double MSG_task_get_remaining_communication(msg_task_t task)
 289 {
 290   xbt_assert((task != NULL) && (task->simdata != NULL), "Invalid parameter");
 291   XBT_DEBUG("calling simcall_communication_get_remains(%p)", task->simdata->comm);
 292   return simcall_comm_get_remains(task->simdata->comm);
 293 }
 294
 295 /** \ingroup m_task_management
 296  * \brief Returns the size of the data attached to a task #msg_task_t.
 297  */
 298 double MSG_task_get_bytes_amount(msg_task_t task)
 299 {
 300   xbt_assert((task != NULL) && (task->simdata != NULL), "Invalid parameter");
 301   return task->simdata->bytes_amount;
 302 }
 303
 304 /** \ingroup m_task_management
 305  * \brief Changes the priority of a computation task. This priority doesn't affect the transfer rate. A priority of 2
 306  *        will make a task receive two times more cpu power than the other ones.
 307  */
 308 void MSG_task_set_priority(msg_task_t task, double priority)
 309 {
 310   xbt_assert((task != NULL) && (task->simdata != NULL), "Invalid parameter");
 311   task->simdata->priority = 1 / priority;
 312   if (task->simdata->compute)
 313     simcall_execution_set_priority(task->simdata->compute,
 314         task->simdata->priority);
 315 }
 316
 317 /** \ingroup m_task_management
 318  * \brief Changes the maximum CPU utilization of a computation task.
 319  *        Unit is flops/s.
 320  *
 321  * For VMs, there is a pitfall. Please see MSG_vm_set_bound().
 322  */
 323 void MSG_task_set_bound(msg_task_t task, double bound)
 324 {
 325   xbt_assert(task, "Invalid parameter");
 326   xbt_assert(task->simdata, "Invalid parameter");
 327
 328   if (bound == 0)
 329     XBT_INFO("bound == 0 means no capping (i.e., unlimited).");
 330
 331   task->simdata->bound = bound;
 332   if (task->simdata->compute)
 333     simcall_execution_set_bound(task->simdata->compute, task->simdata->bound);
 334 }
 335
 336 /** \ingroup m_task_management
 337  * \brief Changes the CPU affinity of a computation task.
 338  *
 339  * When pinning the given task to the first CPU core of the given host, use 0x01 for the mask value. Each bit of the
 340  * mask value corresponds to each CPU core. See taskset(1) on Linux.
 341  *
 342  * \param task a target task
 343  * \param host the host having a multi-core CPU
 344  * \param mask the bit mask of a new CPU affinity setting for the task
 345  *
 346  * Usage:
 347  * 0. Define a host with multiple cores.
 348  *    \<host id="PM0" power="1E8" core="2"/\>
 349  *
 350  * 1. Pin a given task to the first CPU core of a host.
 351  *   MSG_task_set_affinity(task, pm0, 0x01);
 352  *
 353  * 2. Pin a given task to the third CPU core of a host. Turn on the third bit of the mask.
 354  *   MSG_task_set_affinity(task, pm0, 0x04); // 0x04 == 100B
 355  *
 356  * 3. Pin a given VM to the first CPU core of a host.
 357  *   MSG_vm_set_affinity(vm, pm0, 0x01);
 358  *
 359  * See examples/msg/cloud/multicore.c for more information.
 360  *
 361  * Note:
 362  * 1. The current code does not allow an affinity of a task to multiple cores.
 363  *    The mask value 0x03 (i.e., a given task will be executed on the first core or the second core) is not allowed.
 364  *    The mask value 0x01 or 0x02 works. See cpu_cas01.c for details.
 365  *
 366  * 2. It is recommended to first compare simulation results in both the Lazy and Full calculation modes
 367  *    (using --cfg=cpu/optim:Full or not). Fix cpu_cas01.c if you find wrong results in the Lazy mode.
 368  */
 369 void MSG_task_set_affinity(msg_task_t task, msg_host_t host, unsigned long mask)
 370 {
 371   xbt_assert(task, "Invalid parameter");
 372   xbt_assert(task->simdata, "Invalid parameter");
 373
 374   if (mask == 0) {
 375     /* 0 means clear */
 376       /* We need remove_ext() not throwing exception. */
 377       void *ret = xbt_dict_get_or_null_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(msg_host_t));
 378       if (ret != NULL)
 379         xbt_dict_remove_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(host));
 380   } else
 381     xbt_dict_set_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(host), (void *)(uintptr_t) mask, NULL);
 382
 383   /* We set affinity data of this task. If the task is being executed, we actually change the affinity setting of the
 384    * task. Otherwise, this change will be applied when the task is executed. */
 385   if (!task->simdata->compute) {
 386     /* task is not yet executed */
 387     XBT_INFO("set affinity(0x%04lx@%s) for %s (not active now)", mask, MSG_host_get_name(host),
 388              MSG_task_get_name(task));
 389     return;
 390   }
 391
 392   {
 393     simgrid::simix::Exec *compute = task->simdata->compute;
 394     msg_host_t host_now = compute->host;  // simix_private.h is necessary
 395     if (host_now != host) {
 396       /* task is not yet executed on this host */
 397       XBT_INFO("set affinity(0x%04lx@%s) for %s (not active now)", mask, MSG_host_get_name(host),
 398                MSG_task_get_name(task));
 399       return;
 400     }
 401
 402     /* task is being executed on this host. so change the affinity now */
 403     {
 404       /* check it works. remove me if it works. */
 405       xbt_assert((unsigned long)(uintptr_t) xbt_dict_get_or_null_ext(task->simdata->affinity_mask_db,
 406                  (char *) host, sizeof(msg_host_t)) == mask);
 407     }
 408
 409     XBT_INFO("set affinity(0x%04lx@%s) for %s", mask, MSG_host_get_name(host), MSG_task_get_name(task));
 410     simcall_execution_set_affinity(task->simdata->compute, host, mask);
 411   }
 412 }