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 <unistd.h>
   7 #include <sys/syscall.h>
   8 #include <linux/futex.h>
   9 #include <errno.h>
  10 #include "parmap_private.h"
  11
  12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_parmap, xbt, "parmap: parallel map");
  13 XBT_LOG_NEW_SUBCATEGORY(xbt_parmap_unit, xbt_parmap, "parmap unit testing");
  14
  15 static void *_xbt_parmap_worker_main(void *parmap);
  16
  17 static void futex_wait(int *uaddr, int val);
  18 static void futex_wake(int *uaddr, int val);
  19
  20 xbt_parmap_t xbt_parmap_new(unsigned int num_workers)
  21 {
  22   unsigned int i;
  23   xbt_os_thread_t worker = NULL;
  24
  25   DEBUG1("Create new parmap (%u workers)", num_workers);
  26
  27   /* Initialize the thread pool data structure */
  28   xbt_parmap_t parmap = xbt_new0(s_xbt_parmap_t, 1);
  29   parmap->num_workers = num_workers;
  30   parmap->status = PARMAP_WORK;
  31
  32   parmap->workers_ready = xbt_new0(s_xbt_barrier_t, 1);
  33   xbt_barrier_init(parmap->workers_ready, num_workers + 1);
  34   parmap->workers_done = xbt_new0(s_xbt_barrier_t, 1);
  35   xbt_barrier_init(parmap->workers_done, num_workers + 1);
  36
  37   /* Create the pool of worker threads */
  38   for(i=0; i < num_workers; i++){
  39     worker = xbt_os_thread_create(NULL, _xbt_parmap_worker_main, parmap, NULL);
  40     xbt_os_thread_detach(worker);
  41   }
  42
  43   return parmap;
  44 }
  45
  46 void xbt_parmap_destroy(xbt_parmap_t parmap)
  47 {
  48   DEBUG1("Destroy parmap %p", parmap);
  49
  50   parmap->status = PARMAP_DESTROY;
  51
  52   xbt_barrier_wait(parmap->workers_ready);
  53   DEBUG0("Kill job sent");
  54   xbt_barrier_wait(parmap->workers_done);
  55
  56   xbt_free(parmap->workers_ready);
  57   xbt_free(parmap->workers_done);
  58   xbt_free(parmap);
  59 }
  60
  61  void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
  62 {
  63   /* Assign resources to worker threads*/
  64   parmap->fun = fun;
  65   parmap->data = data;
  66
  67   /* Notify workers that there is a job */
  68   xbt_barrier_wait(parmap->workers_ready);
  69   DEBUG0("Job dispatched, lets wait...");
  70   xbt_barrier_wait(parmap->workers_done);
  71
  72   DEBUG0("Job done");
  73   parmap->fun = NULL;
  74   parmap->data = NULL;
  75 }
  76
  77 static void *_xbt_parmap_worker_main(void *arg)
  78 {
  79   unsigned int data_start, data_end, data_size, worker_id;
  80   xbt_parmap_t parmap = (xbt_parmap_t)arg;
  81
  82   /* Fetch a worker id */
  83   worker_id = __sync_fetch_and_add(&parmap->workers_max_id, 1);
  84
  85   DEBUG1("New worker thread created (%u)", worker_id);
  86
  87   /* Worker's main loop */
  88   while(1){
  89     xbt_barrier_wait(parmap->workers_ready);
  90
  91     if(parmap->status == PARMAP_WORK){
  92       DEBUG1("Worker %u got a job", worker_id);
  93
  94       /* Compute how much data does every worker gets */
  95       data_size = (xbt_dynar_length(parmap->data) / parmap->num_workers)
  96                   + ((xbt_dynar_length(parmap->data) % parmap->num_workers) ? 1 : 0);
  97
  98       /* Each worker data segment starts in a position associated with its id*/
  99       data_start = data_size * worker_id;
 100
 101       /* The end of the worker data segment must be bounded by the end of the data vector */
 102       data_end = MIN(data_start + data_size, xbt_dynar_length(parmap->data));
 103
 104       DEBUG4("Worker %u: data_start=%u data_end=%u (data_size=%u)",
 105           worker_id, data_start, data_end, data_size);
 106
 107       /* While the worker don't pass the end of it data segment apply the function */
 108       while(data_start < data_end){
 109         parmap->fun(*(void **)xbt_dynar_get_ptr(parmap->data, data_start));
 110         data_start++;
 111       }
 112
 113       xbt_barrier_wait(parmap->workers_done);
 114
 115     /* We are destroying the parmap */
 116     }else{
 117       xbt_barrier_wait(parmap->workers_done);
 118       DEBUG1("Shutting down worker %u", worker_id);
 119       return NULL;
 120     }
 121   }
 122 }
 123
 124 static void futex_wait(int *uaddr, int val)
 125 {
 126   DEBUG1("Waiting on futex %d", *uaddr);
 127   syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0);
 128 }
 129
 130 static void futex_wake(int *uaddr, int val)
 131 {
 132   DEBUG1("Waking futex %d", *uaddr);
 133   syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, NULL, NULL, 0);
 134 }
 135
 136 /* Futex based implementation of the barrier */
 137 void xbt_barrier_init(xbt_barrier_t barrier, unsigned int threads_to_wait)
 138 {
 139   barrier->threads_to_wait = threads_to_wait;
 140   barrier->thread_count = 0;
 141 }
 142
 143 void xbt_barrier_wait(xbt_barrier_t barrier)
 144 {
 145   int myflag = 0;
 146   unsigned int mycount = 0;
 147
 148   myflag = barrier->futex;
 149   mycount = __sync_add_and_fetch(&barrier->thread_count, 1);
 150   if(mycount < barrier->threads_to_wait){
 151     futex_wait(&barrier->futex, myflag);
 152   }else{
 153     barrier->futex = __sync_add_and_fetch(&barrier->futex, 1);
 154     barrier->thread_count = 0;
 155     futex_wake(&barrier->futex, barrier->threads_to_wait);
 156   }
 157 }
 158
 159 #ifdef SIMGRID_TEST
 160 #include "xbt.h"
 161 #include "xbt/ex.h"
 162
 163 XBT_TEST_SUITE("parmap", "Parallel Map");
 164 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(xbt_parmap_unit);
 165
 166
 167
 168 xbt_parmap_t parmap;
 169
 170 void fun(void *arg);
 171
 172 void fun(void *arg)
 173 {
 174   //INFO1("I'm job %lu", (unsigned long)arg);
 175 }
 176
 177 XBT_TEST_UNIT("basic", test_parmap_basic, "Basic usage")
 178 {
 179   xbt_test_add0("Create the parmap");
 180
 181   unsigned long i,j;
 182   xbt_dynar_t data = xbt_dynar_new(sizeof(void *), NULL);
 183
 184   /* Create the parallel map */
 185   parmap = xbt_parmap_new(10);
 186
 187   for(j=0; j < 100; j++){
 188     xbt_dynar_push_as(data, void *, (void *)j);
 189   }
 190
 191   for(i=0; i < 5; i++)
 192     xbt_parmap_apply(parmap, fun, data);
 193
 194   /* Destroy the parmap */
 195   xbt_parmap_destroy(parmap);
 196 }
 197
 198 #endif /* SIMGRID_TEST */