ditch semaphores in thread pool in favor of barriers

src/base/base_threads.c

@@ -192,3 +192,13 @@ slow_barrier_wait(Barrier barrier)
   
   ProfEnd();
 }
+
+////////////////////////////////
+///~ Semaphore
+
+internal void
+semaphore_drop(Semaphore semaphore)
+{
+  semaphore_drop_count(semaphore, 1);
+}
+

src/base/base_threads.h

@@ -129,6 +129,7 @@ internal Semaphore semaphore_open(String8 name);
 internal void      semaphore_close(Semaphore semaphore);
 internal B32       semaphore_take(Semaphore semaphore, U64 endt_us);
 internal void      semaphore_drop(Semaphore semaphore);
+internal void      semaphore_drop_count(Semaphore semaphore, U64 drop_count);
 
 //- rjf: barriers
 internal Barrier barrier_alloc(U64 count);

src/linker/thread_pool/thread_pool.c

@@ -4,6 +4,8 @@
 internal void
 tp_run_tasks(TP_Context *pool, TP_Worker *worker)
 {
+  barrier_wait(pool->barrier);
+
   for (;;) {
     S64 task_left = ins_atomic_u64_dec_eval(&pool->task_left);
 
@@ -20,12 +22,11 @@ tp_run_tasks(TP_Context *pool, TP_Worker *worker)
     // cache task count so we dont touch pool memory after atomic inc
     U64 task_count = pool->task_count;
 
-    // on last task ping main thread
+    // update task done count
-    U64 task_done = ins_atomic_u64_inc_eval(&pool->task_done);
+    ins_atomic_u64_inc_eval(&pool->task_done);
-    if (task_done == task_count) {
-      semaphore_drop(pool->main_semaphore);
-    }
   }
+
+  barrier_wait(pool->barrier);
 }
 
 internal void
@@ -34,9 +35,7 @@ tp_worker_main(void *raw_worker)
   TP_Worker  *worker = raw_worker;
   TP_Context *pool   = worker->pool;
   for (; pool->is_live; ) {
-    if (semaphore_take(pool->task_semaphore, max_U64)) {
+    tp_run_tasks(pool, worker);
-      tp_run_tasks(pool, worker);
-    }
   }
 }
 
@@ -47,9 +46,7 @@ tp_worker_main_shared(void *raw_worker)
   TP_Context *pool   = worker->pool;
   for (; pool->is_live; ) {
     if (semaphore_take(pool->exec_semaphore, max_U64)) {
-      if (semaphore_take(pool->task_semaphore, max_U64)) {
+      tp_run_tasks(pool, worker);
-        tp_run_tasks(pool, worker);
-      }
     }
   }
 }
@@ -63,17 +60,11 @@ tp_alloc(Arena *arena, U32 worker_count, U32 max_worker_count, String8 name)
   B32 is_shared = (name.size > 0);
 
   // alloc semaphores
-  Semaphore main_semaphore = {0};
-  Semaphore task_semaphore = {0};
   Semaphore exec_semaphore = {0};
   if (worker_count > 1) {
-    main_semaphore = semaphore_alloc(0, 1, str8_zero());
     if (is_shared) {
       AssertAlways(worker_count <= max_worker_count);
-      task_semaphore = semaphore_alloc(0, max_worker_count, name);
       exec_semaphore = semaphore_alloc(0, worker_count, str8_zero());
-    } else {
-      task_semaphore = semaphore_alloc(0, worker_count, str8_zero());
     }
   }
 
@@ -81,14 +72,12 @@ tp_alloc(Arena *arena, U32 worker_count, U32 max_worker_count, String8 name)
   void *worker_entry = is_shared ? tp_worker_main_shared : tp_worker_main;
 
   // init pool
-  TP_Context *pool     = push_array(arena, TP_Context, 1);
+  TP_Context *pool      = push_array(arena, TP_Context, 1);
-  pool->exec_semaphore = exec_semaphore;
+  pool->exec_semaphore  = exec_semaphore;
-  pool->task_semaphore = task_semaphore;
+  pool->barrier         = barrier_alloc(worker_count);
-  pool->main_semaphore = main_semaphore;
+  pool->is_live         = 1;
-  pool->barrier        = barrier_alloc(worker_count);
+  pool->worker_count    = worker_count;
-  pool->is_live        = 1;
+  pool->worker_arr      = push_array(arena, TP_Worker, worker_count);
-  pool->worker_count   = worker_count;
-  pool->worker_arr     = push_array(arena, TP_Worker, worker_count);
   
   // init worker data
   for (U64 i = 0; i < worker_count; i += 1) {
@@ -118,9 +107,6 @@ tp_release(TP_Context *pool)
       semaphore_drop(pool->exec_semaphore);
     }
   }
-  for EachIndex(i, pool->worker_count) {
-    semaphore_drop(pool->task_semaphore);
-  }
   for (U64 i = 1; i < pool->worker_count; i += 1) {
     thread_detach(pool->worker_arr[i].handle);
   }
@@ -128,8 +114,6 @@ tp_release(TP_Context *pool)
     semaphore_release(pool->exec_semaphore);
   }
   barrier_release(pool->barrier);
-  semaphore_release(pool->task_semaphore);
-  semaphore_release(pool->main_semaphore);
 
   MemoryZeroStruct(pool);
 }
@@ -199,34 +183,24 @@ tp_temp_end(TP_Temp temp)
 internal void
 tp_for_parallel(TP_Context *pool, TP_Arena *task_arena, U64 task_count, TP_TaskFunc *task_func, void *task_data)
 {
-  if (task_count > 0) {
+  if (task_count) {
     // init run
     pool->task_arena = task_arena;
     pool->task_func  = task_func;
     pool->task_data  = task_data;
     pool->task_count = task_count;
     pool->task_done  = 0;
-    ins_atomic_u64_eval_assign(&pool->task_left, task_count);
+    pool->task_left  = task_count;
 
-    U64 drop_count = Min(task_count, pool->worker_count);
+    // if we are in shared mode -> ping
-
+    if (*pool->exec_semaphore.u64) {
-    // if we are in shared mode ping local semaphore
+      U64 drop_count64 = Min(task_count, pool->worker_count);
-    if (pool->exec_semaphore.u64[0] != 0) {
+      U32 drop_count   = safe_cast_u32(drop_count64);
-      for (U64 worker_idx = 0; worker_idx < drop_count; worker_idx +=1) {
+      semaphore_drop_count(pool->exec_semaphore, drop_count);
-        semaphore_drop(pool->exec_semaphore);
-      }
-    }
-    
-    // ping shared semaphore
-    for (U64 worker_idx = 0; worker_idx < drop_count; worker_idx += 1) {
-      semaphore_drop(pool->task_semaphore);
     }
 
     // run tasks on main worker
-    tp_run_tasks(pool, &pool->worker_arr[0]);
+    tp_run_tasks(pool, pool->worker_arr);
-    
-    // wait for workers to finish tasks
-    semaphore_take(pool->main_semaphore, max_U64);
   }
 }
 

src/linux/base/linux_base.c

@@ -697,13 +697,15 @@ semaphore_take(Semaphore semaphore, U64 endt_us)
 }
 
 internal void
-semaphore_drop(Semaphore semaphore)
+semaphore_drop_count(Semaphore semaphore, U64 count)
 {
-  int err = -1;
+  for EachIndex(i, count) {
-  if(semaphore.u64[0] != 0)
+    int err = -1;
-  {
+    if(semaphore.u64[0] != 0)
-    err = LNX_RETRY_ON_EINTR(sem_post((sem_t*)semaphore.u64[0]));
+    {
-    Assert(err == 0);
+      err = LNX_RETRY_ON_EINTR(sem_post((sem_t*)*semaphore.u64));
+      Assert(err == 0);
+    }
   }
 }
 

src/win32/base/win32_base.c

@@ -686,10 +686,9 @@ semaphore_take(Semaphore semaphore, U64 endt_us)
 }
 
 internal void
-semaphore_drop(Semaphore semaphore)
+semaphore_drop_count(Semaphore semaphore, U64 drop_count)
 {
-  HANDLE handle = (HANDLE)semaphore.u64[0];
+  ReleaseSemaphore((HANDLE)*semaphore.u64, drop_count, 0);
-  ReleaseSemaphore(handle, 1, 0);
 }
 
 //- rjf: barriers