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