Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
sed -i -e 's/\t/ /g' [sources] Please people, stop using tabs
[simgrid.git] / 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  *
15  *  Since most scheduling algorithms rely on a concept of task
16  *  that can be either <em>computed</em> locally or
17  *  <em>transferred</em> on another processor, it seems to be the
18  *  right level of abstraction for our purposes. A <em>task</em>
19  *  may then be defined by a <em>computing amount</em>, a
20  *  <em>message size</em> and some <em>private data</em>.
21  */
22
23 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_task, msg,
24                                 "Logging specific to MSG (task)");
25
26 /********************************* Task **************************************/
27 /** \ingroup m_task_management
28  * \brief Creates a new #msg_task_t.
29  *
30  * A constructor for #msg_task_t taking four arguments and returning the
31    corresponding object.
32  * \param name a name for the object. It is for user-level information
33    and can be NULL.
34  * \param flop_amount a value of the processing amount (in flop)
35    needed to process this new task. If 0, then it cannot be executed with
36    MSG_task_execute(). This value has to be >=0.
37  * \param message_size a value of the amount of data (in bytes) needed to
38    transfer this new task. If 0, then it cannot be transfered with
39    MSG_task_send() and MSG_task_recv(). This value has to be >=0.
40  * \param data a pointer to any data may want to attach to the new
41    object.  It is for user-level information and can be NULL. It can
42    be retrieved with the function \ref MSG_task_get_data.
43  * \see msg_task_t
44  * \return The new corresponding object.
45  */
46 msg_task_t MSG_task_create(const char *name, double flop_amount,
47                          double message_size, void *data)
48 {
49   msg_task_t task = xbt_new(s_msg_task_t, 1);
50   simdata_task_t simdata = xbt_new(s_simdata_task_t, 1);
51   task->simdata = simdata;
52
53   /* Task structure */
54   task->name = xbt_strdup(name);
55   task->data = data;
56
57   /* Simulator Data */
58   simdata->compute = NULL;
59   simdata->comm = NULL;
60   simdata->bytes_amount = message_size;
61   simdata->flops_amount = flop_amount;
62   simdata->sender = NULL;
63   simdata->receiver = NULL;
64   simdata->source = NULL;
65   simdata->priority = 1.0;
66   simdata->bound = 0;
67   simdata->affinity_mask_db = xbt_dict_new_homogeneous(NULL);
68   simdata->rate = -1.0;
69   simdata->isused = 0;
70
71   simdata->host_nb = 0;
72   simdata->host_list = NULL;
73   simdata->flops_parallel_amount = NULL;
74   simdata->bytes_parallel_amount = NULL;
75   TRACE_msg_task_create(task);
76
77   return task;
78 }
79
80 /** \ingroup m_task_management
81  * \brief Creates a new #msg_task_t (a parallel one....).
82  *
83  * A constructor for #msg_task_t taking six arguments and returning the
84  corresponding object.
85  * \param name a name for the object. It is for user-level information
86  and can be NULL.
87  * \param host_nb the number of hosts implied in the parallel task.
88  * \param host_list an array of \p host_nb msg_host_t.
89  * \param flops_amount an array of \p host_nb doubles.
90  *                     flops_amount[i] is the total number of operations that have to be performed on host_list[i].
91  * \param bytes_amount an array of \p host_nb* \p host_nb doubles.
92  * \param data a pointer to any data may want to attach to the new object.
93  *             It is for user-level information and can be NULL.
94  *             It can be retrieved with the function \ref MSG_task_get_data.
95  * \see msg_task_t
96  * \return The new corresponding object.
97  */
98 msg_task_t
99 MSG_parallel_task_create(const char *name, int host_nb,
100                          const msg_host_t * host_list,
101                          double *flops_amount,
102                          double *bytes_amount, void *data)
103 {
104   msg_task_t task = MSG_task_create(name, 0, 0, data);
105   simdata_task_t simdata = task->simdata;
106   int i;
107
108   /* Simulator Data specific to parallel tasks */
109   simdata->host_nb = host_nb;
110   simdata->host_list = xbt_new0(sg_host_t, host_nb);
111   simdata->flops_parallel_amount = flops_amount;
112   simdata->bytes_parallel_amount = bytes_amount;
113
114   for (i = 0; i < host_nb; i++)
115     simdata->host_list[i] = host_list[i];
116
117   return task;
118 }
119
120 /*************** Begin GPU ***************/
121 /** \ingroup m_task_management
122  * \brief Creates a new #msg_gpu_task_t.
123
124  * A constructor for #msg_gpu_task_t taking four arguments and returning
125    a pointer to the new created GPU task.
126
127  * \param name a name for the object. It is for user-level information
128    and can be NULL.
129
130  * \param flops_amount a value of the processing amount (in flop)
131    needed to process this new task. If 0, then it cannot be executed with
132    MSG_gpu_task_execute(). This value has to be >=0.
133
134  * \param dispatch_latency time in seconds to load this task on the GPU
135
136  * \param collect_latency time in seconds to transfer result from the GPU
137    back to the CPU (host) when done
138
139  * \see msg_gpu_task_t
140  * \return The new corresponding object.
141  */
142 msg_gpu_task_t MSG_gpu_task_create(const char *name, double flops_amount,
143                          double dispatch_latency, double collect_latency)
144 {
145   msg_gpu_task_t task = xbt_new(s_msg_gpu_task_t, 1);
146   simdata_gpu_task_t simdata = xbt_new(s_simdata_gpu_task_t, 1);
147   task->simdata = simdata;
148   /* Task structure */
149   task->name = xbt_strdup(name);
150
151   /* Simulator Data */
152   simdata->flops_amount = flops_amount;
153   simdata->dispatch_latency   = dispatch_latency;
154   simdata->collect_latency    = collect_latency;
155
156   /* TRACE_msg_gpu_task_create(task); FIXME*/
157
158   return task;
159 }
160 /*************** End GPU ***************/
161
162 /** \ingroup m_task_management
163  * \brief Return the user data of a #msg_task_t.
164  *
165  * This function checks whether \a task is a valid pointer or not and return
166    the user data associated to \a task if it is possible.
167  */
168 void *MSG_task_get_data(msg_task_t task)
169 {
170   xbt_assert((task != NULL), "Invalid parameter");
171
172   return (task->data);
173 }
174
175 /** \ingroup m_task_management
176  * \brief Sets the user data of a #msg_task_t.
177  *
178  * This function allows to associate a new pointer to
179    the user data associated of \a task.
180  */
181 void MSG_task_set_data(msg_task_t task, void *data)
182 {
183   xbt_assert((task != NULL), "Invalid parameter");
184
185   task->data = data;
186 }
187
188 /** \ingroup m_task_management
189  * \brief Sets a function to be called when a task has just been copied.
190  * \param callback a callback function
191  */
192 void MSG_task_set_copy_callback(void (*callback)
193     (msg_task_t task, msg_process_t sender, msg_process_t receiver)) {
194
195   msg_global->task_copy_callback = callback;
196
197   if (callback) {
198     SIMIX_comm_set_copy_data_callback(MSG_comm_copy_data_from_SIMIX);
199   }
200   else {
201     SIMIX_comm_set_copy_data_callback(SIMIX_comm_copy_pointer_callback);
202   }
203 }
204
205 /** \ingroup m_task_management
206  * \brief Return the sender of a #msg_task_t.
207  *
208  * This functions returns the #msg_process_t which sent this task
209  */
210 msg_process_t MSG_task_get_sender(msg_task_t task)
211 {
212   xbt_assert(task, "Invalid parameters");
213   return ((simdata_task_t) task->simdata)->sender;
214 }
215
216 /** \ingroup m_task_management
217  * \brief Return the source of a #msg_task_t.
218  *
219  * This functions returns the #msg_host_t from which this task was sent
220  */
221 msg_host_t MSG_task_get_source(msg_task_t task)
222 {
223   xbt_assert(task, "Invalid parameters");
224   return ((simdata_task_t) task->simdata)->source;
225 }
226
227 /** \ingroup m_task_management
228  * \brief Return the name of a #msg_task_t.
229  *
230  * This functions returns the name of a #msg_task_t as specified on creation
231  */
232 const char *MSG_task_get_name(msg_task_t task)
233 {
234   xbt_assert(task, "Invalid parameters");
235   return task->name;
236 }
237
238 /** \ingroup m_task_management
239  * \brief Sets the name of a #msg_task_t.
240  *
241  * This functions allows to associate a name to a task
242  */
243 void MSG_task_set_name(msg_task_t task, const char *name)
244 {
245   xbt_assert(task, "Invalid parameters");
246   task->name = xbt_strdup(name);
247 }
248
249 /** \ingroup m_task_management
250  * \brief Destroy a #msg_task_t.
251  *
252  * Destructor for #msg_task_t. Note that you should free user data, if any, \b
253  * before calling this function.
254  *
255  * Only the process that owns the task can destroy it.
256  * The owner changes after a successful send.
257  * If a task is successfully sent, the receiver becomes the owner and is
258  * supposed to destroy it. The sender should not use it anymore.
259  * If the task failed to be sent, the sender remains the owner of the task.
260  */
261 msg_error_t MSG_task_destroy(msg_task_t task)
262 {
263   smx_synchro_t action = NULL;
264   xbt_assert((task != NULL), "Invalid parameter");
265
266   if (task->simdata->isused) {
267     /* the task is being sent or executed: cancel it first */
268     MSG_task_cancel(task);
269   }
270   TRACE_msg_task_destroy(task);
271
272   xbt_free(task->name);
273
274   action = task->simdata->compute;
275   if (action)
276     simcall_execution_destroy(action);
277
278   /* parallel tasks only */
279   xbt_free(task->simdata->host_list);
280
281   xbt_dict_free(&task->simdata->affinity_mask_db);
282
283   /* free main structures */
284   xbt_free(task->simdata);
285   xbt_free(task);
286
287   return MSG_OK;
288 }
289
290
291 /** \ingroup m_task_usage
292  * \brief Cancel a #msg_task_t.
293  * \param task the task to cancel. If it was executed or transfered, it
294           stops the process that were working on it.
295  */
296 msg_error_t MSG_task_cancel(msg_task_t task)
297 {
298   xbt_assert((task != NULL), "Cannot cancel a NULL task");
299
300   if (task->simdata->compute) {
301     simcall_execution_cancel(task->simdata->compute);
302   }
303   else if (task->simdata->comm) {
304     simdata_task_t simdata = task->simdata;
305     simcall_comm_cancel(simdata->comm);
306     if (msg_global->debug_multiple_use && simdata->isused!=0)
307       xbt_ex_free(*(xbt_ex_t*)simdata->isused);
308     simdata->isused = 0;
309   }
310   return MSG_OK;
311 }
312
313 /** \ingroup m_task_management
314  * \brief Returns the remaining amount of flops needed to execute a task #msg_task_t.
315  *
316  * Once a task has been processed, this amount is set to 0. If you want, you
317  * can reset this value with #MSG_task_set_flops_amount before restarting the task.
318  */
319 double MSG_task_get_flops_amount(msg_task_t task) {
320
321   if (task->simdata->compute) {
322     return simcall_execution_get_remains(task->simdata->compute);
323   } else {
324     return task->simdata->flops_amount;
325   }
326 }
327
328
329 /** \ingroup m_task_management
330  * \brief set the computation amount needed to process a task #msg_task_t.
331  *
332  * \warning If the computation is ongoing (already started and not finished),
333  * it is not modified by this call. Moreover, after its completion, the ongoing
334  * execution with set the flops_amount to zero, overriding any value set during
335  * the execution.
336  */
337
338 void MSG_task_set_flops_amount(msg_task_t task, double flops_amount)
339 {
340   task->simdata->flops_amount = flops_amount;
341 }
342
343 /** \ingroup m_task_management
344  * \brief set the amount data attached with a task #msg_task_t.
345  *
346  * \warning If the transfer is ongoing (already started and not finished),
347  * it is not modified by this call.
348  */
349
350 void MSG_task_set_bytes_amount(msg_task_t task, double data_size)
351 {
352   task->simdata->bytes_amount = data_size;
353 }
354
355
356 /** \ingroup m_task_management
357  * \brief Returns the total amount received by a task #msg_task_t.
358  *        If the communication does not exist it will return 0.
359  *        So, if the communication has FINISHED or FAILED it returns
360  *        zero.
361  */
362 double MSG_task_get_remaining_communication(msg_task_t task)
363 {
364   xbt_assert((task != NULL)
365               && (task->simdata != NULL), "Invalid parameter");
366   XBT_DEBUG("calling simcall_communication_get_remains(%p)",
367          task->simdata->comm);
368   return simcall_comm_get_remains(task->simdata->comm);
369 }
370
371 #ifdef HAVE_LATENCY_BOUND_TRACKING
372 /** \ingroup m_task_management
373  * \brief Return 1 if communication task is limited by latency, 0 otherwise
374  *
375  */
376 int MSG_task_is_latency_bounded(msg_task_t task)
377 {
378   xbt_assert((task != NULL)
379               && (task->simdata != NULL), "Invalid parameter");
380   XBT_DEBUG("calling simcall_communication_is_latency_bounded(%p)",
381          task->simdata->comm);
382   return simcall_comm_is_latency_bounded(task->simdata->comm);
383 }
384 #endif
385
386 /** \ingroup m_task_management
387  * \brief Returns the size of the data attached to a task #msg_task_t.
388  *
389  */
390 double MSG_task_get_bytes_amount(msg_task_t task)
391 {
392   xbt_assert((task != NULL)
393               && (task->simdata != NULL), "Invalid parameter");
394
395   return task->simdata->bytes_amount;
396 }
397
398
399
400 /** \ingroup m_task_management
401  * \brief Changes the priority of a computation task. This priority doesn't affect
402  *        the transfer rate. A priority of 2 will make a task receive two times more
403  *        cpu power than the other ones.
404  *
405  */
406 void MSG_task_set_priority(msg_task_t task, double priority)
407 {
408   xbt_assert((task != NULL)
409               && (task->simdata != NULL), "Invalid parameter");
410
411   task->simdata->priority = 1 / priority;
412   if (task->simdata->compute)
413     simcall_execution_set_priority(task->simdata->compute,
414         task->simdata->priority);
415 }
416
417
418 /** \ingroup m_task_management
419  * \brief Changes the maximum CPU utilization of a computation task.
420  *        Unit is flops/s.
421  *
422  * For VMs, there is a pitfall. Please see MSG_vm_set_bound().
423  */
424 void MSG_task_set_bound(msg_task_t task, double bound)
425 {
426   xbt_assert(task, "Invalid parameter");
427   xbt_assert(task->simdata, "Invalid parameter");
428
429   if (bound == 0)
430     XBT_INFO("bound == 0 means no capping (i.e., unlimited).");
431
432   task->simdata->bound = bound;
433   if (task->simdata->compute)
434     simcall_execution_set_bound(task->simdata->compute,
435                                       task->simdata->bound);
436 }
437
438
439 /** \ingroup m_task_management
440  * \brief Changes the CPU affinity of a computation task.
441  *
442  * When pinning the given task to the first CPU core of the given host, use
443  * 0x01 for the mask value. Each bit of the mask value corresponds to each CPU
444  * core. See taskset(1) on Linux.
445  *
446  * \param task a target task
447  * \param host the host having a multi-core CPU
448  * \param mask the bit mask of a new CPU affinity setting for the task
449  *
450  *
451  * Usage:
452  * 0. Define a host with multiple cores.
453  *    \<host id="PM0" power="1E8" core="2"/\>
454  *
455  * 1. Pin a given task to the first CPU core of a host.
456  *   MSG_task_set_affinity(task, pm0, 0x01);
457  *
458  * 2. Pin a given task to the third CPU core of a host. Turn on the third bit of the mask.
459  *   MSG_task_set_affinity(task, pm0, 0x04); // 0x04 == 100B
460  *
461  * 3. Pin a given VM to the first CPU core of a host.
462  *   MSG_vm_set_affinity(vm, pm0, 0x01);
463  *
464  * See examples/msg/cloud/multicore.c for more information.
465  *
466  *
467  * Note:
468  * 1. The current code does not allow an affinity of a task to multiple cores.
469  * The mask value 0x03 (i.e., a given task will be executed on the first core
470  * or the second core) is not allowed. The mask value 0x01 or 0x02 works. See
471  * cpu_cas01.c for details.
472  *
473  * 2. It is recommended to first compare simulation results in both the Lazy
474  * and Full calculation modes (using --cfg=cpu/optim:Full or not). Fix
475  * cpu_cas01.c if you find wrong results in the Lazy mode.
476  *
477  */
478 void MSG_task_set_affinity(msg_task_t task, msg_host_t host, unsigned long mask)
479 {
480   xbt_assert(task, "Invalid parameter");
481   xbt_assert(task->simdata, "Invalid parameter");
482
483   if (mask == 0) {
484     /* 0 means clear */
485     {
486       /* We need remove_ext() not throwing exception. */
487       void *ret = xbt_dict_get_or_null_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(msg_host_t));
488       if (ret != NULL)
489         xbt_dict_remove_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(host));
490     }
491   } else
492     xbt_dict_set_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(host), (void *)(uintptr_t) mask, NULL);
493
494   /* We set affinity data of this task. If the task is being executed, we
495    * actually change the affinity setting of the task. Otherwise, this change
496    * will be applied when the task is executed. */
497
498   if (!task->simdata->compute) {
499     /* task is not yet executed */
500     XBT_INFO("set affinity(0x%04lx@%s) for %s (not active now)", mask, MSG_host_get_name(host), MSG_task_get_name(task));
501     return;
502   }
503
504   {
505     smx_synchro_t compute = task->simdata->compute;
506     msg_host_t host_now = compute->execution.host;  // simix_private.h is necessary
507     if (host_now != host) {
508       /* task is not yet executed on this host */
509       XBT_INFO("set affinity(0x%04lx@%s) for %s (not active now)", mask, MSG_host_get_name(host), MSG_task_get_name(task));
510       return;
511     }
512
513     /* task is being executed on this host. so change the affinity now */
514     {
515       /* check it works. remove me if it works. */
516       xbt_assert((unsigned long)(uintptr_t) xbt_dict_get_or_null_ext(task->simdata->affinity_mask_db, (char *) host, sizeof(msg_host_t)) == mask);
517     }
518
519     XBT_INFO("set affinity(0x%04lx@%s) for %s", mask, MSG_host_get_name(host), MSG_task_get_name(task));
520     simcall_execution_set_affinity(task->simdata->compute, host, mask);
521   }
522 }