src/xbt/parmap.c

   1 /* Copyright (c) 2004, 2005, 2007, 2009, 2010. 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 #include "gras_config.h"
   7 #include <unistd.h>
   8
   9 #ifndef _XBT_WIN32
  10 #include <sys/syscall.h>
  11 #endif
  12
  13 #ifdef HAVE_FUTEX_H
  14 #include <linux/futex.h>
  15 #endif
  16
  17 #include "xbt/parmap.h"
  18 #include "xbt/log.h"
  19 #include "xbt/function_types.h"
  20 #include "xbt/dynar.h"
  21 #include "xbt/xbt_os_thread.h"
  22 #include "xbt/sysdep.h"
  23
  24 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_parmap, xbt, "parmap: parallel map");
  25 XBT_LOG_NEW_SUBCATEGORY(xbt_parmap_unit, xbt_parmap, "parmap unit testing");
  26
  27 typedef enum {
  28   XBT_PARMAP_WORK = 0,
  29   XBT_PARMAP_DESTROY
  30 } e_xbt_parmap_flag_t;
  31
  32 static void xbt_parmap_set_mode(xbt_parmap_t parmap, e_xbt_parmap_mode_t mode);
  33 static void *xbt_parmap_worker_main(void *parmap);
  34
  35 static void xbt_parmap_posix_start(xbt_parmap_t parmap);
  36 static void xbt_parmap_posix_end(xbt_parmap_t parmap);
  37 static void xbt_parmap_posix_signal(xbt_parmap_t parmap);
  38 static void xbt_parmap_posix_wait(xbt_parmap_t parmap);
  39
  40 #ifdef HAVE_FUTEX_H
  41 static void xbt_parmap_futex_start(xbt_parmap_t parmap);
  42 static void xbt_parmap_futex_end(xbt_parmap_t parmap);
  43 static void xbt_parmap_futex_signal(xbt_parmap_t parmap);
  44 static void xbt_parmap_futex_wait(xbt_parmap_t parmap);
  45 static void futex_wait(int *uaddr, int val);
  46 static void futex_wake(int *uaddr, int val);
  47 #endif
  48
  49 static void xbt_parmap_busy_start(xbt_parmap_t parmap);
  50 static void xbt_parmap_busy_end(xbt_parmap_t parmap);
  51 static void xbt_parmap_busy_signal(xbt_parmap_t parmap);
  52 static void xbt_parmap_busy_wait(xbt_parmap_t parmap);
  53
  54 /**
  55  * \brief Parallel map structure
  56  */
  57 typedef struct s_xbt_parmap {
  58   e_xbt_parmap_flag_t status;      /**< is the parmap active or being destroyed? */
  59   int work;                        /**< index of the current round (1 is the first) */
  60   int done;                        /**< number of rounds already done (futexes only) */
  61   unsigned int thread_counter;     /**< number of threads currently working */
  62   unsigned int num_workers;        /**< total number of worker threads including the controller */
  63   void_f_pvoid_t fun;              /**< function to run in parallel on each element of data */
  64   xbt_dynar_t data;                /**< parameters to pass to fun in parallel */
  65   unsigned int index;              /**< index of the next element of data to pick */
  66
  67   /* posix only */
  68   xbt_os_cond_t ready_cond;
  69   xbt_os_mutex_t ready_mutex;
  70   xbt_os_cond_t done_cond;
  71   xbt_os_mutex_t done_mutex;
  72
  73   /* fields that depend on the synchronization mode */
  74   e_xbt_parmap_mode_t mode;        /**< synchronization mode */
  75   void (*start_f)(xbt_parmap_t);   /**< initializes the worker threads */
  76   void (*end_f)(xbt_parmap_t);     /**< finalizes the worker threads */
  77   void (*signal_f)(xbt_parmap_t);  /**< wakes the workers threads to process tasks */
  78   void (*wait_f)(xbt_parmap_t);    /**< waits for more work */
  79 } s_xbt_parmap_t;
  80
  81 /**
  82  * \brief Creates a parallel map object
  83  * \param num_workers number of worker threads to create
  84  * \param mode how to synchronize the worker threads
  85  * \return the parmap created
  86  */
  87 xbt_parmap_t xbt_parmap_new(unsigned int num_workers, e_xbt_parmap_mode_t mode)
  88 {
  89   unsigned int i;
  90   xbt_os_thread_t worker = NULL;
  91
  92   XBT_DEBUG("Create new parmap (%u workers)", num_workers);
  93
  94   /* Initialize the thread pool data structure */
  95   xbt_parmap_t parmap = xbt_new0(s_xbt_parmap_t, 1);
  96
  97   parmap->num_workers = num_workers;
  98   parmap->status = XBT_PARMAP_WORK;
  99   xbt_parmap_set_mode(parmap, mode);
 100
 101   /* Create the pool of worker threads */
 102   for (i = 0; i < num_workers - 1; i++) {
 103     worker = xbt_os_thread_create(NULL, xbt_parmap_worker_main, parmap, NULL);
 104     xbt_os_thread_detach(worker);
 105   }
 106   parmap->start_f(parmap);
 107   return parmap;
 108 }
 109
 110 /**
 111  * \brief Destroys a parmap
 112  * \param parmap the parmap to destroy
 113  */
 114 void xbt_parmap_destroy(xbt_parmap_t parmap)
 115 {
 116   if (!parmap) {
 117     return;
 118   }
 119
 120   parmap->status = XBT_PARMAP_DESTROY;
 121   parmap->signal_f(parmap);
 122
 123   xbt_os_cond_destroy(parmap->ready_cond);
 124   xbt_os_mutex_destroy(parmap->ready_mutex);
 125   xbt_os_cond_destroy(parmap->done_cond);
 126   xbt_os_mutex_destroy(parmap->done_mutex);
 127
 128   xbt_free(parmap);
 129 }
 130
 131 /**
 132  * \brief Sets the synchronization mode of a parmap.
 133  * \param parmap a parallel map object
 134  * \param mode the synchronization mode
 135  */
 136 static void xbt_parmap_set_mode(xbt_parmap_t parmap, e_xbt_parmap_mode_t mode)
 137 {
 138   if (mode == XBT_PARMAP_DEFAULT) {
 139 #ifdef HAVE_FUTEX_H
 140     mode = XBT_PARMAP_FUTEX;
 141 #else
 142     mode = XBT_PARMAP_POSIX;
 143 #endif
 144   }
 145   parmap->mode = mode;
 146
 147   switch (mode) {
 148
 149     case XBT_PARMAP_POSIX:
 150       parmap->start_f = xbt_parmap_posix_start;
 151       parmap->end_f = xbt_parmap_posix_end;
 152       parmap->signal_f = xbt_parmap_posix_signal;
 153       parmap->wait_f = xbt_parmap_posix_wait;
 154
 155       parmap->ready_cond = xbt_os_cond_init();
 156       parmap->ready_mutex = xbt_os_mutex_init();
 157       parmap->done_cond = xbt_os_cond_init();
 158       parmap->done_mutex = xbt_os_mutex_init();
 159       break;
 160
 161
 162     case XBT_PARMAP_FUTEX:
 163 #ifdef HAVE_FUTEX_H
 164       parmap->start_f = xbt_parmap_futex_start;
 165       parmap->end_f = xbt_parmap_futex_end;
 166       parmap->signal_f = xbt_parmap_futex_signal;
 167       parmap->wait_f = xbt_parmap_futex_wait;
 168
 169       xbt_os_cond_destroy(parmap->ready_cond);
 170       xbt_os_mutex_destroy(parmap->ready_mutex);
 171       xbt_os_cond_destroy(parmap->done_cond);
 172       xbt_os_mutex_destroy(parmap->done_mutex);
 173       break;
 174 #else
 175       xbt_die("Futex is not available on this OS.");
 176 #endif
 177
 178     case XBT_PARMAP_BUSY_WAIT:
 179       parmap->start_f = xbt_parmap_busy_start;
 180       parmap->end_f = xbt_parmap_busy_end;
 181       parmap->signal_f = xbt_parmap_busy_signal;
 182       parmap->wait_f = xbt_parmap_busy_wait;
 183
 184       xbt_os_cond_destroy(parmap->ready_cond);
 185       xbt_os_mutex_destroy(parmap->ready_mutex);
 186       xbt_os_cond_destroy(parmap->done_cond);
 187       xbt_os_mutex_destroy(parmap->done_mutex);
 188       break;
 189
 190     case XBT_PARMAP_DEFAULT:
 191       THROW_IMPOSSIBLE;
 192       break;
 193   }
 194 }
 195
 196 /**
 197  * \brief Applies a list of tasks in parallel.
 198  * \param parmap a parallel map object
 199  * \param fun the function to call in parallel
 200  * \param data each element of this dynar will be passed as an argument to fun
 201  */
 202 void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
 203 {
 204   /* Assign resources to worker threads */
 205   parmap->fun = fun;
 206   parmap->data = data;
 207   parmap->index = 0;
 208   parmap->signal_f(parmap);
 209   XBT_DEBUG("Job done");
 210 }
 211
 212 /**
 213  * \brief Returns a next task to process.
 214  *
 215  * Worker threads call this function to get more work.
 216  *
 217  * \return the next task to process, or NULL if there is no more work
 218  */
 219 void* xbt_parmap_next(xbt_parmap_t parmap)
 220 {
 221   unsigned int index = __sync_fetch_and_add(&parmap->index, 1);
 222   if (index < xbt_dynar_length(parmap->data)) {
 223     return xbt_dynar_get_as(parmap->data, index, void*);
 224   }
 225   return NULL;
 226 }
 227
 228 /**
 229  * \brief Main function of a worker thread.
 230  * \param arg the parmap
 231  */
 232 static void *xbt_parmap_worker_main(void *arg)
 233 {
 234   xbt_parmap_t parmap = (xbt_parmap_t) arg;
 235
 236   XBT_DEBUG("New worker thread created");
 237
 238   /* Worker's main loop */
 239   while (1) {
 240     parmap->wait_f(parmap);
 241     if (parmap->status == XBT_PARMAP_WORK) {
 242
 243       XBT_DEBUG("Worker got a job");
 244
 245       void* work = xbt_parmap_next(parmap);
 246       while (work != NULL) {
 247         parmap->fun(work);
 248         work = xbt_parmap_next(parmap);
 249       }
 250
 251       XBT_DEBUG("Worker has finished");
 252
 253     /* We are destroying the parmap */
 254     } else {
 255       parmap->end_f(parmap);
 256       return NULL;
 257     }
 258   }
 259 }
 260
 261 #ifdef HAVE_FUTEX_H
 262 static void futex_wait(int *uaddr, int val)
 263 {
 264   XBT_VERB("Waiting on futex %p", uaddr);
 265   syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0);
 266 }
 267
 268 static void futex_wake(int *uaddr, int val)
 269 {
 270   XBT_VERB("Waking futex %p", uaddr);
 271   syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, NULL, NULL, 0);
 272 }
 273 #endif
 274
 275 /**
 276  * \brief Starts the parmap: waits for all workers to be ready and returns.
 277  *
 278  * This function is called by the controller thread.
 279  *
 280  * \param parmap a parmap
 281  */
 282 static void xbt_parmap_posix_start(xbt_parmap_t parmap)
 283 {
 284   unsigned int counter = __sync_fetch_and_add(&parmap->thread_counter, 1);
 285   if (counter < parmap->num_workers) {
 286     /* wait for all workers to be initialized */
 287     xbt_os_cond_wait(parmap->done_cond, parmap->done_mutex);
 288   }
 289 }
 290
 291 /**
 292  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
 293  *
 294  * This function is called by all worker threads when they end (not including
 295  * the controller).
 296  *
 297  * \param parmap a parmap
 298  */
 299 static void xbt_parmap_posix_end(xbt_parmap_t parmap)
 300 {
 301   unsigned int counter = __sync_add_and_fetch(&parmap->thread_counter, 1);
 302   XBT_DEBUG("Shutting down worker %d", counter);
 303   if (counter == parmap->num_workers) {
 304     /* all workers have finished, wake the controller */
 305     xbt_os_cond_signal(parmap->done_cond);
 306   }
 307 }
 308
 309 /**
 310  * \brief Wakes all workers and waits for them to finish the tasks.
 311  *
 312  * This function is called by the controller thread.
 313  *
 314  * \param parmap a parmap
 315  */
 316 static void xbt_parmap_posix_signal(xbt_parmap_t parmap)
 317 {
 318   parmap->thread_counter = 0;
 319   parmap->work++;
 320   XBT_DEBUG("Starting work %d", parmap->work);
 321
 322   /* wake all workers */
 323   xbt_os_cond_broadcast(parmap->ready_cond);
 324
 325   if (parmap->status == XBT_PARMAP_WORK) {
 326     /* also work myself */
 327     void* work = xbt_parmap_next(parmap);
 328     while (work != NULL) {
 329       parmap->fun(work);
 330       work = xbt_parmap_next(parmap);
 331     }
 332   }
 333
 334   unsigned int counter = __sync_add_and_fetch(&parmap->thread_counter, 1);
 335   if (counter < parmap->num_workers) {
 336     /* some workers have not finished yet */
 337     XBT_DEBUG("Some workers have not finished yet, waiting for them");
 338     xbt_os_cond_wait(parmap->done_cond, parmap->done_mutex);
 339   }
 340 }
 341
 342 /**
 343  * \brief Waits for some work to process.
 344  *
 345  * This function is called by each worker thread (not including the controller)
 346  * when it has no more work to do.
 347  *
 348  * \param parmap a parmap
 349  */
 350 static void xbt_parmap_posix_wait(xbt_parmap_t parmap)
 351 {
 352   int work = parmap->work;
 353   unsigned int counter = __sync_add_and_fetch(&parmap->thread_counter, 1);
 354   if (counter == parmap->num_workers) {
 355     /* all workers have finished, wake the controller */
 356     parmap->done++;
 357     XBT_DEBUG("Last worker has finished, waking the controller");
 358     xbt_os_cond_signal(parmap->done_cond);
 359   }
 360
 361   /* wait for more work */
 362   XBT_DEBUG("Worker %d waiting for more work", counter);
 363   if (parmap->work == work) {
 364     xbt_os_cond_wait(parmap->ready_cond, parmap->ready_mutex);
 365   }
 366 }
 367
 368 #ifdef HAVE_FUTEX_H
 369 /**
 370  * \brief Starts the parmap: waits for all workers to be ready and returns.
 371  *
 372  * This function is called by the controller thread.
 373  *
 374  * \param parmap a parmap
 375  */
 376 static void xbt_parmap_futex_start(xbt_parmap_t parmap)
 377 {
 378   int myflag = parmap->done;
 379   __sync_fetch_and_add(&parmap->thread_counter, 1);
 380   if (parmap->thread_counter < parmap->num_workers) {
 381     /* wait for all workers to be ready */
 382     futex_wait(&parmap->done, myflag);
 383   }
 384 }
 385
 386 /**
 387  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
 388  *
 389  * This function is called by all worker threads when they end (not including
 390  * the controller).
 391  *
 392  * \param parmap a parmap
 393  */
 394 static void xbt_parmap_futex_end(xbt_parmap_t parmap)
 395 {
 396   unsigned int mycount;
 397
 398   mycount = __sync_add_and_fetch(&parmap->thread_counter, 1);
 399   if (mycount == parmap->num_workers) {
 400     /* all workers have finished, wake the controller */
 401     parmap->done++;
 402     futex_wake(&parmap->done, 1);
 403   }
 404 }
 405
 406 /**
 407  * \brief Wakes all workers and waits for them to finish the tasks.
 408  *
 409  * This function is called by the controller thread.
 410  *
 411  * \param parmap a parmap
 412  */
 413 static void xbt_parmap_futex_signal(xbt_parmap_t parmap)
 414 {
 415   int myflag = parmap->done;
 416   parmap->thread_counter = 0;
 417   parmap->work++;
 418
 419   /* wake all workers */
 420   futex_wake(&parmap->work, parmap->num_workers);
 421
 422   if (parmap->status == XBT_PARMAP_WORK) {
 423     /* also work myself */
 424     void* work = xbt_parmap_next(parmap);
 425     while (work != NULL) {
 426       parmap->fun(work);
 427       work = xbt_parmap_next(parmap);
 428     }
 429   }
 430
 431   unsigned int mycount = __sync_add_and_fetch(&parmap->thread_counter, 1);
 432   if (mycount < parmap->num_workers) {
 433     /* some workers have not finished yet */
 434     futex_wait(&parmap->done, myflag);
 435   }
 436 }
 437
 438 /**
 439  * \brief Waits for some work to process.
 440  *
 441  * This function is called by each worker thread (not including the controller)
 442  * when it has no more work to do.
 443  *
 444  * \param parmap a parmap
 445  */
 446 static void xbt_parmap_futex_wait(xbt_parmap_t parmap)
 447 {
 448   int myflag;
 449   unsigned int mycount;
 450
 451   myflag = parmap->work;
 452   mycount = __sync_add_and_fetch(&parmap->thread_counter, 1);
 453   if (mycount == parmap->num_workers) {
 454     /* all workers have finished, wake the controller */
 455     parmap->done++;
 456     futex_wake(&parmap->done, 1);
 457   }
 458
 459   /* wait for more work */
 460   futex_wait(&parmap->work, myflag);
 461 }
 462 #endif
 463
 464 /**
 465  * \brief Starts the parmap: waits for all workers to be ready and returns.
 466  *
 467  * This function is called by the controller thread.
 468  *
 469  * \param parmap a parmap
 470  */
 471 static void xbt_parmap_busy_start(xbt_parmap_t parmap)
 472 {
 473   __sync_fetch_and_add(&parmap->thread_counter, 1);
 474   while (parmap->thread_counter < parmap->num_workers) {
 475     xbt_os_thread_yield();
 476   }
 477 }
 478
 479 /**
 480  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
 481  *
 482  * This function is called by all worker threads when they end.
 483  *
 484  * \param parmap a parmap
 485  */
 486 static void xbt_parmap_busy_end(xbt_parmap_t parmap)
 487 {
 488   __sync_add_and_fetch(&parmap->thread_counter, 1);
 489 }
 490
 491 /**
 492  * \brief Wakes all workers and waits for them to finish the tasks.
 493  *
 494  * This function is called by the controller thread.
 495  *
 496  * \param parmap a parmap
 497  */
 498 static void xbt_parmap_busy_signal(xbt_parmap_t parmap)
 499 {
 500   parmap->thread_counter = 0;
 501   parmap->work++;
 502
 503   if (parmap->status == XBT_PARMAP_WORK) {
 504     /* also work myself */
 505     void* work = xbt_parmap_next(parmap);
 506     while (work != NULL) {
 507       parmap->fun(work);
 508       work = xbt_parmap_next(parmap);
 509     }
 510   }
 511
 512   /* I have finished, wait for the others */
 513   __sync_add_and_fetch(&parmap->thread_counter, 1);
 514   while (parmap->thread_counter < parmap->num_workers) {
 515     xbt_os_thread_yield();
 516   }
 517 }
 518
 519 /**
 520  * \brief Waits for some work to process.
 521  *
 522  * This function is called by each worker thread (not including the controller)
 523  * when it has no more work to do.
 524  *
 525  * \param parmap a parmap
 526  */
 527 static void xbt_parmap_busy_wait(xbt_parmap_t parmap)
 528 {
 529   int work = parmap->work;
 530   __sync_add_and_fetch(&parmap->thread_counter, 1);
 531
 532   /* wait for more work */
 533   while (parmap->work == work) {
 534     xbt_os_thread_yield();
 535   }
 536 }
 537
 538 #ifdef SIMGRID_TEST
 539 #include "xbt.h"
 540 #include "xbt/ex.h"
 541
 542 XBT_TEST_SUITE("parmap", "Parallel Map");
 543 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(xbt_parmap_unit);
 544
 545 xbt_parmap_t parmap;
 546
 547 void fun(void *arg);
 548
 549 void fun(void *arg)
 550 {
 551   //XBT_INFO("I'm job %lu", (unsigned long)arg);
 552 }
 553
 554 XBT_TEST_UNIT("basic", test_parmap_basic, "Basic usage")
 555 {
 556   xbt_test_add("Create the parmap");
 557
 558   unsigned long i, j;
 559   xbt_dynar_t data = xbt_dynar_new(sizeof(void *), NULL);
 560
 561   /* Create the parallel map */
 562 #ifdef HAVE_FUTEX_H
 563   parmap = xbt_parmap_new(10, XBT_PARMAP_FUTEX);
 564 #else
 565   parmap = xbt_parmap_new(10, XBT_PARMAP_BUSY_WAIT);
 566 #endif
 567   for (j = 0; j < 100; j++) {
 568     xbt_dynar_push_as(data, void *, (void *)j);
 569   }
 570
 571   for (i = 0; i < 5; i++) {
 572     xbt_parmap_apply(parmap, fun, data);
 573   }
 574
 575   /* Destroy the parmap */
 576   xbt_parmap_destroy(parmap);
 577   xbt_dynar_free(&data);
 578 }
 579
 580 #endif /* SIMGRID_TEST */