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 #ifndef _XBT_WIN32
   9 #include <sys/syscall.h>
  10 #endif
  11
  12 #ifdef HAVE_FUTEX_H
  13         #include <linux/futex.h>
  14 #else
  15         #include "xbt/xbt_os_thread.h"
  16 #endif
  17 #include <errno.h>
  18 #include "parmap_private.h"
  19
  20 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_parmap, xbt, "parmap: parallel map");
  21 XBT_LOG_NEW_SUBCATEGORY(xbt_parmap_unit, xbt_parmap, "parmap unit testing");
  22
  23 static void *_xbt_parmap_worker_main(void *parmap);
  24 #ifdef HAVE_FUTEX_H
  25         static void futex_wait(int *uaddr, int val);
  26         static void futex_wake(int *uaddr, int val);
  27 #endif
  28 xbt_parmap_t xbt_parmap_new(unsigned int num_workers)
  29 {
  30   unsigned int i;
  31   xbt_os_thread_t worker = NULL;
  32
  33   XBT_DEBUG("Create new parmap (%u workers)", num_workers);
  34
  35   /* Initialize the thread pool data structure */
  36   xbt_parmap_t parmap = xbt_new0(s_xbt_parmap_t, 1);
  37 #ifdef HAVE_FUTEX_H
  38   parmap->sync_event = xbt_new0(s_xbt_event_t, 1);
  39 #endif
  40   parmap->num_workers = num_workers;
  41   parmap->status = PARMAP_WORK;
  42 #ifdef HAVE_FUTEX_H
  43   parmap->sync_event->threads_to_wait = num_workers;
  44 #endif
  45   /* Create the pool of worker threads */
  46   for(i=0; i < num_workers; i++){
  47     worker = xbt_os_thread_create(NULL, _xbt_parmap_worker_main, parmap, NULL);
  48     xbt_os_thread_detach(worker);
  49   }
  50 #ifdef HAVE_FUTEX_H
  51   xbt_event_init(parmap->sync_event);
  52 #endif
  53   return parmap;
  54 }
  55
  56 void xbt_parmap_destroy(xbt_parmap_t parmap)
  57 {
  58   parmap->status = PARMAP_DESTROY;
  59 #ifdef HAVE_FUTEX_H
  60   xbt_event_signal(parmap->sync_event);
  61   xbt_free(parmap->sync_event);
  62 #endif
  63   xbt_free(parmap);
  64 }
  65
  66  void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
  67 {
  68   /* Assign resources to worker threads*/
  69   parmap->fun = fun;
  70   parmap->data = data;
  71   parmap->index = 0;
  72 #ifdef HAVE_FUTEX_H
  73   xbt_event_signal(parmap->sync_event);
  74 #endif
  75   XBT_DEBUG("Job done");
  76 }
  77
  78 void* xbt_parmap_next(xbt_parmap_t parmap) {
  79
  80   unsigned int index = __sync_fetch_and_add(&parmap->index, 1);
  81   if (index < xbt_dynar_length(parmap->data)) {
  82     return xbt_dynar_get_as(parmap->data, index, void*);
  83   }
  84   return NULL;
  85 }
  86
  87 unsigned long xbt_parmap_get_worker_id(xbt_parmap_t parmap) {
  88   return (unsigned long) xbt_os_thread_get_extra_data();
  89 }
  90
  91 static void *_xbt_parmap_worker_main(void *arg)
  92 {
  93   unsigned int worker_id;
  94   xbt_parmap_t parmap = (xbt_parmap_t) arg;
  95
  96   /* Fetch a worker id */
  97   worker_id = __sync_fetch_and_add(&parmap->workers_max_id, 1);
  98   xbt_os_thread_set_extra_data((void*) (unsigned long) worker_id);
  99
 100   XBT_DEBUG("New worker thread created (%u)", worker_id);
 101
 102   /* Worker's main loop */
 103   while (1) {
 104 #ifdef HAVE_FUTEX_H
 105     xbt_event_wait(parmap->sync_event);
 106 #endif
 107     if (parmap->status == PARMAP_WORK) {
 108
 109       XBT_DEBUG("Worker %u got a job", worker_id);
 110
 111       void* work = xbt_parmap_next(parmap);
 112       if (work != NULL) {
 113         parmap->fun(work);
 114       }
 115
 116       XBT_DEBUG("Worker %u has finished", worker_id);
 117
 118     /* We are destroying the parmap */
 119     } else {
 120 #ifdef HAVE_FUTEX_H
 121       xbt_event_end(parmap->sync_event);
 122 #endif
 123       XBT_DEBUG("Shutting down worker %u", worker_id);
 124       return NULL;
 125     }
 126   }
 127 }
 128
 129 #ifdef HAVE_FUTEX_H
 130 static void futex_wait(int *uaddr, int val)
 131 {
 132   XBT_VERB("Waiting on futex %p", uaddr);
 133   syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0);
 134 }
 135
 136 static void futex_wake(int *uaddr, int val)
 137 {
 138   XBT_VERB("Waking futex %p", uaddr);
 139   syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, NULL, NULL, 0);
 140 }
 141
 142 void xbt_event_init(xbt_event_t event)
 143 {
 144   int myflag = event->done;
 145   if(event->thread_counter < event->threads_to_wait)
 146     futex_wait(&event->done, myflag);
 147 }
 148
 149 void xbt_event_signal(xbt_event_t event)
 150 {
 151   int myflag = event->done;
 152   event->thread_counter = 0;
 153   event->work++;
 154   futex_wake(&event->work, event->threads_to_wait);
 155   futex_wait(&event->done, myflag);
 156 }
 157
 158 void xbt_event_wait(xbt_event_t event)
 159 {
 160   int myflag;
 161   unsigned int mycount;
 162
 163   myflag = event->work;
 164   mycount = __sync_add_and_fetch(&event->thread_counter, 1);
 165   if(mycount == event->threads_to_wait){
 166     event->done++;
 167     futex_wake(&event->done, 1);
 168   }
 169
 170   futex_wait(&event->work, myflag);
 171 }
 172
 173 void xbt_event_end(xbt_event_t event)
 174 {
 175   unsigned int mycount;
 176
 177   mycount = __sync_add_and_fetch(&event->thread_counter, 1);
 178   if(mycount == event->threads_to_wait){
 179     event->done++;
 180     futex_wake(&event->done, 1);
 181   }
 182 }
 183 #endif
 184
 185 #ifdef SIMGRID_TEST
 186 #include "xbt.h"
 187 #include "xbt/ex.h"
 188
 189 XBT_TEST_SUITE("parmap", "Parallel Map");
 190 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(xbt_parmap_unit);
 191
 192
 193
 194 xbt_parmap_t parmap;
 195
 196 void fun(void *arg);
 197
 198 void fun(void *arg)
 199 {
 200   //XBT_INFO("I'm job %lu", (unsigned long)arg);
 201 }
 202
 203 XBT_TEST_UNIT("basic", test_parmap_basic, "Basic usage")
 204 {
 205   xbt_test_add("Create the parmap");
 206
 207   unsigned long i,j;
 208   xbt_dynar_t data = xbt_dynar_new(sizeof(void *), NULL);
 209
 210   /* Create the parallel map */
 211   parmap = xbt_parmap_new(10);
 212
 213   for(j=0; j < 100; j++){
 214     xbt_dynar_push_as(data, void *, (void *)j);
 215   }
 216
 217   for(i=0; i < 5; i++)
 218     xbt_parmap_apply(parmap, fun, data);
 219
 220   /* Destroy the parmap */
 221   xbt_parmap_destroy(parmap);
 222   xbt_dynar_free(&data);
 223 }
 224
 225 #endif /* SIMGRID_TEST */