X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/1a158b7a4db45818e8f62b7bdac7f0d1bf36d140..6e5cfd7ff86900354c20502af95ee5f751492753:/src/xbt/parmap.c

diff --git a/src/xbt/parmap.c b/src/xbt/parmap.c
index aea987c035..19dcb58027 100644
--- a/src/xbt/parmap.c
+++ b/src/xbt/parmap.c
@@ -34,41 +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;
-
-  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,59 +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");
+}
 
-  /* 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);
+void* xbt_parmap_next(xbt_parmap_t parmap) {
 
-  XBT_DEBUG("Job done");
-  parmap->fun = NULL;
-  parmap->data = NULL;
+  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){
-    xbt_barrier_wait(parmap->workers_ready);
+  while (1) {
+#ifdef HAVE_FUTEX_H
+    xbt_event_wait(parmap->sync_event);
+#endif
+    if (parmap->status == PARMAP_WORK) {
 
-    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));
-
-      XBT_DEBUG("Worker %u: data_start=%u data_end=%u (data_size=%u)",
-          worker_id, data_start, data_end, data_size);
-
-      /* 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_barrier_wait(parmap->workers_done);
+      XBT_DEBUG("Worker %u has finished", worker_id);
 
     /* We are destroying the parmap */
-    }else{
-      xbt_barrier_wait(parmap->workers_done);
+    } else {
+#ifdef HAVE_FUTEX_H
+      xbt_event_end(parmap->sync_event);
+#endif
       XBT_DEBUG("Shutting down worker %u", worker_id);
       return NULL;
     }
@@ -137,56 +127,59 @@ 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)
+{
+  unsigned int mycount;
+
+  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 */