From 3f5406f1d452c0b963010eaea4daa178a025167b Mon Sep 17 00:00:00 2001
From: Patrick Pelissier <Patrick.Pelissier@gmail.com>
Date: Mon, 16 Apr 2018 23:57:28 +0200
Subject: [PATCH] API breakage: worker_sync takes an extra parameter (worker_t)
 to perform a wait-on-event instead of a busy waiting, reducing consummer
 ressources.

worker_init takes an int instead of an unsigned int for the number of workers.
It allows additional extra value -1 to initialize the number of workers to
twice the number of CPU, allowing to overload the CPU with loads and keep
them fully working in some scenario.
---
 m-worker.h           | 33 +++++++++++++++++++++++++--------
 tests/test-mworker.c |  4 ++--
 2 files changed, 27 insertions(+), 10 deletions(-)

diff --git a/m-worker.h b/m-worker.h
index a8fd1eb3..d0d78691 100644
--- a/m-worker.h
+++ b/m-worker.h
@@ -131,6 +131,10 @@ typedef struct worker_s {
 
   /* The global clear function */
   void (*clearFunc_g)(void);
+
+  m_mutex_t lock;
+  m_cond_t  a_thread_ends;
+
 } worker_t[1];
 
 /* Return the number of CPU of the system */
@@ -193,6 +197,9 @@ workeri_thread(void *arg)
     if (w.block == NULL) break;
     workeri_exec(&w);
     worker_queue_pop_release(g->queue_g);
+    m_mutex_lock(g->lock);
+    m_cond_broadcast(g->a_thread_ends);
+    m_mutex_unlock(g->lock);
   }
   if (g->clearFunc_g != NULL)
     g->clearFunc_g();
@@ -200,25 +207,28 @@ workeri_thread(void *arg)
 
 /* Initialization of the worker module 
    Input:
-   @numWorker: number of worker to create 
+   @numWorker: number of worker to create (0=autodetect, -1=2*autodetect)
    @extraQueue: number of extra work order we can get if all workers are full
    @resetFunc: function to reset the state of a worker between work orders (or NULL if none)
    @clearFunc: function to clear the state of a worker before terminaning (or NULL if none)
 */
 static inline void
-worker_init(worker_t g, unsigned int numWorker, unsigned int extraQueue, void (*resetFunc)(void), void (*clearFunc)(void))
+worker_init(worker_t g, int numWorker, unsigned int extraQueue, void (*resetFunc)(void), void (*clearFunc)(void))
 {
-  if (numWorker == 0)
-    numWorker = workeri_get_cpu_count()-1;
+  if (numWorker <= 0)
+    numWorker = (1 + (numWorker == -1))*workeri_get_cpu_count()-1;
   //printf ("Starting queue with: %d\n", numWorker + extraQueue);
   worker_queue_init(g->queue_g, numWorker + extraQueue);
   size_t numWorker_st = numWorker;
   g->worker = M_MEMORY_REALLOC(worker_thread_t, NULL, numWorker_st);
   M_ASSERT_INIT (g->worker != NULL);
-  g->numWorker_g = numWorker;
+  g->numWorker_g = numWorker_st;
   g->resetFunc_g = resetFunc;
   g->clearFunc_g = clearFunc;
-  for(size_t i = 0; i < numWorker; i++) {
+  m_mutex_init(g->lock);
+  m_cond_init(g->a_thread_ends);
+  
+  for(size_t i = 0; i < numWorker_st; i++) {
     m_thread_create(g->worker[i].id, workeri_thread, M_ASSIGN_CAST(void*, g));
   }
 }
@@ -242,6 +252,8 @@ worker_clear(worker_t g)
   }
   M_MEMORY_FREE(g->worker);
 
+  m_mutex_clear(g->lock);
+  m_cond_clear(g->a_thread_ends);
   worker_queue_clear(g->queue_g);
 }
 
@@ -321,10 +333,15 @@ worker_sync_p(worker_sync_t block)
 
 /* Wait for all work orders to be finished */
 static inline void
-worker_sync(worker_sync_t block)
+worker_sync(worker_t g, worker_sync_t block)
 {
   //printf ("Waiting for thread terminasion.\n");
-  while (!worker_sync_p(block)) ;
+  if (worker_sync_p(block)) return;
+  m_mutex_lock(g->lock);
+  while (!worker_sync_p(block)) {
+    m_cond_wait(g->a_thread_ends, g->lock);
+  }
+  m_mutex_unlock(g->lock);
 }
 
 /* Flush any work order in the queue if some remains.*/
diff --git a/tests/test-mworker.c b/tests/test-mworker.c
index 4c9b86db..b1078068 100644
--- a/tests/test-mworker.c
+++ b/tests/test-mworker.c
@@ -52,7 +52,7 @@ static int fib(int n)
   f.n = n - 2;
   worker_spawn (w_g, b, subfunc_1, &f);
   int y = fib (n-1);
-  worker_sync(b);
+  worker_sync(w_g, b);
   return f.x + y;
 }
 
@@ -93,7 +93,7 @@ static int fib2(int n)
   worker_start(b);
   WORKER_SPAWN(w_g, b, (n), { y2 = fib2(n-2); }, (y2));
   y1 = fib (n-1);
-  worker_sync(b);
+  worker_sync(w_g, b);
   return y1 + y2;
 }