Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
use new/delete instead of xbt_new0/xbt_free on C++ objects
[simgrid.git] / src / xbt / parmap.cpp
1 /* Copyright (c) 2004-2005, 2007, 2009-2014. 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 <atomic>
8
9 #include "internal_config.h"
10 #ifdef HAVE_UNISTD_H
11 #include <unistd.h>
12 #endif
13
14 #ifndef _XBT_WIN32
15 #include <sys/syscall.h>
16 #endif
17
18 #ifdef HAVE_FUTEX_H
19 #include <linux/futex.h>
20 #include <limits.h>
21 #endif
22
23 #include "xbt/parmap.h"
24 #include "xbt/log.h"
25 #include "xbt/function_types.h"
26 #include "xbt/dynar.h"
27 #include "xbt/xbt_os_thread.h"
28 #include "xbt/sysdep.h"
29 #include "simix/smx_private.h"
30
31 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_parmap, xbt, "parmap: parallel map");
32
33 typedef enum {
34   XBT_PARMAP_WORK,
35   XBT_PARMAP_DESTROY
36 } e_xbt_parmap_flag_t;
37
38 static void xbt_parmap_set_mode(xbt_parmap_t parmap, e_xbt_parmap_mode_t mode);
39 static void *xbt_parmap_worker_main(void *parmap);
40 static void xbt_parmap_work(xbt_parmap_t parmap);
41
42 static void xbt_parmap_posix_master_wait(xbt_parmap_t parmap);
43 static void xbt_parmap_posix_worker_signal(xbt_parmap_t parmap);
44 static void xbt_parmap_posix_master_signal(xbt_parmap_t parmap);
45 static void xbt_parmap_posix_worker_wait(xbt_parmap_t parmap, unsigned round);
46
47 #ifdef HAVE_FUTEX_H
48 static void xbt_parmap_futex_master_wait(xbt_parmap_t parmap);
49 static void xbt_parmap_futex_worker_signal(xbt_parmap_t parmap);
50 static void xbt_parmap_futex_master_signal(xbt_parmap_t parmap);
51 static void xbt_parmap_futex_worker_wait(xbt_parmap_t parmap, unsigned round);
52 static void futex_wait(unsigned *uaddr, unsigned val);
53 static void futex_wake(unsigned *uaddr, unsigned val);
54 #endif
55
56 static void xbt_parmap_busy_master_wait(xbt_parmap_t parmap);
57 static void xbt_parmap_busy_worker_signal(xbt_parmap_t parmap);
58 static void xbt_parmap_busy_master_signal(xbt_parmap_t parmap);
59 static void xbt_parmap_busy_worker_wait(xbt_parmap_t parmap, unsigned round);
60
61 #ifdef HAVE_MC
62 static void xbt_parmap_mc_work(xbt_parmap_t parmap, int worker_id);
63 static void *xbt_parmap_mc_worker_main(void *arg);
64 #endif
65
66 /**
67  * \brief Parallel map structure
68  */
69 typedef struct s_xbt_parmap {
70   e_xbt_parmap_flag_t status;      /**< is the parmap active or being destroyed? */
71   unsigned work;                   /**< index of the current round */
72   unsigned thread_counter;         /**< number of workers that have done the work */
73
74   unsigned int num_workers;        /**< total number of worker threads including the controller */
75   xbt_os_thread_t *workers;        /**< worker thread handlers */
76   void_f_pvoid_t fun;              /**< function to run in parallel on each element of data */
77   xbt_dynar_t data;                /**< parameters to pass to fun in parallel */
78   std::atomic<unsigned int> index; /**< index of the next element of data to pick */
79
80 #ifdef HAVE_MC
81   int finish;
82   void* ref_snapshot;
83   int_f_pvoid_pvoid_t snapshot_compare;
84   unsigned int length;
85   void* mc_data;
86 #endif
87
88   /* posix only */
89   xbt_os_cond_t ready_cond;
90   xbt_os_mutex_t ready_mutex;
91   xbt_os_cond_t done_cond;
92   xbt_os_mutex_t done_mutex;
93
94   /* fields that depend on the synchronization mode */
95   e_xbt_parmap_mode_t mode;        /**< synchronization mode */
96   void (*master_wait_f)(xbt_parmap_t);    /**< wait for the workers to have done the work */
97   void (*worker_signal_f)(xbt_parmap_t);  /**< signal the master that a worker has done the work */
98   void (*master_signal_f)(xbt_parmap_t);  /**< wakes the workers threads to process tasks */
99   void (*worker_wait_f)(xbt_parmap_t, unsigned); /**< waits for more work */
100 } s_xbt_parmap_t;
101
102 /**
103  * \brief Thread data transmission structure
104  */
105 typedef struct s_xbt_parmap_thread_data{
106   xbt_parmap_t parmap;
107   int worker_id;
108 } s_xbt_parmap_thread_data_t;
109
110 typedef s_xbt_parmap_thread_data_t *xbt_parmap_thread_data_t;
111
112 /**
113  * \brief Creates a parallel map object
114  * \param num_workers number of worker threads to create
115  * \param mode how to synchronize the worker threads
116  * \return the parmap created
117  */
118 xbt_parmap_t xbt_parmap_new(unsigned int num_workers, e_xbt_parmap_mode_t mode)
119 {
120   unsigned int i;
121
122   XBT_DEBUG("Create new parmap (%u workers)", num_workers);
123
124   /* Initialize the thread pool data structure */
125   xbt_parmap_t parmap = new s_xbt_parmap_t();
126   parmap->workers = xbt_new(xbt_os_thread_t, num_workers);
127
128   parmap->num_workers = num_workers;
129   parmap->status = XBT_PARMAP_WORK;
130   xbt_parmap_set_mode(parmap, mode);
131
132   /* Create the pool of worker threads */
133   xbt_parmap_thread_data_t data;
134   parmap->workers[0] = NULL;
135 #ifdef CORE_BINDING  
136   unsigned int core_bind = 0;
137 #endif  
138   for (i = 1; i < num_workers; i++) {
139     data = xbt_new0(s_xbt_parmap_thread_data_t, 1);
140     data->parmap = parmap;
141     data->worker_id = i;
142     parmap->workers[i] = xbt_os_thread_create(NULL, xbt_parmap_worker_main,
143                                               data, NULL);
144 #ifdef CORE_BINDING  
145     xbt_os_thread_bind(parmap->workers[i], core_bind); 
146     if(core_bind!=xbt_os_get_numcores()){
147       core_bind++;
148     } else {
149       core_bind = 0; 
150     }
151 #endif    
152   }
153   return parmap;
154 }
155
156 #ifdef HAVE_MC
157 /**
158  * \brief Creates a parallel map object
159  * \param num_workers number of worker threads to create
160  * \param mode how to synchronize the worker threads
161  * \return the parmap created
162  */
163 xbt_parmap_t xbt_parmap_mc_new(unsigned int num_workers, e_xbt_parmap_mode_t mode)
164 {
165   unsigned int i;
166
167   XBT_DEBUG("Create new parmap (%u workers)", num_workers);
168
169   /* Initialize the thread pool data structure */
170   xbt_parmap_t parmap = new s_xbt_parmap_t();
171   parmap->workers = xbt_new(xbt_os_thread_t, num_workers);
172
173   parmap->num_workers = num_workers;
174   parmap->status = XBT_PARMAP_WORK;
175   xbt_parmap_set_mode(parmap, mode);
176
177   /* Create the pool of worker threads */
178   xbt_parmap_thread_data_t data;
179   parmap->workers[0] = NULL;
180   for (i = 1; i < num_workers; i++) {
181     data = xbt_new0(s_xbt_parmap_thread_data_t, 1);
182     data->parmap = parmap;
183     data->worker_id = i;
184     parmap->workers[i] = xbt_os_thread_create(NULL, xbt_parmap_mc_worker_main,
185                                               data, NULL);
186   }
187   return parmap;
188 }
189 #endif
190
191 /**
192  * \brief Destroys a parmap
193  * \param parmap the parmap to destroy
194  */
195 void xbt_parmap_destroy(xbt_parmap_t parmap)
196 {
197   if (!parmap) {
198     return;
199   }
200
201   parmap->status = XBT_PARMAP_DESTROY;
202   parmap->master_signal_f(parmap);
203
204   unsigned int i;
205   for (i = 1; i < parmap->num_workers; i++)
206     xbt_os_thread_join(parmap->workers[i], NULL);
207
208   xbt_os_cond_destroy(parmap->ready_cond);
209   xbt_os_mutex_destroy(parmap->ready_mutex);
210   xbt_os_cond_destroy(parmap->done_cond);
211   xbt_os_mutex_destroy(parmap->done_mutex);
212
213   xbt_free(parmap->workers);
214   delete parmap;
215 }
216
217 /**
218  * \brief Sets the synchronization mode of a parmap.
219  * \param parmap a parallel map object
220  * \param mode the synchronization mode
221  */
222 static void xbt_parmap_set_mode(xbt_parmap_t parmap, e_xbt_parmap_mode_t mode)
223 {
224   if (mode == XBT_PARMAP_DEFAULT) {
225 #ifdef HAVE_FUTEX_H
226     mode = XBT_PARMAP_FUTEX;
227 #else
228     mode = XBT_PARMAP_POSIX;
229 #endif
230   }
231   parmap->mode = mode;
232
233   switch (mode) {
234
235     case XBT_PARMAP_POSIX:
236       parmap->master_wait_f = xbt_parmap_posix_master_wait;
237       parmap->worker_signal_f = xbt_parmap_posix_worker_signal;
238       parmap->master_signal_f = xbt_parmap_posix_master_signal;
239       parmap->worker_wait_f = xbt_parmap_posix_worker_wait;
240
241       parmap->ready_cond = xbt_os_cond_init();
242       parmap->ready_mutex = xbt_os_mutex_init();
243       parmap->done_cond = xbt_os_cond_init();
244       parmap->done_mutex = xbt_os_mutex_init();
245       break;
246
247
248     case XBT_PARMAP_FUTEX:
249 #ifdef HAVE_FUTEX_H
250       parmap->master_wait_f = xbt_parmap_futex_master_wait;
251       parmap->worker_signal_f = xbt_parmap_futex_worker_signal;
252       parmap->master_signal_f = xbt_parmap_futex_master_signal;
253       parmap->worker_wait_f = xbt_parmap_futex_worker_wait;
254
255       xbt_os_cond_destroy(parmap->ready_cond);
256       xbt_os_mutex_destroy(parmap->ready_mutex);
257       xbt_os_cond_destroy(parmap->done_cond);
258       xbt_os_mutex_destroy(parmap->done_mutex);
259       break;
260 #else
261       xbt_die("Futex is not available on this OS.");
262 #endif
263
264     case XBT_PARMAP_BUSY_WAIT:
265       parmap->master_wait_f = xbt_parmap_busy_master_wait;
266       parmap->worker_signal_f = xbt_parmap_busy_worker_signal;
267       parmap->master_signal_f = xbt_parmap_busy_master_signal;
268       parmap->worker_wait_f = xbt_parmap_busy_worker_wait;
269
270       xbt_os_cond_destroy(parmap->ready_cond);
271       xbt_os_mutex_destroy(parmap->ready_mutex);
272       xbt_os_cond_destroy(parmap->done_cond);
273       xbt_os_mutex_destroy(parmap->done_mutex);
274       break;
275
276     case XBT_PARMAP_DEFAULT:
277       THROW_IMPOSSIBLE;
278       break;
279   }
280 }
281
282 /**
283  * \brief Applies a list of tasks in parallel.
284  * \param parmap a parallel map object
285  * \param fun the function to call in parallel
286  * \param data each element of this dynar will be passed as an argument to fun
287  */
288 void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
289 {
290   /* Assign resources to worker threads (we are maestro here)*/
291   parmap->fun = fun;
292   parmap->data = data;
293   parmap->index = 0;
294   parmap->master_signal_f(parmap); // maestro runs futex_wait to wake all the minions (the working threads)
295   xbt_parmap_work(parmap);         // maestro works with its minions
296   parmap->master_wait_f(parmap);   // When there is no more work to do, then maestro waits for the last minion to stop
297   XBT_DEBUG("Job done");           //   ... and proceeds
298 }
299
300 /**
301  * \brief Returns a next task to process.
302  *
303  * Worker threads call this function to get more work.
304  *
305  * \return the next task to process, or NULL if there is no more work
306  */
307 void* xbt_parmap_next(xbt_parmap_t parmap)
308 {
309   unsigned int index = parmap->index++;
310   if (index < xbt_dynar_length(parmap->data)) {
311     return xbt_dynar_get_as(parmap->data, index, void*);
312   }
313   return NULL;
314 }
315
316 static void xbt_parmap_work(xbt_parmap_t parmap)
317 {
318   unsigned index;
319   while ((index = parmap->index++)
320          < xbt_dynar_length(parmap->data))
321     parmap->fun(xbt_dynar_get_as(parmap->data, index, void*));
322 }
323
324 /**
325  * \brief Main function of a worker thread.
326  * \param arg the parmap
327  */
328 static void *xbt_parmap_worker_main(void *arg)
329 {
330   xbt_parmap_thread_data_t data = (xbt_parmap_thread_data_t) arg;
331   xbt_parmap_t parmap = data->parmap;
332   unsigned round = 0;
333   smx_context_t context = SIMIX_context_new(NULL, 0, NULL, NULL, NULL);
334   SIMIX_context_set_current(context);
335
336   XBT_DEBUG("New worker thread created");
337
338   /* Worker's main loop */
339   while (1) {
340     parmap->worker_wait_f(parmap, ++round);
341     if (parmap->status == XBT_PARMAP_WORK) {
342
343       XBT_DEBUG("Worker %d got a job", data->worker_id);
344
345       xbt_parmap_work(parmap);
346       parmap->worker_signal_f(parmap);
347
348       XBT_DEBUG("Worker %d has finished", data->worker_id);
349
350     /* We are destroying the parmap */
351     } else {
352       SIMIX_context_free(context);
353       xbt_free(data);
354       return NULL;
355     }
356   }
357 }
358
359 #ifdef HAVE_MC
360
361 /**
362  * \brief Applies a list of tasks in parallel.
363  * \param parmap a parallel map object
364  * \param fun the function to call in parallel
365  * \param data each element of this dynar will be passed as an argument to fun
366  */
367 int xbt_parmap_mc_apply(xbt_parmap_t parmap, int_f_pvoid_pvoid_t fun, 
368                          void* data, unsigned int length,  void* ref_snapshot)
369 {
370   /* Assign resources to worker threads */
371   parmap->snapshot_compare = fun;
372   parmap->mc_data = data;
373   parmap->index = 0;
374   parmap->finish = -1;
375   parmap->length = length;
376   parmap->ref_snapshot = ref_snapshot;
377   parmap->master_signal_f(parmap);
378   xbt_parmap_mc_work(parmap, 0);
379   parmap->master_wait_f(parmap);
380   XBT_DEBUG("Job done");
381   return parmap->finish;
382 }
383
384 static void xbt_parmap_mc_work(xbt_parmap_t parmap, int worker_id)
385 {
386   unsigned int data_size = (parmap->length / parmap->num_workers) +
387     ((parmap->length % parmap->num_workers) ? 1 :0);
388   void* start = (char*)parmap->mc_data + (data_size*worker_id*sizeof(void*));
389   void* end = MIN((char *)start + data_size* sizeof(void*), (char*)parmap->mc_data + parmap->length*sizeof(void*));
390   
391   //XBT_CRITICAL("Worker %d : %p -> %p (%d)", worker_id, start, end, data_size);
392
393   while ( start < end && parmap->finish == -1) {
394     //XBT_CRITICAL("Starting with %p", start);
395     int res = parmap->snapshot_compare(*(void**)start, parmap->ref_snapshot);
396     start = (char *)start + sizeof(start);
397     if (!res){
398     
399       parmap->finish = ((char*)start - (char*)parmap->mc_data) / sizeof(void*);
400       //XBT_CRITICAL("Find good one %p (%p)", start, parmap->mc_data);
401       break;
402     }
403   }
404 }
405
406 /**
407  * \brief Main function of a worker thread.
408  * \param arg the parmap
409  */
410 static void *xbt_parmap_mc_worker_main(void *arg)
411 {
412   xbt_parmap_thread_data_t data = (xbt_parmap_thread_data_t) arg;
413   xbt_parmap_t parmap = data->parmap;
414   unsigned round = 0;
415   /* smx_context_t context = SIMIX_context_new(NULL, 0, NULL, NULL, NULL); */
416   /* SIMIX_context_set_current(context); */
417
418   XBT_DEBUG("New worker thread created");
419
420   /* Worker's main loop */
421   while (1) {
422     parmap->worker_wait_f(parmap, ++round);
423     if (parmap->status == XBT_PARMAP_WORK) {
424
425       XBT_DEBUG("Worker %d got a job", data->worker_id);
426
427       xbt_parmap_mc_work(parmap, data->worker_id);
428       parmap->worker_signal_f(parmap);
429
430       XBT_DEBUG("Worker %d has finished", data->worker_id);
431
432     /* We are destroying the parmap */
433     } else {
434       xbt_free(data);
435       return NULL;
436     }
437   }
438 }
439 #endif
440
441 #ifdef HAVE_FUTEX_H
442 static void futex_wait(unsigned *uaddr, unsigned val)
443 {
444   XBT_VERB("Waiting on futex %p", uaddr);
445   syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0);
446 }
447
448 static void futex_wake(unsigned *uaddr, unsigned val)
449 {
450   XBT_VERB("Waking futex %p", uaddr);
451   syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, NULL, NULL, 0);
452 }
453 #endif
454
455 /**
456  * \brief Starts the parmap: waits for all workers to be ready and returns.
457  *
458  * This function is called by the controller thread.
459  *
460  * \param parmap a parmap
461  */
462 static void xbt_parmap_posix_master_wait(xbt_parmap_t parmap)
463 {
464   xbt_os_mutex_acquire(parmap->done_mutex);
465   if (parmap->thread_counter < parmap->num_workers) {
466     /* wait for all workers to be ready */
467     xbt_os_cond_wait(parmap->done_cond, parmap->done_mutex);
468   }
469   xbt_os_mutex_release(parmap->done_mutex);
470 }
471
472 /**
473  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
474  *
475  * This function is called by all worker threads when they end (not including
476  * the controller).
477  *
478  * \param parmap a parmap
479  */
480 static void xbt_parmap_posix_worker_signal(xbt_parmap_t parmap)
481 {
482   xbt_os_mutex_acquire(parmap->done_mutex);
483   if (++parmap->thread_counter == parmap->num_workers) {
484     /* all workers have finished, wake the controller */
485     xbt_os_cond_signal(parmap->done_cond);
486   }
487   xbt_os_mutex_release(parmap->done_mutex);
488 }
489
490 /**
491  * \brief Wakes all workers and waits for them to finish the tasks.
492  *
493  * This function is called by the controller thread.
494  *
495  * \param parmap a parmap
496  */
497 static void xbt_parmap_posix_master_signal(xbt_parmap_t parmap)
498 {
499   xbt_os_mutex_acquire(parmap->ready_mutex);
500   parmap->thread_counter = 1;
501   parmap->work++;
502   /* wake all workers */
503   xbt_os_cond_broadcast(parmap->ready_cond);
504   xbt_os_mutex_release(parmap->ready_mutex);
505 }
506
507 /**
508  * \brief Waits for some work to process.
509  *
510  * This function is called by each worker thread (not including the controller)
511  * when it has no more work to do.
512  *
513  * \param parmap a parmap
514  * \param round  the expected round number
515  */
516 static void xbt_parmap_posix_worker_wait(xbt_parmap_t parmap, unsigned round)
517 {
518   xbt_os_mutex_acquire(parmap->ready_mutex);
519   /* wait for more work */
520   if (parmap->work != round) {
521     xbt_os_cond_wait(parmap->ready_cond, parmap->ready_mutex);
522   }
523   xbt_os_mutex_release(parmap->ready_mutex);
524 }
525
526 #ifdef HAVE_FUTEX_H
527 /**
528  * \brief Starts the parmap: waits for all workers to be ready and returns.
529  *
530  * This function is called by the controller thread.
531  *
532  * \param parmap a parmap
533  */
534 static void xbt_parmap_futex_master_wait(xbt_parmap_t parmap)
535 {
536   unsigned count = parmap->thread_counter;
537   while (count < parmap->num_workers) {
538     /* wait for all workers to be ready */
539     futex_wait(&parmap->thread_counter, count);
540     count = parmap->thread_counter;
541   }
542 }
543
544 /**
545  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
546  *
547  * This function is called by all worker threads when they end (not including
548  * the controller).
549  *
550  * \param parmap a parmap
551  */
552 static void xbt_parmap_futex_worker_signal(xbt_parmap_t parmap)
553 {
554   unsigned count = __sync_add_and_fetch(&parmap->thread_counter, 1);
555   if (count == parmap->num_workers) {
556     /* all workers have finished, wake the controller */
557     futex_wake(&parmap->thread_counter, INT_MAX);
558   }
559 }
560
561 /**
562  * \brief Wakes all workers and waits for them to finish the tasks.
563  *
564  * This function is called by the controller thread.
565  *
566  * \param parmap a parmap
567  */
568 static void xbt_parmap_futex_master_signal(xbt_parmap_t parmap)
569 {
570   parmap->thread_counter = 1;
571   __sync_add_and_fetch(&parmap->work, 1);
572   /* wake all workers */
573   futex_wake(&parmap->work, INT_MAX);
574 }
575
576 /**
577  * \brief Waits for some work to process.
578  *
579  * This function is called by each worker thread (not including the controller)
580  * when it has no more work to do.
581  *
582  * \param parmap a parmap
583  * \param round  the expected round number
584  */
585 static void xbt_parmap_futex_worker_wait(xbt_parmap_t parmap, unsigned round)
586 {
587   unsigned work = parmap->work;
588   /* wait for more work */
589   while (work != round) {
590     futex_wait(&parmap->work, work);
591     work = parmap->work;
592   }
593 }
594 #endif
595
596 /**
597  * \brief Starts the parmap: waits for all workers to be ready and returns.
598  *
599  * This function is called by the controller thread.
600  *
601  * \param parmap a parmap
602  */
603 static void xbt_parmap_busy_master_wait(xbt_parmap_t parmap)
604 {
605   while (parmap->thread_counter < parmap->num_workers) {
606     xbt_os_thread_yield();
607   }
608 }
609
610 /**
611  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
612  *
613  * This function is called by all worker threads when they end.
614  *
615  * \param parmap a parmap
616  */
617 static void xbt_parmap_busy_worker_signal(xbt_parmap_t parmap)
618 {
619   __sync_add_and_fetch(&parmap->thread_counter, 1);
620 }
621
622 /**
623  * \brief Wakes all workers and waits for them to finish the tasks.
624  *
625  * This function is called by the controller thread.
626  *
627  * \param parmap a parmap
628  */
629 static void xbt_parmap_busy_master_signal(xbt_parmap_t parmap)
630 {
631   parmap->thread_counter = 1;
632   __sync_add_and_fetch(&parmap->work, 1);
633 }
634
635 /**
636  * \brief Waits for some work to process.
637  *
638  * This function is called by each worker thread (not including the controller)
639  * when it has no more work to do.
640  *
641  * \param parmap a parmap
642  * \param round  the expected round number
643  */
644 static void xbt_parmap_busy_worker_wait(xbt_parmap_t parmap, unsigned round)
645 {
646   /* wait for more work */
647   while (parmap->work != round) {
648     xbt_os_thread_yield();
649   }
650 }