slow & highly contentious implementation of barriers using more basic sync primitives; fall back if sync barriers not present in win32 version

src/base/base_core.h

@@ -253,13 +253,13 @@
 # include <intrin.h>
 # if ARCH_X64
 #  define ins_atomic_u128_eval_cond_assign(x,k,c) (B32)InterlockedCompareExchange128((__int64 *)(x), ((__int64 *)&(k))[1], ((__int64 *)&(k))[0], (__int64 *)c)
-#  define ins_atomic_u64_eval(x)                  *((volatile U64 *)(x))
+#  define ins_atomic_u64_eval(x)                  InterlockedAdd64((__int64 *)(x), 0)
 #  define ins_atomic_u64_inc_eval(x)              InterlockedIncrement64((__int64 *)(x))
 #  define ins_atomic_u64_dec_eval(x)              InterlockedDecrement64((__int64 *)(x))
 #  define ins_atomic_u64_eval_assign(x,c)         InterlockedExchange64((__int64 *)(x),(c))
 #  define ins_atomic_u64_add_eval(x,c)            InterlockedAdd64((__int64 *)(x), c)
 #  define ins_atomic_u64_eval_cond_assign(x,k,c)  InterlockedCompareExchange64((__int64 *)(x),(k),(c))
-#  define ins_atomic_u32_eval(x)                  *((volatile U32 *)(x))
+#  define ins_atomic_u32_eval(x)                  InterlockedAdd((LONG *)(x), 0)
 #  define ins_atomic_u32_inc_eval(x)              InterlockedIncrement((LONG *)(x))
 #  define ins_atomic_u32_dec_eval(x)              InterlockedDecrement((LONG *)(x))
 #  define ins_atomic_u32_eval_assign(x,c)         InterlockedExchange((LONG *)(x),(c))

src/base/base_threads.c

@@ -58,3 +58,137 @@ set_thread_namef(char *fmt, ...)
   va_end(args);
   scratch_end(scratch);
 }
+
+////////////////////////////////
+//~ rjf: Platform-Abstracted Synchronization Primitive Functions
+
+//- rjf: slow barriers
+
+typedef struct BarrierNode BarrierNode;
+struct BarrierNode
+{
+  BarrierNode *next;
+  U64 count;
+  U64 threads_left_to_enter;
+  U64 threads_left_to_leave;
+  RWMutex rw_mutex;
+  CondVar cv;
+};
+
+typedef struct BarrierTCTX BarrierTCTX;
+struct BarrierTCTX
+{
+  Arena *arena;
+  BarrierNode *free_barrier_node;
+};
+
+thread_static BarrierTCTX *barrier_tctx = 0;
+
+internal Barrier
+slow_barrier_alloc(U64 count)
+{
+  if(barrier_tctx == 0)
+  {
+    Arena *arena = arena_alloc();
+    barrier_tctx = push_array(arena, BarrierTCTX, 1);
+    barrier_tctx->arena = arena;
+  }
+  BarrierNode *n = barrier_tctx->free_barrier_node;
+  if(n != 0)
+  {
+    SLLStackPop(barrier_tctx->free_barrier_node);
+  }
+  else
+  {
+    n = push_array_no_zero(barrier_tctx->arena, BarrierNode, 1);
+  }
+  MemoryZeroStruct(n);
+  n->count = count;
+  n->threads_left_to_enter = count;
+  n->rw_mutex = rw_mutex_alloc();
+  n->cv = cond_var_alloc();
+  Barrier result = {(U64)n};
+  return result;
+}
+
+internal void
+slow_barrier_release(Barrier barrier)
+{
+  if(barrier_tctx == 0)
+  {
+    Arena *arena = arena_alloc();
+    barrier_tctx = push_array(arena, BarrierTCTX, 1);
+    barrier_tctx->arena = arena;
+  }
+  BarrierNode *n = (BarrierNode *)barrier.u64[0];
+  rw_mutex_release(n->rw_mutex);
+  cond_var_release(n->cv);
+  SLLStackPush(barrier_tctx->free_barrier_node, n);
+}
+
+internal void
+slow_barrier_wait(Barrier barrier)
+{
+  ProfBeginFunction();
+  BarrierNode *n = (BarrierNode *)barrier.u64[0];
+  U64 threads_left_to_enter = ins_atomic_u64_dec_eval(&n->threads_left_to_enter);
+  
+  //- rjf: threads left to enter > 0 => wait
+  if(threads_left_to_enter > 0)
+  {
+    // rjf: first try a spin loop
+    B32 done_waiting = 0;
+    ProfScope("spin loop wait") for(U64 spin_count = 0; spin_count < 10000; spin_count += 1)
+    {
+      if(ins_atomic_u64_eval(&n->threads_left_to_leave) != 0)
+      {
+        done_waiting = 1;
+        break;
+      }
+    }
+    
+    // rjf: not done waiting -> need to do slow wait on condition variable
+    if(!done_waiting) ProfScope("slow wait")
+    {
+      RWMutexScope(n->rw_mutex, 0) for(;;)
+      {
+        if(ins_atomic_u64_eval(&n->threads_left_to_leave) != 0)
+        {
+          break;
+        }
+        cond_var_wait_rw(n->cv, n->rw_mutex, 0, max_U64);
+      }
+    }
+    
+    // rjf: decrement leave counter
+    if(ins_atomic_u64_dec_eval(&n->threads_left_to_leave) > 0)
+    {
+      ProfScope("signal") cond_var_signal(n->cv);
+    }
+  }
+  
+  //- rjf: threads left to enter == 0 -> last thread, wakeup
+  else
+  {
+    ins_atomic_u64_eval_assign(&n->threads_left_to_enter, n->count);
+    ProfScope("wake up") RWMutexScope(n->rw_mutex, 1)
+    {
+      ins_atomic_u64_eval_assign(&n->threads_left_to_leave, n->count-1);
+    }
+    ProfScope("signal") cond_var_signal(n->cv);
+  }
+  
+  //- rjf: wait for threads left to leave == 0
+  ProfScope("wait for threads to leave") 
+  {
+    for(U64 spin_count = 0;; spin_count += 1)
+    {
+      if(ins_atomic_u64_eval(&n->threads_left_to_leave) == 0)
+      {
+        break;
+      }
+    }
+  }
+  
+  ProfEnd();
+}

src/base/base_threads.h

@@ -131,9 +131,9 @@ internal B32       semaphore_take(Semaphore semaphore, U64 endt_us);
 internal void      semaphore_drop(Semaphore semaphore);
 
 //- rjf: barriers
-internal Barrier   barrier_alloc(U64 count);
-internal void      barrier_release(Barrier barrier);
-internal void      barrier_wait(Barrier barrier);
+internal Barrier barrier_alloc(U64 count);
+internal void barrier_release(Barrier barrier);
+internal void barrier_wait(Barrier barrier);
 
 //- rjf: scope macros
 #define MutexScope(mutex) DeferLoop(mutex_take(mutex), mutex_drop(mutex))
@@ -142,6 +142,14 @@ internal void      barrier_wait(Barrier barrier);
 #define MutexScopeW(mutex) DeferLoop(rw_mutex_take_w(mutex), rw_mutex_drop_w(mutex))
 #define MutexScopeRWPromote(mutex) DeferLoop((rw_mutex_drop_r(mutex), rw_mutex_take_w(mutex)), (rw_mutex_drop_w(mutex), rw_mutex_take_r(mutex)))
 
+////////////////////////////////
+//~ rjf: Platform-Abstracted Synchronization Primitive Functions
+
+//- rjf: slow barriers
+internal Barrier   slow_barrier_alloc(U64 count);
+internal void      slow_barrier_release(Barrier barrier);
+internal void      slow_barrier_wait(Barrier barrier);
+
 ////////////////////////////////
 //~ rjf: @per_os_impl Safe Calls