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   xbt_assert((task != NULL), "Invalid parameter");
 200
 201   if (task->simdata->isused) {
 202     /* the task is being sent or executed: cancel it first */
 203     MSG_task_cancel(task);
 204   }
 205   TRACE_msg_task_destroy(task);
 206
 207   xbt_free(task->name);
 208
 209   if (task->simdata->compute)
 210     task->simdata->compute->unref();
 211
 212   /* parallel tasks only */
 213   xbt_free(task->simdata->host_list);
 214
 215   xbt_dict_free(&task->simdata->affinity_mask_db);
 216
 217   /* free main structures */
 218   xbt_free(task->simdata);
 219   xbt_free(task);
 220
 221   return MSG_OK;
 222 }
 223
 224 /** \ingroup m_task_usage
 225  * \brief Cancel a #msg_task_t.
 226  * \param task the task to cancel. If it was executed or transfered, it stops the process that were working on it.
 227  */
 228 msg_error_t MSG_task_cancel(msg_task_t task)
 229 {
 230   xbt_assert((task != NULL), "Cannot cancel a NULL task");
 231
 232   if (task->simdata->compute) {
 233     simcall_execution_cancel(task->simdata->compute);
 234   }
 235   else if (task->simdata->comm) {
 236     simdata_task_t simdata = task->simdata;
 237     simcall_comm_cancel(simdata->comm);
 238     if (msg_global->debug_multiple_use && simdata->isused!=0)
 239       xbt_ex_free(*(xbt_ex_t*)simdata->isused);
 240     simdata->isused = 0;
 241   }
 242   return MSG_OK;
 243 }
 244
 245 /** \ingroup m_task_management
 246  * \brief Returns the remaining amount of flops needed to execute a task #msg_task_t.
 247  *
 248  * Once a task has been processed, this amount is set to 0. If you want, you can reset this value with
 249  * #MSG_task_set_flops_amount before restarting the task.
 250  */
 251 double MSG_task_get_flops_amount(msg_task_t task) {
 252   if (task->simdata->compute) {
 253     return task->simdata->compute->remains();
 254   } else {
 255     return task->simdata->flops_amount;
 256   }
 257 }
 258
 259 /** \ingroup m_task_management
 260  * \brief set the computation amount needed to process a task #msg_task_t.
 261  *
 262  * \warning If the computation is ongoing (already started and not finished),
 263  * it is not modified by this call. Moreover, after its completion, the ongoing execution with set the flops_amount to
 264  * zero, overriding any value set during the execution.
 265  */
 266 void MSG_task_set_flops_amount(msg_task_t task, double flops_amount)
 267 {
 268   task->simdata->flops_amount = flops_amount;
 269 }
 270
 271 /** \ingroup m_task_management
 272  * \brief set the amount data attached with a task #msg_task_t.
 273  *
 274  * \warning If the transfer is ongoing (already started and not finished), it is not modified by this call.
 275  */
 276 void MSG_task_set_bytes_amount(msg_task_t task, double data_size)
 277 {
 278   task->simdata->bytes_amount = data_size;
 279 }
 280
 281 /** \ingroup m_task_management
 282  * \brief Returns the total amount received by a task #msg_task_t.
 283  *        If the communication does not exist it will return 0.
 284  *        So, if the communication has FINISHED or FAILED it returns zero.
 285  */
 286 double MSG_task_get_remaining_communication(msg_task_t task)
 287 {
 288   XBT_DEBUG("calling simcall_communication_get_remains(%p)", task->simdata->comm);
 289   return task->simdata->comm->remains();
 290 }
 291
 292 /** \ingroup m_task_management
 293  * \brief Returns the size of the data attached to a task #msg_task_t.
 294  */
 295 double MSG_task_get_bytes_amount(msg_task_t task)
 296 {
 297   xbt_assert((task != NULL) && (task->simdata != NULL), "Invalid parameter");
 298   return task->simdata->bytes_amount;
 299 }
 300
 301 /** \ingroup m_task_management
 302  * \brief Changes the priority of a computation task. This priority doesn't affect the transfer rate. A priority of 2
 303  *        will make a task receive two times more cpu power than the other ones.
 304  */
 305 void MSG_task_set_priority(msg_task_t task, double priority)
 306 {
 307   xbt_assert((task != NULL) && (task->simdata != NULL), "Invalid parameter");
 308   task->simdata->priority = 1 / priority;
 309   if (task->simdata->compute)
 310     simcall_execution_set_priority(task->simdata->compute,
 311         task->simdata->priority);
 312 }
 313
 314 /** \ingroup m_task_management
 315  * \brief Changes the maximum CPU utilization of a computation task.
 316  *        Unit is flops/s.
 317  *
 318  * For VMs, there is a pitfall. Please see MSG_vm_set_bound().
 319  */
 320 void MSG_task_set_bound(msg_task_t task, double bound)
 321 {
 322   xbt_assert(task, "Invalid parameter");
 323   xbt_assert(task->simdata, "Invalid parameter");
 324
 325   if (bound == 0)
 326     XBT_INFO("bound == 0 means no capping (i.e., unlimited).");
 327
 328   task->simdata->bound = bound;
 329   if (task->simdata->compute)
 330     simcall_execution_set_bound(task->simdata->compute, task->simdata->bound);
 331 }
 332
 333 /** \ingroup m_task_management
 334  * \brief Changes the CPU affinity of a computation task.
 335  *
 336  * When pinning the given task to the first CPU core of the given host, use 0x01 for the mask value. Each bit of the
 337  * mask value corresponds to each CPU core. See taskset(1) on Linux.
 338  *
 339  * \param task a target task
 340  * \param host the host having a multi-core CPU
 341  * \param mask the bit mask of a new CPU affinity setting for the task
 342  *
 343  * Usage:
 344  * 0. Define a host with multiple cores.
 345  *    \<host id="PM0" power="1E8" core="2"/\>
 346  *
 347  * 1. Pin a given task to the first CPU core of a host.
 348  *   MSG_task_set_affinity(task, pm0, 0x01);
 349  *
 350  * 2. Pin a given task to the third CPU core of a host. Turn on the third bit of the mask.
 351  *   MSG_task_set_affinity(task, pm0, 0x04); // 0x04 == 100B
 352  *
 353  * 3. Pin a given VM to the first CPU core of a host.
 354  *   MSG_vm_set_affinity(vm, pm0, 0x01);
 355  *
 356  * See examples/msg/cloud/multicore.c for more information.
 357  *
 358  * Note:
 359  * 1. The current code does not allow an affinity of a task to multiple cores.
 360  *    The mask value 0x03 (i.e., a given task will be executed on the first core or the second core) is not allowed.
 361  *    The mask value 0x01 or 0x02 works. See cpu_cas01.c for details.
 362  *
 363  * 2. It is recommended to first compare simulation results in both the Lazy and Full calculation modes
 364  *    (using --cfg=cpu/optim:Full or not). Fix cpu_cas01.c if you find wrong results in the Lazy mode.
 365  */
 366 void MSG_task_set_affinity(msg_task_t task, msg_host_t host, unsigned long mask)
 367 {
 368   xbt_assert(task, "Invalid parameter");
 369   xbt_assert(task->simdata, "Invalid parameter");
 370
 371   if (mask == 0) {
 372     /* 0 means clear */
 373       /* We need remove_ext() not throwing exception. */
 374       void *ret = xbt_dict_get_or_null_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(msg_host_t));
 375       if (ret != NULL)
 376         xbt_dict_remove_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(host));
 377   } else
 378     xbt_dict_set_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(host), (void *)(uintptr_t) mask, NULL);
 379
 380   /* We set affinity data of this task. If the task is being executed, we actually change the affinity setting of the
 381    * task. Otherwise, this change will be applied when the task is executed. */
 382   if (!task->simdata->compute) {
 383     /* task is not yet executed */
 384     XBT_INFO("set affinity(0x%04lx@%s) for %s (not active now)", mask, MSG_host_get_name(host),
 385              MSG_task_get_name(task));
 386     return;
 387   }
 388
 389   {
 390     simgrid::simix::Exec *compute = task->simdata->compute;
 391     msg_host_t host_now = compute->host;  // simix_private.h is necessary
 392     if (host_now != host) {
 393       /* task is not yet executed on this host */
 394       XBT_INFO("set affinity(0x%04lx@%s) for %s (not active now)", mask, MSG_host_get_name(host),
 395                MSG_task_get_name(task));
 396       return;
 397     }
 398
 399     /* task is being executed on this host. so change the affinity now */
 400     {
 401       /* check it works. remove me if it works. */
 402       xbt_assert((unsigned long)(uintptr_t) xbt_dict_get_or_null_ext(task->simdata->affinity_mask_db,
 403                  (char *) host, sizeof(msg_host_t)) == mask);
 404     }
 405
 406     XBT_INFO("set affinity(0x%04lx@%s) for %s", mask, MSG_host_get_name(host), MSG_task_get_name(task));
 407     simcall_execution_set_affinity(task->simdata->compute, host, mask);
 408   }
 409 }