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 /**
  56  * \brief Parallel map structure
  57  */
  58 typedef struct s_xbt_parmap {
  59   e_xbt_parmap_flag_t status;      /**< is the parmap active or being destroyed? */
  60   int work;                        /**< index of the current round (1 is the first) */
  61   int done;                        /**< number of rounds already done (futexes only) */
  62   unsigned int thread_counter;     /**< number of threads currently working */
  63   unsigned int num_workers;        /**< total number of worker threads including the controller */
  64   void_f_pvoid_t fun;              /**< function to run in parallel on each element of data */
  65   xbt_dynar_t data;                /**< parameters to pass to fun in parallel */
  66   unsigned int index;              /**< index of the next element of data to pick */
  67
  68   /* fields that depend on the synchronization mode */
  69   e_xbt_parmap_mode_t mode;        /**< synchronization mode */
  70   void (*start_f)(xbt_parmap_t);   /**< initializes the worker threads */
  71   void (*end_f)(xbt_parmap_t);     /**< finalizes the worker threads */
  72   void (*signal_f)(xbt_parmap_t);  /**< wakes the workers threads to process tasks */
  73   void (*wait_f)(xbt_parmap_t);    /**< waits for more work */
  74 } s_xbt_parmap_t;
  75
  76 /**
  77  * \brief Creates a parallel map object
  78  * \param num_workers number of worker threads to create
  79  * \param mode how to synchronize the worker threads
  80  * \return the parmap created
  81  */
  82 xbt_parmap_t xbt_parmap_new(unsigned int num_workers, e_xbt_parmap_mode_t mode)
  83 {
  84   unsigned int i;
  85   xbt_os_thread_t worker = NULL;
  86
  87   XBT_DEBUG("Create new parmap (%u workers)", num_workers);
  88
  89   /* Initialize the thread pool data structure */
  90   xbt_parmap_t parmap = xbt_new0(s_xbt_parmap_t, 1);
  91
  92   parmap->num_workers = num_workers;
  93   parmap->status = XBT_PARMAP_WORK;
  94   xbt_parmap_set_mode(parmap, mode);
  95
  96   /* Create the pool of worker threads */
  97   for (i = 0; i < num_workers - 1; i++) {
  98     worker = xbt_os_thread_create(NULL, xbt_parmap_worker_main, parmap, NULL);
  99     xbt_os_thread_detach(worker);
 100   }
 101   parmap->start_f(parmap);
 102   return parmap;
 103 }
 104
 105 /**
 106  * \brief Destroys a parmap
 107  * \param parmap the parmap to destroy
 108  */
 109 void xbt_parmap_destroy(xbt_parmap_t parmap)
 110 {
 111   parmap->status = XBT_PARMAP_DESTROY;
 112   parmap->signal_f(parmap);
 113   xbt_free(parmap);
 114 }
 115
 116 /**
 117  * \brief Sets the synchronization mode of a parmap.
 118  * \param parmap a parallel map object
 119  * \param mode the synchronization mode
 120  */
 121 static void xbt_parmap_set_mode(xbt_parmap_t parmap, e_xbt_parmap_mode_t mode)
 122 {
 123   if (mode == XBT_PARMAP_DEFAULT) {
 124 #ifdef HAVE_FUTEX_H
 125     mode = XBT_PARMAP_FUTEX;
 126 #else
 127     //For now use busy wait because posix is unimplemented
 128     mode = XBT_PARMAP_BUSY_WAIT;
 129 #endif
 130   }
 131   parmap->mode = mode;
 132
 133   switch (mode) {
 134
 135     case XBT_PARMAP_POSIX:
 136       parmap->start_f = xbt_parmap_posix_start;
 137       parmap->end_f = xbt_parmap_posix_end;
 138       parmap->signal_f = xbt_parmap_posix_signal;
 139       parmap->wait_f = xbt_parmap_posix_wait;
 140       break;
 141
 142
 143     case XBT_PARMAP_FUTEX:
 144 #ifdef HAVE_FUTEX_H
 145       parmap->start_f = xbt_parmap_futex_start;
 146       parmap->end_f = xbt_parmap_futex_end;
 147       parmap->signal_f = xbt_parmap_futex_signal;
 148       parmap->wait_f = xbt_parmap_futex_wait;
 149       break;
 150 #else
 151       xbt_die("Futex is not available on this OS (maybe you are on a Mac).");
 152 #endif
 153
 154     case XBT_PARMAP_BUSY_WAIT:
 155       parmap->start_f = xbt_parmap_busy_start;
 156       parmap->end_f = xbt_parmap_busy_end;
 157       parmap->signal_f = xbt_parmap_busy_signal;
 158       parmap->wait_f = xbt_parmap_busy_wait;
 159       break;
 160
 161     case XBT_PARMAP_DEFAULT:
 162       THROW_IMPOSSIBLE;
 163       break;
 164   }
 165 }
 166
 167 /**
 168  * \brief Applies a list of tasks in parallel.
 169  * \param parmap a parallel map object
 170  * \param fun the function to call in parallel
 171  * \param data each element of this dynar will be passed as an argument to fun
 172  */
 173 void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
 174 {
 175   /* Assign resources to worker threads */
 176   parmap->fun = fun;
 177   parmap->data = data;
 178   parmap->index = 0;
 179   parmap->signal_f(parmap);
 180   XBT_DEBUG("Job done");
 181 }
 182
 183 /**
 184  * \brief Returns a next task to process.
 185  *
 186  * Worker threads call this function to get more work.
 187  *
 188  * \return the next task to process, or NULL if there is no more work
 189  */
 190 void* xbt_parmap_next(xbt_parmap_t parmap)
 191 {
 192   unsigned int index = __sync_fetch_and_add(&parmap->index, 1);
 193   if (index < xbt_dynar_length(parmap->data)) {
 194     return xbt_dynar_get_as(parmap->data, index, void*);
 195   }
 196   return NULL;
 197 }
 198
 199 /**
 200  * \brief Main function of a worker thread.
 201  * \param arg the parmap
 202  */
 203 static void *xbt_parmap_worker_main(void *arg)
 204 {
 205   xbt_parmap_t parmap = (xbt_parmap_t) arg;
 206
 207   XBT_DEBUG("New worker thread created");
 208
 209   /* Worker's main loop */
 210   while (1) {
 211     parmap->wait_f(parmap);
 212     if (parmap->status == XBT_PARMAP_WORK) {
 213
 214       XBT_DEBUG("Worker got a job");
 215
 216       void* work = xbt_parmap_next(parmap);
 217       while (work != NULL) {
 218         parmap->fun(work);
 219         work = xbt_parmap_next(parmap);
 220       }
 221
 222       XBT_DEBUG("Worker has finished");
 223
 224     /* We are destroying the parmap */
 225     } else {
 226       parmap->end_f(parmap);
 227       XBT_DEBUG("Shutting down worker");
 228       return NULL;
 229     }
 230   }
 231 }
 232
 233 #ifdef HAVE_FUTEX_H
 234 static void futex_wait(int *uaddr, int val)
 235 {
 236   XBT_VERB("Waiting on futex %p", uaddr);
 237   syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0);
 238 }
 239
 240 static void futex_wake(int *uaddr, int val)
 241 {
 242   XBT_VERB("Waking futex %p", uaddr);
 243   syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, NULL, NULL, 0);
 244 }
 245 #endif
 246
 247 static void xbt_parmap_posix_start(xbt_parmap_t parmap)
 248 {
 249   THROW_UNIMPLEMENTED;
 250 }
 251
 252 static void xbt_parmap_posix_end(xbt_parmap_t parmap)
 253 {
 254   THROW_UNIMPLEMENTED;
 255 }
 256
 257 static void xbt_parmap_posix_signal(xbt_parmap_t parmap)
 258 {
 259   THROW_UNIMPLEMENTED;
 260 }
 261
 262 static void xbt_parmap_posix_wait(xbt_parmap_t parmap)
 263 {
 264   THROW_UNIMPLEMENTED;
 265 }
 266
 267 #ifdef HAVE_FUTEX_H
 268 /**
 269  * \brief Starts the parmap: waits for all workers to be ready and returns.
 270  *
 271  * This function is called by the controller thread.
 272  *
 273  * \param parmap a parmap
 274  */
 275 static void xbt_parmap_futex_start(xbt_parmap_t parmap)
 276 {
 277   int myflag = parmap->done;
 278   __sync_fetch_and_add(&parmap->thread_counter, 1);
 279   if (parmap->thread_counter < parmap->num_workers) {
 280     /* wait for all workers to be ready */
 281     futex_wait(&parmap->done, myflag);
 282   }
 283 }
 284
 285 /**
 286  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
 287  *
 288  * This function is called by all worker threads when they end (not including
 289  * the controller).
 290  *
 291  * \param parmap a parmap
 292  */
 293 static void xbt_parmap_futex_end(xbt_parmap_t parmap)
 294 {
 295   unsigned int mycount;
 296
 297   mycount = __sync_add_and_fetch(&parmap->thread_counter, 1);
 298   if (mycount == parmap->num_workers) {
 299     /* all workers have finished, wake the controller */
 300     parmap->done++;
 301     futex_wake(&parmap->done, 1);
 302   }
 303 }
 304
 305 /**
 306  * \brief Wakes all workers and waits for them to finish the tasks.
 307  *
 308  * This function is called by the controller thread.
 309  *
 310  * \param parmap a parmap
 311  */
 312 static void xbt_parmap_futex_signal(xbt_parmap_t parmap)
 313 {
 314   int myflag = parmap->done;
 315   parmap->thread_counter = 0;
 316   parmap->work++;
 317
 318   /* wake all workers */
 319   futex_wake(&parmap->work, parmap->num_workers);
 320
 321   if (parmap->status == XBT_PARMAP_WORK) {
 322     /* also work myself */
 323     void* work = xbt_parmap_next(parmap);
 324     while (work != NULL) {
 325       parmap->fun(work);
 326       work = xbt_parmap_next(parmap);
 327     }
 328   }
 329
 330   unsigned int mycount = __sync_add_and_fetch(&parmap->thread_counter, 1);
 331   if (mycount < parmap->num_workers) {
 332     /* some workers have not finished yet */
 333     futex_wait(&parmap->done, myflag);
 334   }
 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  */
 345 static void xbt_parmap_futex_wait(xbt_parmap_t parmap)
 346 {
 347   int myflag;
 348   unsigned int mycount;
 349
 350   myflag = parmap->work;
 351   mycount = __sync_add_and_fetch(&parmap->thread_counter, 1);
 352   if (mycount == parmap->num_workers) {
 353     /* all workers have finished, wake the controller */
 354     parmap->done++;
 355     futex_wake(&parmap->done, 1);
 356   }
 357
 358   /* wait for more work */
 359   futex_wait(&parmap->work, myflag);
 360 }
 361 #endif
 362
 363 /**
 364  * \brief Starts the parmap: waits for all workers to be ready and returns.
 365  *
 366  * This function is called by the controller thread.
 367  *
 368  * \param parmap a parmap
 369  */
 370 static void xbt_parmap_busy_start(xbt_parmap_t parmap)
 371 {
 372   __sync_fetch_and_add(&parmap->thread_counter, 1);
 373   while (parmap->thread_counter < parmap->num_workers) {
 374     xbt_os_thread_yield();
 375   }
 376 }
 377
 378 /**
 379  * \brief Ends the parmap: wakes the controller thread when all workers terminate.
 380  *
 381  * This function is called by all worker threads when they end.
 382  *
 383  * \param parmap a parmap
 384  */
 385 static void xbt_parmap_busy_end(xbt_parmap_t parmap)
 386 {
 387   __sync_add_and_fetch(&parmap->thread_counter, 1);
 388 }
 389
 390 /**
 391  * \brief Wakes all workers and waits for them to finish the tasks.
 392  *
 393  * This function is called by the controller thread.
 394  *
 395  * \param parmap a parmap
 396  */
 397 static void xbt_parmap_busy_signal(xbt_parmap_t parmap)
 398 {
 399   parmap->thread_counter = 0;
 400   parmap->work++;
 401
 402   if (parmap->status == XBT_PARMAP_WORK) {
 403     /* also work myself */
 404     void* work = xbt_parmap_next(parmap);
 405     while (work != NULL) {
 406       parmap->fun(work);
 407       work = xbt_parmap_next(parmap);
 408     }
 409   }
 410
 411   /* I have finished, wait for the others */
 412   __sync_add_and_fetch(&parmap->thread_counter, 1);
 413   while (parmap->thread_counter < parmap->num_workers) {
 414     xbt_os_thread_yield();
 415   }
 416 }
 417
 418 /**
 419  * \brief Waits for some work to process.
 420  *
 421  * This function is called by each worker thread (not including the controller)
 422  * when it has no more work to do.
 423  *
 424  * \param parmap a parmap
 425  */
 426 static void xbt_parmap_busy_wait(xbt_parmap_t parmap)
 427 {
 428   int work = parmap->work;
 429   __sync_add_and_fetch(&parmap->thread_counter, 1);
 430
 431   /* wait for more work */
 432   while (parmap->work == work) {
 433     xbt_os_thread_yield();
 434   }
 435 }
 436
 437 #ifdef SIMGRID_TEST
 438 #include "xbt.h"
 439 #include "xbt/ex.h"
 440
 441 XBT_TEST_SUITE("parmap", "Parallel Map");
 442 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(xbt_parmap_unit);
 443
 444 xbt_parmap_t parmap;
 445
 446 void fun(void *arg);
 447
 448 void fun(void *arg)
 449 {
 450   //XBT_INFO("I'm job %lu", (unsigned long)arg);
 451 }
 452
 453 XBT_TEST_UNIT("basic", test_parmap_basic, "Basic usage")
 454 {
 455   xbt_test_add("Create the parmap");
 456
 457   unsigned long i, j;
 458   xbt_dynar_t data = xbt_dynar_new(sizeof(void *), NULL);
 459
 460   /* Create the parallel map */
 461 #ifdef HAVE_FUTEX_H
 462   parmap = xbt_parmap_new(10, XBT_PARMAP_FUTEX);
 463 #else
 464   parmap = xbt_parmap_new(10, XBT_PARMAP_BUSY_WAIT);
 465 #endif
 466   for (j = 0; j < 100; j++) {
 467     xbt_dynar_push_as(data, void *, (void *)j);
 468   }
 469
 470   for (i = 0; i < 5; i++) {
 471     xbt_parmap_apply(parmap, fun, data);
 472   }
 473
 474   /* Destroy the parmap */
 475   xbt_parmap_destroy(parmap);
 476   xbt_dynar_free(&data);
 477 }
 478
 479 #endif /* SIMGRID_TEST */