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
   7 #include "parmap_private.h"
   8
   9 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_parmap, xbt,
  10                                 "parmap: parallel map");
  11
  12 static void *_xbt_parmap_worker_main(void *parmap);
  13
  14 xbt_parmap_t xbt_parmap_new(unsigned int num_workers)
  15 {
  16   unsigned int i;
  17   xbt_os_thread_t worker = NULL;
  18
  19   DEBUG1("Create new parmap (%u workers)", num_workers);
  20
  21   /* Initialize thread pool data structure */
  22   xbt_parmap_t parmap = xbt_new0(s_xbt_parmap_t, 1);
  23   parmap->mutex = xbt_os_mutex_init();
  24   parmap->job_posted = xbt_os_cond_init();
  25   parmap->all_done = xbt_os_cond_init();
  26   parmap->flags = xbt_new0(e_xbt_parmap_flag_t, num_workers + 1);
  27   parmap->num_workers = num_workers;
  28   parmap->num_idle_workers = 0;
  29   parmap->workers_max_id = 0;
  30
  31   /* Init our flag to wait (for workers' initialization) */
  32   parmap->flags[num_workers] = PARMAP_WAIT;
  33
  34   /* Create the pool of worker threads */
  35   for(i=0; i < num_workers; i++){
  36     worker = xbt_os_thread_create(NULL, _xbt_parmap_worker_main, parmap, NULL);
  37     xbt_os_thread_detach(worker);
  38   }
  39
  40   /* wait for the workers to initialize */
  41   xbt_os_mutex_acquire(parmap->mutex);
  42   while(parmap->flags[num_workers] == PARMAP_WAIT)
  43     xbt_os_cond_wait(parmap->all_done, parmap->mutex);
  44   xbt_os_mutex_release(parmap->mutex);
  45
  46   return parmap;
  47 }
  48
  49 void xbt_parmap_destroy(xbt_parmap_t parmap)
  50 {
  51   DEBUG1("Destroy parmap %p", parmap);
  52
  53   unsigned int i;
  54
  55   /* Lock the parmap, then signal every worker an wait for each to finish */
  56   xbt_os_mutex_acquire(parmap->mutex);
  57   for(i=0; i < parmap->num_workers; i++){
  58     parmap->flags[i] = PARMAP_DESTROY;
  59   }
  60
  61   xbt_os_cond_broadcast(parmap->job_posted);
  62   while(parmap->num_workers){
  63     DEBUG1("Still %u workers, waiting...", parmap->num_workers);
  64     xbt_os_cond_wait(parmap->all_done, parmap->mutex);
  65   }
  66
  67   /* Destroy pool's data structures */
  68   xbt_os_cond_destroy(parmap->job_posted);
  69   xbt_os_cond_destroy(parmap->all_done);
  70   xbt_free(parmap->flags);
  71   xbt_os_mutex_release(parmap->mutex);
  72   xbt_os_mutex_destroy(parmap->mutex);
  73   xbt_free(parmap);
  74 }
  75
  76 void xbt_parmap_apply(xbt_parmap_t parmap, void_f_pvoid_t fun, xbt_dynar_t data)
  77 {
  78   unsigned int i;
  79   unsigned int myflag_idx = parmap->num_workers;
  80
  81   /* Assign resources to worker threads */
  82   xbt_os_mutex_acquire(parmap->mutex);
  83   parmap->fun = fun;
  84   parmap->data = data;
  85   parmap->num_idle_workers = 0;
  86
  87   /* Set worker flags to work */
  88   for(i=0; i < parmap->num_workers; i++){
  89     parmap->flags[i] = PARMAP_WORK;
  90   }
  91
  92   /* Set our flag to wait (for the job to be completed)*/
  93   parmap->flags[myflag_idx] = PARMAP_WAIT;
  94
  95   /* Notify workers that there is a job */
  96   xbt_os_cond_broadcast(parmap->job_posted);
  97   DEBUG0("Job dispatched, lets wait...");
  98
  99   /* wait for the workers to finish */
 100   while(parmap->flags[myflag_idx] == PARMAP_WAIT)
 101     xbt_os_cond_wait(parmap->all_done, parmap->mutex);
 102
 103   DEBUG0("Job done");
 104   parmap->fun = NULL;
 105   parmap->data = NULL;
 106
 107   xbt_os_mutex_release(parmap->mutex);
 108   return;
 109 }
 110
 111 static void *_xbt_parmap_worker_main(void *arg)
 112 {
 113   unsigned int data_start, data_end, data_size, worker_id;
 114   xbt_parmap_t parmap = (xbt_parmap_t)arg;
 115
 116   /* Fetch a worker id */
 117   xbt_os_mutex_acquire(parmap->mutex);
 118   worker_id = parmap->workers_max_id++;
 119   xbt_os_mutex_release(parmap->mutex);
 120
 121   DEBUG1("New worker thread created (%u)", worker_id);
 122
 123   /* Worker's main loop */
 124   while(1){
 125     xbt_os_mutex_acquire(parmap->mutex);
 126     parmap->flags[worker_id] = PARMAP_WAIT;
 127     parmap->num_idle_workers++;
 128
 129     /* If everybody is done set the parmap work flag and signal it */
 130     if(parmap->num_idle_workers == parmap->num_workers){
 131       DEBUG1("Worker %u: All done, signal the parmap", worker_id);
 132       parmap->flags[parmap->num_workers] = PARMAP_WORK;
 133       xbt_os_cond_signal(parmap->all_done);
 134     }
 135
 136     /* If the wait flag is set then ... wait. */
 137     while(parmap->flags[worker_id] == PARMAP_WAIT)
 138       xbt_os_cond_wait(parmap->job_posted, parmap->mutex);
 139
 140     DEBUG1("Worker %u got a job", worker_id);
 141
 142     /* If we are shutting down, the last worker is going to signal the
 143      * parmap so it can finish destroying the data structure */
 144     if(parmap->flags[worker_id] == PARMAP_DESTROY){
 145       DEBUG1("Shutting down worker %u", worker_id);
 146       parmap->num_workers--;
 147       if(parmap->num_workers == 0)
 148         xbt_os_cond_signal(parmap->all_done);
 149       xbt_os_mutex_release(parmap->mutex);
 150       return NULL;
 151     }
 152     xbt_os_mutex_release(parmap->mutex);
 153
 154     /* Compute how much data does every worker gets */
 155     data_size = (xbt_dynar_length(parmap->data) / parmap->num_workers)
 156                 + ((xbt_dynar_length(parmap->data) % parmap->num_workers) ? 1 : 0);
 157
 158     /* Each worker data segment starts in a position associated with its id*/
 159     data_start = data_size * worker_id;
 160
 161     /* The end of the worker data segment must be bounded by the end of the data vector */
 162     data_end = MIN(data_start + data_size, xbt_dynar_length(parmap->data));
 163
 164     DEBUG4("Worker %u: data_start=%u data_end=%u (data_size=%u)", worker_id, data_start, data_end, data_size);
 165
 166     /* While the worker don't pass the end of it data segment apply the function */
 167     while(data_start < data_end){
 168       parmap->fun(*(void **)xbt_dynar_get_ptr(parmap->data, data_start));
 169       data_start++;
 170     }
 171   }
 172 }
 173
 174 #ifdef SIMGRID_TEST
 175 #include "xbt.h"
 176 #include "xbt/ex.h"
 177
 178 XBT_TEST_SUITE("parmap", "Parallel Map");
 179
 180 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(xbt_parmap_unit, xbt,
 181                                 "unit parmap");
 182
 183
 184 xbt_parmap_t parmap;
 185
 186 void fun(void *arg);
 187
 188 void fun(void *arg)
 189 {
 190   INFO1("I'm job %lu", (unsigned long)arg);
 191 }
 192
 193 XBT_TEST_UNIT("basic", test_parmap_basic, "Basic usage")
 194 {
 195   xbt_test_add0("Create the parmap");
 196
 197   unsigned long j;
 198   xbt_dynar_t data = xbt_dynar_new(sizeof(void *), NULL);
 199
 200   /* Create the parallel map */
 201   parmap = xbt_parmap_new(5);
 202
 203   for(j=0; j < 200; j++){
 204     xbt_dynar_push_as(data, void *, (void *)j);
 205   }
 206
 207   xbt_parmap_apply(parmap, fun, data);
 208
 209   /* Destroy the parmap */
 210   xbt_parmap_destroy(parmap);
 211 }
 212
 213 #endif /* SIMGRID_TEST */