X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/2cd5e518bad23ef9e976363df62d45d381a1fde1..6e5cfd7ff86900354c20502af95ee5f751492753:/src/xbt/parmap.c diff --git a/src/xbt/parmap.c b/src/xbt/parmap.c index 9e9a0dff25..19dcb58027 100644 --- a/src/xbt/parmap.c +++ b/src/xbt/parmap.c @@ -34,27 +34,32 @@ xbt_parmap_t xbt_parmap_new(unsigned int num_workers) /* Initialize the thread pool data structure */ xbt_parmap_t parmap = xbt_new0(s_xbt_parmap_t, 1); +#ifdef HAVE_FUTEX_H parmap->sync_event = xbt_new0(s_xbt_event_t, 1); +#endif parmap->num_workers = num_workers; parmap->status = PARMAP_WORK; +#ifdef HAVE_FUTEX_H parmap->sync_event->threads_to_wait = num_workers; - +#endif /* Create the pool of worker threads */ for(i=0; i < num_workers; i++){ worker = xbt_os_thread_create(NULL, _xbt_parmap_worker_main, parmap, NULL); xbt_os_thread_detach(worker); } - +#ifdef HAVE_FUTEX_H xbt_event_init(parmap->sync_event); +#endif return parmap; } void xbt_parmap_destroy(xbt_parmap_t parmap) { - XBT_DEBUG("Destroy parmap %p", parmap); parmap->status = PARMAP_DESTROY; +#ifdef HAVE_FUTEX_H xbt_event_signal(parmap->sync_event); xbt_free(parmap->sync_event); +#endif xbt_free(parmap); } @@ -63,51 +68,58 @@ void xbt_parmap_destroy(xbt_parmap_t parmap) /* Assign resources to worker threads*/ parmap->fun = fun; parmap->data = data; - + parmap->index = 0; +#ifdef HAVE_FUTEX_H xbt_event_signal(parmap->sync_event); - +#endif XBT_DEBUG("Job done"); } +void* xbt_parmap_next(xbt_parmap_t parmap) { + + unsigned int index = __sync_fetch_and_add(&parmap->index, 1); + if (index < xbt_dynar_length(parmap->data)) { + return xbt_dynar_get_as(parmap->data, index, void*); + } + return NULL; +} + +unsigned long xbt_parmap_get_worker_id(xbt_parmap_t parmap) { + return (unsigned long) xbt_os_thread_get_extra_data(); +} + static void *_xbt_parmap_worker_main(void *arg) { - unsigned int data_start, data_end, data_size, worker_id; - xbt_parmap_t parmap = (xbt_parmap_t)arg; + unsigned int worker_id; + xbt_parmap_t parmap = (xbt_parmap_t) arg; /* Fetch a worker id */ worker_id = __sync_fetch_and_add(&parmap->workers_max_id, 1); - xbt_os_thread_set_extra_data((void *)(unsigned long)worker_id); + xbt_os_thread_set_extra_data((void*) (unsigned long) worker_id); XBT_DEBUG("New worker thread created (%u)", worker_id); /* Worker's main loop */ - while(1){ + while (1) { +#ifdef HAVE_FUTEX_H xbt_event_wait(parmap->sync_event); - if(parmap->status == PARMAP_WORK){ - XBT_DEBUG("Worker %u got a job", worker_id); - - /* Compute how much data does every worker gets */ - data_size = (xbt_dynar_length(parmap->data) / parmap->num_workers) - + ((xbt_dynar_length(parmap->data) % parmap->num_workers) ? 1 : 0); - - /* Each worker data segment starts in a position associated with its id*/ - data_start = data_size * worker_id; - - /* The end of the worker data segment must be bounded by the end of the data vector */ - data_end = MIN(data_start + data_size, xbt_dynar_length(parmap->data)); +#endif + if (parmap->status == PARMAP_WORK) { - XBT_DEBUG("Worker %u: data_start=%u data_end=%u (data_size=%u)", - worker_id, data_start, data_end, data_size); + XBT_DEBUG("Worker %u got a job", worker_id); - /* While the worker don't pass the end of it data segment apply the function */ - while(data_start < data_end){ - parmap->fun(*(void **)xbt_dynar_get_ptr(parmap->data, data_start)); - data_start++; + void* work = xbt_parmap_next(parmap); + if (work != NULL) { + parmap->fun(work); } + XBT_DEBUG("Worker %u has finished", worker_id); + /* We are destroying the parmap */ - }else{ + } else { +#ifdef HAVE_FUTEX_H xbt_event_end(parmap->sync_event); +#endif XBT_DEBUG("Shutting down worker %u", worker_id); return NULL; } @@ -160,10 +172,8 @@ void xbt_event_wait(xbt_event_t event) void xbt_event_end(xbt_event_t event) { - int myflag; unsigned int mycount; - myflag = event->work; mycount = __sync_add_and_fetch(&event->thread_counter, 1); if(mycount == event->threads_to_wait){ event->done++;