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