Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
62a691d0b5a4070a93351c0268daf60c90cfb5b5
[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 "src/internal_config.h"
10 #if HAVE_UNISTD_H
11 #include <unistd.h>
12 #endif
13
14 #ifndef _WIN32
15 #include <sys/syscall.h>
16 #endif
17
18 #if 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 "src/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 #if 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 /**
62  * \brief Parallel map structure
63  */
64 typedef struct s_xbt_parmap {
65   e_xbt_parmap_flag_t status;      /**< is the parmap active or being destroyed? */
66   unsigned work;                   /**< index of the current round */
67   unsigned thread_counter;         /**< number of workers that have done the work */
68
69   unsigned int num_workers;        /**< total number of worker threads including the controller */
70   xbt_os_thread_t *workers;        /**< worker thread handlers */
71   void_f_pvoid_t fun;              /**< function to run in parallel on each element of data */
72   xbt_dynar_t data;                /**< parameters to pass to fun in parallel */
73   std::atomic<unsigned int> index; /**< index of the next element of data to pick */
74
75   /* posix only */
76   xbt_os_cond_t ready_cond;
77   xbt_os_mutex_t ready_mutex;
78   xbt_os_cond_t done_cond;
79   xbt_os_mutex_t done_mutex;
80
81   /* fields that depend on the synchronization mode */
82   e_xbt_parmap_mode_t mode;        /**< synchronization mode */
83   void (*master_wait_f)(xbt_parmap_t);    /**< wait for the workers to have done the work */
84   void (*worker_signal_f)(xbt_parmap_t);  /**< signal the master that a worker has done the work */
85   void (*master_signal_f)(xbt_parmap_t);  /**< wakes the workers threads to process tasks */
86   void (*worker_wait_f)(xbt_parmap_t, unsigned); /**< waits for more work */
87 } s_xbt_parmap_t;
88
89 /**
90  * \brief Thread data transmission structure
91  */
92 typedef struct s_xbt_parmap_thread_data{
93   xbt_parmap_t parmap;
94   int worker_id;
95 } s_xbt_parmap_thread_data_t;
96
97 typedef s_xbt_parmap_thread_data_t *xbt_parmap_thread_data_t;
98
99 /**
100  * \brief Creates a parallel map object
101  * \param num_workers number of worker threads to create
102  * \param mode how to synchronize the worker threads
103  * \return the parmap created
104  */
105 xbt_parmap_t xbt_parmap_new(unsigned int num_workers, e_xbt_parmap_mode_t mode)
106 {
107   XBT_DEBUG("Create new parmap (%u workers)", num_workers);
108
109   /* Initialize the thread pool data structure */
110   xbt_parmap_t parmap = new s_xbt_parmap_t();
111   parmap->workers = xbt_new(xbt_os_thread_t, num_workers);
112
113   parmap->num_workers = num_workers;
114   parmap->status = XBT_PARMAP_WORK;
115   xbt_parmap_set_mode(parmap, mode);
116
117   /* Create the pool of worker threads */
118   xbt_parmap_thread_data_t data;
119   parmap->workers[0] = nullptr;
120 #if HAVE_PTHREAD_SETAFFINITY
121   int core_bind = 0;
122 #endif  
123   for (unsigned int i = 1; i < num_workers; i++) {
124     data = xbt_new0(s_xbt_parmap_thread_data_t, 1);
125     data->parmap = parmap;
126     data->worker_id = i;
127     parmap->workers[i] = xbt_os_thread_create(nullptr, xbt_parmap_worker_main, data, nullptr);
128 #if HAVE_PTHREAD_SETAFFINITY
129     xbt_os_thread_bind(parmap->workers[i], core_bind);
130     if (core_bind != xbt_os_get_numcores() - 1)
131       core_bind++;
132     else
133       core_bind = 0; 
134 #endif
135   }
136   return parmap;
137 }
138
139 /**
140  * \brief Destroys a parmap
141  * \param parmap the parmap to destroy
142  */
143 void xbt_parmap_destroy(xbt_parmap_t parmap)
144 {
145   if (!parmap) {
146     return;
147   }
148
149   parmap->status = XBT_PARMAP_DESTROY;
150   parmap->master_signal_f(parmap);
151
152   unsigned int i;
153   for (i = 1; i < parmap->num_workers; i++)
154     xbt_os_thread_join(parmap->workers[i], nullptr);
155
156   xbt_os_cond_destroy(parmap->ready_cond);
157   xbt_os_mutex_destroy(parmap->ready_mutex);
158   xbt_os_cond_destroy(parmap->done_cond);
159   xbt_os_mutex_destroy(parmap->done_mutex);
160
161   xbt_free(parmap->workers);
162   delete parmap;
163 }
164
165 /**
166  * \brief Sets the synchronization mode of a parmap.
167  * \param parmap a parallel map object
168  * \param mode the synchronization mode
169  */
170 static void xbt_parmap_set_mode(xbt_parmap_t parmap, e_xbt_parmap_mode_t mode)
171 {
172   if (mode == XBT_PARMAP_DEFAULT) {
173 #if HAVE_FUTEX_H
174     mode = XBT_PARMAP_FUTEX;
175 #else
176     mode = XBT_PARMAP_POSIX;
177 #endif
178   }
179   parmap->mode = mode;
180
181   switch (mode) {
182     case XBT_PARMAP_POSIX:
183       parmap->master_wait_f = xbt_parmap_posix_master_wait;
184       parmap->worker_signal_f = xbt_parmap_posix_worker_signal;
185       parmap->master_signal_f = xbt_parmap_posix_master_signal;
186       parmap->worker_wait_f = xbt_parmap_posix_worker_wait;
187
188       parmap->ready_cond = xbt_os_cond_init();
189       parmap->ready_mutex = xbt_os_mutex_init();
190       parmap->done_cond = xbt_os_cond_init();
191       parmap->done_mutex = xbt_os_mutex_init();
192       break;
193     case XBT_PARMAP_FUTEX:
194 #if HAVE_FUTEX_H
195       parmap->master_wait_f = xbt_parmap_futex_master_wait;
196       parmap->worker_signal_f = xbt_parmap_futex_worker_signal;
197       parmap->master_signal_f = xbt_parmap_futex_master_signal;
198       parmap->worker_wait_f = xbt_parmap_futex_worker_wait;
199
200       xbt_os_cond_destroy(parmap->ready_cond);
201       xbt_os_mutex_destroy(parmap->ready_mutex);
202       xbt_os_cond_destroy(parmap->done_cond);
203       xbt_os_mutex_destroy(parmap->done_mutex);
204       break;
205 #else
206       xbt_die("Futex is not available on this OS.");
207 #endif
208     case XBT_PARMAP_BUSY_WAIT:
209       parmap->master_wait_f = xbt_parmap_busy_master_wait;
210       parmap->worker_signal_f = xbt_parmap_busy_worker_signal;
211       parmap->master_signal_f = xbt_parmap_busy_master_signal;
212       parmap->worker_wait_f = xbt_parmap_busy_worker_wait;
213
214       xbt_os_cond_destroy(parmap->ready_cond);
215       xbt_os_mutex_destroy(parmap->ready_mutex);
216       xbt_os_cond_destroy(parmap->done_cond);
217       xbt_os_mutex_destroy(parmap->done_mutex);
218       break;
219     case XBT_PARMAP_DEFAULT:
220       THROW_IMPOSSIBLE;
221       break;
222   }
223 }
224
225 /**
226  * \brief Applies a list of tasks in parallel.
227  * \param parmap a parallel map object
228  * \param fun the function to call in parallel
229  * \param data each element of this dynar will be passed as an argument to fun
230  */
231 void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
232 {
233   /* Assign resources to worker threads (we are maestro here)*/
234   parmap->fun = fun;
235   parmap->data = data;
236   parmap->index = 0;
237   parmap->master_signal_f(parmap); // maestro runs futex_wait to wake all the minions (the working threads)
238   xbt_parmap_work(parmap);         // maestro works with its minions
239   parmap->master_wait_f(parmap);   // When there is no more work to do, then maestro waits for the last minion to stop
240   XBT_DEBUG("Job done");           //   ... and proceeds
241 }
242
243 /**
244  * \brief Returns a next task to process.
245  *
246  * Worker threads call this function to get more work.
247  *
248  * \return the next task to process, or nullptr if there is no more work
249  */
250 void* xbt_parmap_next(xbt_parmap_t parmap)
251 {
252   unsigned int index = parmap->index++;
253   if (index < xbt_dynar_length(parmap->data)) {
254     return xbt_dynar_get_as(parmap->data, index, void*);
255   }
256   return nullptr;
257 }
258
259 static void xbt_parmap_work(xbt_parmap_t parmap)
260 {
261   unsigned index;
262   while ((index = parmap->index++) < xbt_dynar_length(parmap->data))
263     parmap->fun(xbt_dynar_get_as(parmap->data, index, void*));
264 }
265
266 /**
267  * \brief Main function of a worker thread.
268  * \param arg the parmap
269  */
270 static void *xbt_parmap_worker_main(void *arg)
271 {
272   xbt_parmap_thread_data_t data = (xbt_parmap_thread_data_t) arg;
273   xbt_parmap_t parmap = data->parmap;
274   unsigned round = 0;
275   smx_context_t context = SIMIX_context_new(std::function<void()>(), nullptr, nullptr);
276   SIMIX_context_set_current(context);
277
278   XBT_DEBUG("New worker thread created");
279
280   /* Worker's main loop */
281   while (1) {
282     parmap->worker_wait_f(parmap, ++round);
283     if (parmap->status == XBT_PARMAP_WORK) {
284       XBT_DEBUG("Worker %d got a job", data->worker_id);
285
286       xbt_parmap_work(parmap);
287       parmap->worker_signal_f(parmap);
288
289       XBT_DEBUG("Worker %d has finished", data->worker_id);
290     /* We are destroying the parmap */
291     } else {
292       delete context;
293       xbt_free(data);
294       return nullptr;
295     }
296   }
297 }
298
299 #if HAVE_FUTEX_H
300 static void futex_wait(unsigned *uaddr, unsigned val)
301 {
302   XBT_VERB("Waiting on futex %p", uaddr);
303   syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, nullptr, nullptr, 0);
304 }
305
306 static void futex_wake(unsigned *uaddr, unsigned val)
307 {
308   XBT_VERB("Waking futex %p", uaddr);
309   syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, nullptr, nullptr, 0);
310 }
311 #endif
312
313 /**
314  * \brief Starts the parmap: waits for all workers to be ready and returns.
315  *
316  * This function is called by the controller thread.
317  *
318  * \param parmap a parmap
319  */
320 static void xbt_parmap_posix_master_wait(xbt_parmap_t parmap)
321 {
322   xbt_os_mutex_acquire(parmap->done_mutex);
323   if (parmap->thread_counter < parmap->num_workers) {
324     /* wait for all workers to be ready */
325     xbt_os_cond_wait(parmap->done_cond, parmap->done_mutex);
326   }
327   xbt_os_mutex_release(parmap->done_mutex);
328 }
329
330 /**
331  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
332  *
333  * This function is called by all worker threads when they end (not including the controller).
334  *
335  * \param parmap a parmap
336  */
337 static void xbt_parmap_posix_worker_signal(xbt_parmap_t parmap)
338 {
339   xbt_os_mutex_acquire(parmap->done_mutex);
340   if (++parmap->thread_counter == parmap->num_workers) {
341     /* all workers have finished, wake the controller */
342     xbt_os_cond_signal(parmap->done_cond);
343   }
344   xbt_os_mutex_release(parmap->done_mutex);
345 }
346
347 /**
348  * \brief Wakes all workers and waits for them to finish the tasks.
349  *
350  * This function is called by the controller thread.
351  *
352  * \param parmap a parmap
353  */
354 static void xbt_parmap_posix_master_signal(xbt_parmap_t parmap)
355 {
356   xbt_os_mutex_acquire(parmap->ready_mutex);
357   parmap->thread_counter = 1;
358   parmap->work++;
359   /* wake all workers */
360   xbt_os_cond_broadcast(parmap->ready_cond);
361   xbt_os_mutex_release(parmap->ready_mutex);
362 }
363
364 /**
365  * \brief Waits for some work to process.
366  *
367  * This function is called by each worker thread (not including the controller) when it has no more work to do.
368  *
369  * \param parmap a parmap
370  * \param round  the expected round number
371  */
372 static void xbt_parmap_posix_worker_wait(xbt_parmap_t parmap, unsigned round)
373 {
374   xbt_os_mutex_acquire(parmap->ready_mutex);
375   /* wait for more work */
376   if (parmap->work != round) {
377     xbt_os_cond_wait(parmap->ready_cond, parmap->ready_mutex);
378   }
379   xbt_os_mutex_release(parmap->ready_mutex);
380 }
381
382 #if HAVE_FUTEX_H
383 /**
384  * \brief Starts the parmap: waits for all workers to be ready and returns.
385  *
386  * This function is called by the controller thread.
387  *
388  * \param parmap a parmap
389  */
390 static void xbt_parmap_futex_master_wait(xbt_parmap_t parmap)
391 {
392   unsigned count = parmap->thread_counter;
393   while (count < parmap->num_workers) {
394     /* wait for all workers to be ready */
395     futex_wait(&parmap->thread_counter, count);
396     count = parmap->thread_counter;
397   }
398 }
399
400 /**
401  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
402  *
403  * This function is called by all worker threads when they end (not including the controller).
404  *
405  * \param parmap a parmap
406  */
407 static void xbt_parmap_futex_worker_signal(xbt_parmap_t parmap)
408 {
409   unsigned count = __sync_add_and_fetch(&parmap->thread_counter, 1);
410   if (count == parmap->num_workers) {
411     /* all workers have finished, wake the controller */
412     futex_wake(&parmap->thread_counter, INT_MAX);
413   }
414 }
415
416 /**
417  * \brief Wakes all workers and waits for them to finish the tasks.
418  *
419  * This function is called by the controller thread.
420  *
421  * \param parmap a parmap
422  */
423 static void xbt_parmap_futex_master_signal(xbt_parmap_t parmap)
424 {
425   parmap->thread_counter = 1;
426   __sync_add_and_fetch(&parmap->work, 1);
427   /* wake all workers */
428   futex_wake(&parmap->work, INT_MAX);
429 }
430
431 /**
432  * \brief Waits for some work to process.
433  *
434  * This function is called by each worker thread (not including the controller) when it has no more work to do.
435  *
436  * \param parmap a parmap
437  * \param round  the expected round number
438  */
439 static void xbt_parmap_futex_worker_wait(xbt_parmap_t parmap, unsigned round)
440 {
441   unsigned work = parmap->work;
442   /* wait for more work */
443   while (work != round) {
444     futex_wait(&parmap->work, work);
445     work = parmap->work;
446   }
447 }
448 #endif
449
450 /**
451  * \brief Starts the parmap: waits for all workers to be ready and returns.
452  *
453  * This function is called by the controller thread.
454  *
455  * \param parmap a parmap
456  */
457 static void xbt_parmap_busy_master_wait(xbt_parmap_t parmap)
458 {
459   while (parmap->thread_counter < parmap->num_workers) {
460     xbt_os_thread_yield();
461   }
462 }
463
464 /**
465  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
466  *
467  * This function is called by all worker threads when they end.
468  *
469  * \param parmap a parmap
470  */
471 static void xbt_parmap_busy_worker_signal(xbt_parmap_t parmap)
472 {
473   __sync_add_and_fetch(&parmap->thread_counter, 1);
474 }
475
476 /**
477  * \brief Wakes all workers and waits for them to finish the tasks.
478  *
479  * This function is called by the controller thread.
480  *
481  * \param parmap a parmap
482  */
483 static void xbt_parmap_busy_master_signal(xbt_parmap_t parmap)
484 {
485   parmap->thread_counter = 1;
486   __sync_add_and_fetch(&parmap->work, 1);
487 }
488
489 /**
490  * \brief Waits for some work to process.
491  *
492  * This function is called by each worker thread (not including the controller) when it has no more work to do.
493  *
494  * \param parmap a parmap
495  * \param round  the expected round number
496  */
497 static void xbt_parmap_busy_worker_wait(xbt_parmap_t parmap, unsigned round)
498 {
499   /* wait for more work */
500   while (parmap->work != round) {
501     xbt_os_thread_yield();
502   }
503 }