X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/1a158b7a4db45818e8f62b7bdac7f0d1bf36d140..1189d1797cc934d847d6641d809bbe060729f064:/src/xbt/parmap.c diff --git a/src/xbt/parmap.c b/src/xbt/parmap.c index aea987c035..b48954960f 100644 --- a/src/xbt/parmap.c +++ b/src/xbt/parmap.c @@ -34,41 +34,33 @@ 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; - - parmap->workers_ready = xbt_new0(s_xbt_barrier_t, 1); - xbt_barrier_init(parmap->workers_ready, num_workers + 1); - parmap->workers_done = xbt_new0(s_xbt_barrier_t, 1); - xbt_barrier_init(parmap->workers_done, num_workers + 1); -#ifndef HAVE_FUTEX_H - parmap->workers_ready->mutex = xbt_os_mutex_init(); - parmap->workers_ready->cond = xbt_os_cond_init(); +#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; - - xbt_barrier_wait(parmap->workers_ready); - XBT_DEBUG("Kill job sent"); - xbt_barrier_wait(parmap->workers_done); -#ifndef HAVE_FUTEX_H - xbt_os_mutex_destroy(parmap->workers_ready->mutex); - xbt_os_cond_destroy(parmap->workers_ready->cond); +#ifdef HAVE_FUTEX_H + xbt_event_signal(parmap->sync_event); + xbt_free(parmap->sync_event); #endif - xbt_free(parmap->workers_ready); - xbt_free(parmap->workers_done); xbt_free(parmap); } @@ -77,15 +69,10 @@ void xbt_parmap_destroy(xbt_parmap_t parmap) /* Assign resources to worker threads*/ parmap->fun = fun; parmap->data = data; - - /* Notify workers that there is a job */ - xbt_barrier_wait(parmap->workers_ready); - XBT_DEBUG("Job dispatched, lets wait..."); - xbt_barrier_wait(parmap->workers_done); - +#ifdef HAVE_FUTEX_H + xbt_event_signal(parmap->sync_event); +#endif XBT_DEBUG("Job done"); - parmap->fun = NULL; - parmap->data = NULL; } static void *_xbt_parmap_worker_main(void *arg) @@ -101,8 +88,9 @@ static void *_xbt_parmap_worker_main(void *arg) /* Worker's main loop */ while(1){ - xbt_barrier_wait(parmap->workers_ready); - +#ifdef HAVE_FUTEX_H + xbt_event_wait(parmap->sync_event); +#endif if(parmap->status == PARMAP_WORK){ XBT_DEBUG("Worker %u got a job", worker_id); @@ -125,11 +113,11 @@ static void *_xbt_parmap_worker_main(void *arg) data_start++; } - xbt_barrier_wait(parmap->workers_done); - /* We are destroying the parmap */ }else{ - xbt_barrier_wait(parmap->workers_done); +#ifdef HAVE_FUTEX_H + xbt_event_end(parmap->sync_event); +#endif XBT_DEBUG("Shutting down worker %u", worker_id); return NULL; } @@ -137,56 +125,61 @@ static void *_xbt_parmap_worker_main(void *arg) } #ifdef HAVE_FUTEX_H - static void futex_wait(int *uaddr, int val) - { - XBT_DEBUG("Waiting on futex %d", *uaddr); - syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0); - } - - static void futex_wake(int *uaddr, int val) - { - XBT_DEBUG("Waking futex %d", *uaddr); - syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, NULL, NULL, 0); - } -#endif +static void futex_wait(int *uaddr, int val) +{ + XBT_VERB("Waiting on futex %p", uaddr); + syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0); +} -/* Futex based implementation of the barrier */ -void xbt_barrier_init(xbt_barrier_t barrier, unsigned int threads_to_wait) +static void futex_wake(int *uaddr, int val) { - barrier->threads_to_wait = threads_to_wait; - barrier->thread_count = 0; + XBT_VERB("Waking futex %p", uaddr); + syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, val, NULL, NULL, 0); } -#ifdef HAVE_FUTEX_H - void xbt_barrier_wait(xbt_barrier_t barrier) - { - int myflag = 0; - unsigned int mycount = 0; - - myflag = barrier->futex; - mycount = __sync_add_and_fetch(&barrier->thread_count, 1); - if(mycount < barrier->threads_to_wait){ - futex_wait(&barrier->futex, myflag); - }else{ - barrier->futex = __sync_add_and_fetch(&barrier->futex, 1); - barrier->thread_count = 0; - futex_wake(&barrier->futex, barrier->threads_to_wait); - } - } -#else - void xbt_barrier_wait(xbt_barrier_t barrier) - { - xbt_os_mutex_acquire(barrier->mutex); - - barrier->thread_count++; - if(barrier->thread_count < barrier->threads_to_wait){ - xbt_os_cond_wait(barrier->cond,barrier->mutex); - }else{ - barrier->thread_count = 0; - xbt_os_cond_broadcast(barrier->cond); - } - xbt_os_mutex_release(barrier->mutex); - } +void xbt_event_init(xbt_event_t event) +{ + int myflag = event->done; + if(event->thread_counter < event->threads_to_wait) + futex_wait(&event->done, myflag); +} + +void xbt_event_signal(xbt_event_t event) +{ + int myflag = event->done; + event->thread_counter = 0; + event->work++; + futex_wake(&event->work, event->threads_to_wait); + futex_wait(&event->done, myflag); +} + +void xbt_event_wait(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++; + futex_wake(&event->done, 1); + } + + futex_wait(&event->work, myflag); +} + +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++; + futex_wake(&event->done, 1); + } +} #endif #ifdef SIMGRID_TEST @@ -226,6 +219,7 @@ XBT_TEST_UNIT("basic", test_parmap_basic, "Basic usage") /* Destroy the parmap */ xbt_parmap_destroy(parmap); + xbt_dynar_free(&data); } #endif /* SIMGRID_TEST */