Merge branch 'master' of scm.gforge.inria.fr:/gitroot/simgrid/simgrid

[simgrid.git] / src / xbt / parmap.c

diff --git a/src/xbt/parmap.c b/src/xbt/parmap.c
index 9e9a0df..19dcb58 100644 (file)
--- 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++;