sys wake count

src/resource.c

@@ -161,7 +161,7 @@ INTERNAL APP_EXIT_CALLBACK_FUNC_DEF(resource_shutdown)
 
     {
         struct snc_lock lock = snc_lock_e(&G.watch_dispatcher_mutex);
-        snc_cv_broadcast(&G.watch_dispatcher_cv);
+        snc_cv_signal(&G.watch_dispatcher_cv, I32_MAX);
         sys_watch_wake(G.watch);
         snc_unlock(&lock);
     }
@@ -203,7 +203,7 @@ INTERNAL SYS_THREAD_DEF(resource_watch_monitor_thread_entry_point, _)
                     G.watch_dispatcher_info_list = list_part;
                 }
             }
-            snc_cv_broadcast(&G.watch_dispatcher_cv);
+            snc_cv_signal(&G.watch_dispatcher_cv, I32_MAX);
             snc_unlock(&lock);
         }
         arena_temp_end(temp);

src/snc.c

@@ -113,7 +113,7 @@ void snc_unlock(struct snc_lock *l)
     } else {
         atomic_u32_fetch_add_i32(&m->v, -1);
     }
-    sys_wake_all(&m->v);
+    sys_wake(&m->v, I32_MAX);
     MEMZERO_STRUCT(l);
 }
 
@@ -144,10 +144,10 @@ void snc_cv_wait_time(struct snc_cv *cv, struct snc_lock *l, i64 timeout_ns)
     }
 }
 
-void snc_cv_broadcast(struct snc_cv *cv)
+void snc_cv_signal(struct snc_cv *cv, i32 count)
 {
     atomic_u64_fetch_add_u64(&cv->wake_gen, 1);
-    sys_wake_all(&cv->wake_gen);
+    sys_wake(&cv->wake_gen, count);
 }
 
 /* ========================== *
@@ -159,7 +159,7 @@ void snc_counter_add(struct snc_counter *counter, i64 x)
     i64 old_v = atomic_i64_fetch_add(&counter->v, x);
     i64 new_v = old_v + x;
     if (old_v > 0 && new_v <= 0) {
-        sys_wake_all(&counter->v);
+        sys_wake(&counter->v, I32_MAX);
     }
 }
 

src/snc.h

@@ -54,7 +54,7 @@ STATIC_ASSERT(alignof(struct snc_cv) == 64);  /* Prevent false sharing */
 
 void snc_cv_wait(struct snc_cv *cv, struct snc_lock *lock);
 void snc_cv_wait_time(struct snc_cv *cv, struct snc_lock *l, i64 timeout_ns);
-void snc_cv_broadcast(struct snc_cv *cv);
+void snc_cv_signal(struct snc_cv *cv, i32 count);
 
 /* ========================== *
  * Counter

src/sprite.c

@@ -264,7 +264,7 @@ INTERNAL APP_EXIT_CALLBACK_FUNC_DEF(sprite_shutdown)
     {
         struct snc_lock lock = snc_lock_e(&G.evictor_scheduler_mutex);
         G.evictor_scheduler_shutdown = 1;
-        snc_cv_broadcast(&G.evictor_scheduler_shutdown_cv);
+        snc_cv_signal(&G.evictor_scheduler_shutdown_cv, I32_MAX);
         snc_unlock(&lock);
     }
     /* Wait for evictor shutdown */

src/sys.h

@@ -443,8 +443,7 @@ i64 sys_current_scheduler_period_ns(void);
 /* Futex-like wait & wake */
 
 void sys_wait(void *addr, void *cmp, u32 size, i64 timeout_ns);
-void sys_wake_single(void *addr);
+void sys_wake(void *addr, i32 count);
-void sys_wake_all(void *addr);
 
 /* ========================== *
  * Fiber

src/sys_win32.c

@@ -42,6 +42,9 @@
 #define FIBER_NAME_SUFFIX_CSTR "]"
 #define FIBER_NAME_MAX_SIZE 64
 
+/* Arbitrary threshold for determining when to fall back from a looped WakeByAddressSingle to WakeByAddressAll */
+#define WAKE_ALL_THRESHOLD 8
+
 struct win32_thread {
     sys_thread_func *entry_point;
     void *thread_data;
@@ -407,28 +410,7 @@ void sys_wait(void *addr, void *cmp, u32 size, i64 timeout_ns)
     }
 }
 
-
+void sys_wake(void *addr, i32 count)
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-void sys_wake_single(void *addr)
-{
-    /* FIXME: Real wake single */
-    sys_wake_all(addr);
-}
-
-void sys_wake_all(void *addr)
 {
     struct arena_temp scratch = scratch_begin_no_conflict();
 
@@ -454,7 +436,7 @@ void sys_wake_all(void *addr)
             if (wait_addr_list) {
                 /* Build waiters array */
                 waiters = arena_push_array_no_zero(scratch.arena, struct fiber *, wait_addr_list->num_waiters);
-                for (struct fiber *waiter = fiber_from_id(wait_addr_list->first_waiter); waiter; waiter = fiber_from_id(waiter->next_addr_waiter)) {
+                for (struct fiber *waiter = fiber_from_id(wait_addr_list->first_waiter); waiter && num_waiters < count; waiter = fiber_from_id(waiter->next_addr_waiter)) {
                     if (atomic_u64_fetch_test_set(&waiter->wake_gen, 0, wake_gen) == 0) {
                         waiters[num_waiters] = waiter;
                         ++num_waiters;
@@ -592,7 +574,13 @@ void sys_wake_all(void *addr)
     }
 
     /* Wake blocking waiters */
-    WakeByAddressAll(addr);
+    if (count >= WAKE_ALL_THRESHOLD) {
+        WakeByAddressAll(addr);
+    } else {
+        for (i32 i = 0; i < count; ++i) {
+            WakeByAddressSingle(addr);
+        }
+    }
 
     /* Wake workers */
     /* TODO: Only wake necessary amount of workers */
@@ -600,7 +588,7 @@ void sys_wake_all(void *addr)
         struct snc_lock lock = snc_lock_e(&G.workers_wake_mutex);
         {
             atomic_i64_fetch_add(&G.num_jobs_in_queue.v, num_waiters);
-            snc_cv_broadcast(&G.workers_wake_cv);
+            snc_cv_signal(&G.workers_wake_cv, num_waiters);
         }
         snc_unlock(&lock);
     }
@@ -766,7 +754,7 @@ void sys_run(i32 count, sys_job_func *func, void *sig, enum sys_priority priorit
     struct snc_lock lock = snc_lock_e(&G.workers_wake_mutex);
     {
         atomic_i64_fetch_add(&G.num_jobs_in_queue.v, count);
-        snc_cv_broadcast(&G.workers_wake_cv);
+        snc_cv_signal(&G.workers_wake_cv, count);
     }
     snc_unlock(&lock);
 }
@@ -1119,7 +1107,7 @@ INTERNAL SYS_THREAD_DEF(job_worker_entry, worker_ctx_arg)
         {
             shutdown = atomic_i32_fetch(&G.workers_shutdown.v);
             while (atomic_i64_fetch(&G.num_jobs_in_queue.v) <= 0 && !shutdown) {
-                __profnc("Wait for job", RGB32_F(0.75, 0.75, 0));
+                //__profnc("Wait for job", RGB32_F(0.75, 0.75, 0));
                 snc_cv_wait(&G.workers_wake_cv, &wake_lock);
                 shutdown = atomic_i32_fetch(&G.workers_shutdown.v);
             }
@@ -1367,7 +1355,7 @@ INTERNAL SYS_THREAD_DEF(job_scheduler_entry, _)
                 struct snc_lock lock = snc_lock_e(&G.workers_wake_mutex);
                 {
                     atomic_i64_fetch_add(&G.num_jobs_in_queue.v, num_waiters);
-                    snc_cv_broadcast(&G.workers_wake_cv);
+                    snc_cv_signal(&G.workers_wake_cv, num_waiters);
                 }
                 snc_unlock(&lock);
             }
@@ -3393,7 +3381,7 @@ int CALLBACK wWinMain(_In_ HINSTANCE instance, _In_opt_ HINSTANCE prev_instance,
         {
             struct snc_lock lock = snc_lock_e(&G.workers_wake_mutex);
             atomic_i32_fetch_set(&G.workers_shutdown.v, 1);
-            snc_cv_broadcast(&G.workers_wake_cv);
+            snc_cv_signal(&G.workers_wake_cv, I32_MAX);
             snc_unlock(&lock);
         }
         sys_thread_wait_release(test_thread);