jacob/power_play
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

worker sleeping

Browse Source
This commit is contained in:
jacob 2025-07-06 22:47:08 -05:00
parent 2b08223472
commit e720e7e2af
7 changed files with 214 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/common.h
View File

@ -657,13 +657,14 @@ INLINE f64 clamp_f64(f64 v, f64 min, f64 max) { return v < min ? min : v > max ?
#include "prof_tracy.h" #include "prof_tracy.h"
#define PROF_THREAD_GROUP_WORKERS -MEBI(7) #define PROF_THREAD_GROUP_WORKERS -MEBI(8)
#define PROF_THREAD_GROUP_FIBERS -MEBI(6) #define PROF_THREAD_GROUP_FIBERS -MEBI(7)
#define PROF_THREAD_GROUP_IO -MEBI(5) #define PROF_THREAD_GROUP_SCHEDULER -MEBI(6)
#define PROF_THREAD_GROUP_WINDOW -MEBI(4) #define PROF_THREAD_GROUP_IO -MEBI(5)
#define PROF_THREAD_GROUP_EVICTORS -MEBI(3) #define PROF_THREAD_GROUP_WINDOW -MEBI(4)
#define PROF_THREAD_GROUP_APP -MEBI(2) #define PROF_THREAD_GROUP_EVICTORS -MEBI(3)
#define PROF_THREAD_GROUP_MAIN -MEBI(1) #define PROF_THREAD_GROUP_APP -MEBI(2)
#define PROF_THREAD_GROUP_MAIN -MEBI(1)
#ifdef __cplusplus #ifdef __cplusplus
} }

44
src/playback_wasapi.c
View File

@ -11,6 +11,7 @@
#include "mixer.h" #include "mixer.h"
#include "atomic.h" #include "atomic.h"
#include "app.h" #include "app.h"
#include "snc.h"
#define COBJMACROS #define COBJMACROS
#define WIN32_LEAN_AND_MEAN #define WIN32_LEAN_AND_MEAN
@ -42,6 +43,7 @@ GLOBAL struct {
HANDLE event; HANDLE event;
IAudioRenderClient *playback; IAudioRenderClient *playback;
WAVEFORMATEX *buffer_format; WAVEFORMATEX *buffer_format;
struct sys_thread *playback_scheduler_thread;
u32 buffer_frames; u32 buffer_frames;
} G = ZI, DEBUG_ALIAS(G, G_playback_wasapi); } G = ZI, DEBUG_ALIAS(G, G_playback_wasapi);
@ -51,14 +53,14 @@ GLOBAL struct {
INTERNAL void wasapi_initialize(void); INTERNAL void wasapi_initialize(void);
INTERNAL APP_EXIT_CALLBACK_FUNC_DEF(playback_shutdown); INTERNAL APP_EXIT_CALLBACK_FUNC_DEF(playback_shutdown);
INTERNAL SYS_JOB_DEF(playback_job, _); INTERNAL SYS_THREAD_DEF(playback_scheduler_entry, _);
struct playback_startup_receipt playback_startup(struct mixer_startup_receipt *mixer_sr) struct playback_startup_receipt playback_startup(struct mixer_startup_receipt *mixer_sr)
{ {
(UNUSED)mixer_sr; (UNUSED)mixer_sr;
wasapi_initialize(); wasapi_initialize();
sys_run(1, playback_job, NULL, SYS_PRIORITY_HIGH, NULL); G.playback_scheduler_thread = sys_thread_alloc(playback_scheduler_entry, NULL, LIT("Playback scheduler thread"), PROF_THREAD_GROUP_SCHEDULER);
app_register_exit_callback(&playback_shutdown); app_register_exit_callback(&playback_shutdown);
return (struct playback_startup_receipt) { 0 }; return (struct playback_startup_receipt) { 0 };
@ -68,6 +70,7 @@ INTERNAL APP_EXIT_CALLBACK_FUNC_DEF(playback_shutdown)
{ {
__prof; __prof;
atomic_i32_fetch_set(&G.shutdown, true); atomic_i32_fetch_set(&G.shutdown, true);
sys_thread_wait_release(G.playback_scheduler_thread);
} }
/* ========================== * /* ========================== *
@ -172,12 +175,6 @@ INTERNAL struct wasapi_buffer wasapi_update_begin(void)
__prof; __prof;
struct wasapi_buffer wspbuf = ZI; struct wasapi_buffer wspbuf = ZI;
/* Wait */
{
__profn("Wasapi wait");
WaitForSingleObject(G.event, INFINITE);
}
/* Get padding frames */ /* Get padding frames */
u32 padding_frames; u32 padding_frames;
IAudioClient_GetCurrentPadding(G.client, &padding_frames); IAudioClient_GetCurrentPadding(G.client, &padding_frames);
@ -226,11 +223,22 @@ INTERNAL void wasapi_update_end(struct wasapi_buffer *wspbuf, struct mixed_pcm_f
* Playback thread entry * Playback thread entry
* ========================== */ * ========================== */
INTERNAL SYS_JOB_DEF(playback_job, _) INTERNAL SYS_JOB_DEF(playback_mix_job, _)
{ {
__prof;
(UNUSED)_; (UNUSED)_;
struct arena_temp scratch = scratch_begin_no_conflict(); struct arena_temp scratch = scratch_begin_no_conflict();
{
struct wasapi_buffer wspbuf = wasapi_update_begin();
struct mixed_pcm_f32 pcm = mixer_update(scratch.arena, wspbuf.frames_count);
wasapi_update_end(&wspbuf, pcm);
}
scratch_end(scratch);
}
INTERNAL SYS_THREAD_DEF(playback_scheduler_entry, _)
{
(UNUSED)_;
/* https://learn.microsoft.com/en-us/windows/win32/procthread/multimedia-class-scheduler-service#registry-settings */ /* https://learn.microsoft.com/en-us/windows/win32/procthread/multimedia-class-scheduler-service#registry-settings */
DWORD task = 0; DWORD task = 0;
HANDLE mmc_handle = AvSetMmThreadCharacteristicsW(L"Pro Audio", &task); HANDLE mmc_handle = AvSetMmThreadCharacteristicsW(L"Pro Audio", &task);
@ -240,13 +248,15 @@ INTERNAL SYS_JOB_DEF(playback_job, _)
/* FIXME: If playback fails at any point and mixer stops advancing, we /* FIXME: If playback fails at any point and mixer stops advancing, we
* need to halt mixer to prevent memory leak when sounds are played. */ * need to halt mixer to prevent memory leak when sounds are played. */
while (!atomic_i32_fetch(&G.shutdown)) { while (!atomic_i32_fetch(&G.shutdown)) {
struct arena_temp temp = arena_temp_begin(scratch.arena); {
struct wasapi_buffer wspbuf = wasapi_update_begin(); __profn("Wait for audio event");
struct mixed_pcm_f32 pcm = mixer_update(temp.arena, wspbuf.frames_count); WaitForSingleObject(G.event, INFINITE);
wasapi_update_end(&wspbuf, pcm); }
arena_temp_end(temp); {
__profn("Run mix job & wait");
struct snc_counter counter = ZI;
sys_run(1, playback_mix_job, NULL, SYS_PRIORITY_HIGH, &counter);
snc_counter_wait(&counter);
}
} }
scratch_end(scratch);
} }

10
src/resource.c
View File

@ -159,8 +159,12 @@ INTERNAL APP_EXIT_CALLBACK_FUNC_DEF(resource_shutdown)
__prof; __prof;
atomic_i32_fetch_set(&G.watch_shutdown, 1); atomic_i32_fetch_set(&G.watch_shutdown, 1);
snc_cv_broadcast(&G.watch_dispatcher_cv); {
sys_watch_wake(G.watch); struct snc_lock lock = snc_lock_e(&G.watch_dispatcher_mutex);
snc_cv_broadcast(&G.watch_dispatcher_cv);
sys_watch_wake(G.watch);
snc_unlock(&lock);
}
sys_thread_wait_release(G.resource_watch_dispatch_thread); sys_thread_wait_release(G.resource_watch_dispatch_thread);
sys_thread_wait_release(G.resource_watch_monitor_thread); sys_thread_wait_release(G.resource_watch_monitor_thread);
@ -199,8 +203,8 @@ INTERNAL SYS_THREAD_DEF(resource_watch_monitor_thread_entry_point, _)
G.watch_dispatcher_info_list = list_part; G.watch_dispatcher_info_list = list_part;
} }
} }
snc_unlock(&lock);
snc_cv_broadcast(&G.watch_dispatcher_cv); snc_cv_broadcast(&G.watch_dispatcher_cv);
snc_unlock(&lock);
} }
arena_temp_end(temp); arena_temp_end(temp);
} }

103
src/snc.c
View File

@ -2,6 +2,7 @@
#include "atomic.h" #include "atomic.h"
#include "sys.h" #include "sys.h"
#include "memory.h" #include "memory.h"
#include "intrinsics.h"
#define DEFAULT_MUTEX_SPIN 4000 #define DEFAULT_MUTEX_SPIN 4000
@ -11,33 +12,48 @@
struct snc_lock snc_lock_spin_e(struct snc_mutex *m, i32 spin) struct snc_lock snc_lock_spin_e(struct snc_mutex *m, i32 spin)
{ {
__prof;
b32 locked = false; b32 locked = false;
i32 spin_cnt = 0;
while (!locked) { while (!locked) {
/* Spin lock */ ++spin_cnt;
i32 spin_cnt = 0; u32 v = atomic_u32_fetch_test_set(&m->v, 0, 0x80000000);
i32 v = atomic_i32_fetch_test_set(&m->v, 0, (1 << 31)); if (v == 0) {
do { locked = true;
if (v == 0) { } else if (v == 0x40000000) {
locked = true; /* Lock has pending bit set, try to lock */
} else { u32 swp = atomic_u32_fetch_test_set(&m->v, v, 0x80000000);
/* Set pending */ while (swp != v && swp == 0x40000000) {
if ((v & (1 << 30)) == 0) { v = swp;
i32 old = atomic_i32_fetch_test_set(&m->v, v | (1 << 30), v); swp = atomic_u32_fetch_test_set(&m->v, v, 0x80000000);
while (old != v && (old & (1 << 30)) == 0) { }
v = old; v = swp;
old = atomic_i32_fetch_test_set(&m->v, v | (1 << 30), v); if (v == 0x40000000) {
} locked = true;
v = old; }
} }
if (!locked && (v & 0xC0000000) == 0) {
/* Lock has shared lockers and no pending waiter, set pending bit */
u32 swp = atomic_u32_fetch_test_set(&m->v, v, v | 0x40000000);
while (swp != v && (swp & 0xC0000000) == 0 && swp != 0) {
v = swp;
swp = atomic_u32_fetch_test_set(&m->v, v, v | 0x40000000);
}
v = swp;
}
/* Pause or wait */
if (!locked && v != 0 && v != 0x40000000) {
if (spin_cnt < spin) {
ix_pause();
} else {
sys_wait(&m->v, &v, 4, F32_INFINITY);
} }
++spin_cnt;
} while (spin_cnt < spin && !locked);
/* Wait if not successful */
if (!locked) {
sys_wait(&m->v, &v, 4);
} }
} }
#if RTC
atomic_i32_fetch_set(&m->exclusive_fiber_id, sys_current_fiber_id());
#endif
struct snc_lock lock = ZI; struct snc_lock lock = ZI;
lock.exclusive = true; lock.exclusive = true;
lock.mutex = m; lock.mutex = m;
@ -46,24 +62,27 @@ struct snc_lock snc_lock_spin_e(struct snc_mutex *m, i32 spin)
struct snc_lock snc_lock_spin_s(struct snc_mutex *m, i32 spin) struct snc_lock snc_lock_spin_s(struct snc_mutex *m, i32 spin)
{ {
__prof;
b32 locked = false; b32 locked = false;
i32 spin_cnt = 0;
while (!locked) { while (!locked) {
/* Spin lock */ ++spin_cnt;
i32 spin_cnt = 0; u32 v = atomic_u32_fetch(&m->v);
i32 v = atomic_i32_fetch(&m->v); while (!locked && (v & 0xC0000000) == 0) {
do { /* Lock has no exclusive or pending exclusive lock, increment shared count */
while (!locked && (v & 0xC0000000) == 0) { u32 swp = atomic_u32_fetch_test_set(&m->v, v, v + 1);
/* Lock has no exclusive or pending exclusive lock, increment shared count */ if (v == swp) {
i32 old = atomic_i32_fetch_test_set(&m->v, v, v + 1); locked = true;
if (v == old) { } else {
locked = true; v = swp;
}
} }
} while (spin_cnt < spin && !locked); }
/* Wait if not successful */ /* Pause or wait */
if (!locked) { if (!locked) {
sys_wait(&m->v, &v, 4); if (spin_cnt < spin) {
ix_pause();
} else {
sys_wait(&m->v, &v, 4, F32_INFINITY);
}
} }
} }
struct snc_lock lock = ZI; struct snc_lock lock = ZI;
@ -83,12 +102,14 @@ struct snc_lock snc_lock_s(struct snc_mutex *m)
void snc_unlock(struct snc_lock *l) void snc_unlock(struct snc_lock *l)
{ {
__prof;
struct snc_mutex *m = l->mutex; struct snc_mutex *m = l->mutex;
if (l->exclusive) { if (l->exclusive) {
atomic_i32_fetch_set(&m->v, 0); #if RTC
atomic_i32_fetch_set(&m->exclusive_fiber_id, 0);
#endif
atomic_u32_fetch_set(&m->v, 0);
} else { } else {
atomic_i32_fetch_add(&m->v, -1); atomic_u32_fetch_add_i32(&m->v, -1);
} }
sys_wake_all(&m->v); sys_wake_all(&m->v);
MEMZERO_STRUCT(l); MEMZERO_STRUCT(l);
@ -107,7 +128,7 @@ void snc_cv_wait(struct snc_cv *cv, struct snc_lock *l)
{ {
snc_unlock(l); snc_unlock(l);
do { do {
sys_wait(&cv->wake_gen, &old_wake_gen, sizeof(old_wake_gen)); sys_wait(&cv->wake_gen, &old_wake_gen, sizeof(old_wake_gen), F32_INFINITY);
wake_gen = atomic_u64_fetch(&cv->wake_gen); wake_gen = atomic_u64_fetch(&cv->wake_gen);
} while (wake_gen == old_wake_gen); } while (wake_gen == old_wake_gen);
sys_wake_all(&cv->wake_gen); sys_wake_all(&cv->wake_gen);
@ -142,7 +163,7 @@ void snc_counter_wait(struct snc_counter *counter)
{ {
i64 v = atomic_i64_fetch(&counter->v); i64 v = atomic_i64_fetch(&counter->v);
while (v > 0) { while (v > 0) {
sys_wait(&counter->v, &v, sizeof(v)); sys_wait(&counter->v, &v, sizeof(v), F32_INFINITY);
v = atomic_i64_fetch(&counter->v); v = atomic_i64_fetch(&counter->v);
} }
} }

6
src/snc.h
View File

@ -11,7 +11,11 @@ struct snc_lock {
}; };
struct snc_mutex { struct snc_mutex {
struct atomic_i32 v; struct atomic_u32 v;
#if RTC
struct atomic_i32 exclusive_fiber_id;
#endif
}; };
struct snc_lock snc_lock_spin_e(struct snc_mutex *m, i32 spin); struct snc_lock snc_lock_spin_e(struct snc_mutex *m, i32 spin);

2
src/sys.h
View File

@ -449,7 +449,7 @@ b32 sys_run_command(struct string cmd);
/* Futex-like wait & wake */ /* Futex-like wait & wake */
void sys_wait(void *addr, void *cmp, u32 size); void sys_wait(void *addr, void *cmp, u32 size, f32 timeout_seconds);
void sys_wake_single(void *addr); void sys_wake_single(void *addr);
void sys_wake_all(void *addr); void sys_wake_all(void *addr);

130
src/sys_win32.c
View File

@ -256,7 +256,6 @@ STATIC_ASSERT(alignof(struct fiber_ctx) == 64); /* Avoid false sharing */
struct alignas(64) worker_ctx { struct alignas(64) worker_ctx {
i32 id; i32 id;
HANDLE sleep_timer;
}; };
struct job_info { struct job_info {
@ -357,7 +356,10 @@ GLOBAL struct {
struct job_queue job_queues[NUM_JOB_QUEUE_KINDS]; struct job_queue job_queues[NUM_JOB_QUEUE_KINDS];
/* Workers */ /* Workers */
struct atomic_i32 workers_shutdown; /* TODO: Prevent false sharing */ struct atomic_i64 workers_wake_gen; /* TODO: Prevent false sharing */
struct snc_mutex workers_wake_mutex;
struct snc_cv workers_wake_cv;
i32 num_worker_threads; i32 num_worker_threads;
struct arena *worker_threads_arena; struct arena *worker_threads_arena;
struct sys_thread **worker_threads; struct sys_thread **worker_threads;
@ -390,12 +392,8 @@ INTERNAL void job_fiber_yield(struct fiber *fiber, struct fiber *parent_fiber);
* Wait / wake * Wait / wake
* ========================== */ * ========================== */
void sys_wait(void *addr, void *cmp, u32 size) void sys_wait(void *addr, void *cmp, u32 size, f32 timeout_seconds)
{ {
__prof;
#if 0
WaitOnAddress(addr, cmp, size, INFINITE);
#else
struct fiber *fiber = fiber_from_id(sys_current_fiber_id()); struct fiber *fiber = fiber_from_id(sys_current_fiber_id());
i16 parent_fiber_id = fiber->parent_id; i16 parent_fiber_id = fiber->parent_id;
/* Yield if job fiber, otherwise fall back to windows blocking function */ /* Yield if job fiber, otherwise fall back to windows blocking function */
@ -412,15 +410,18 @@ void sys_wait(void *addr, void *cmp, u32 size)
struct fiber *parent_fiber = fiber_from_id(parent_fiber_id); struct fiber *parent_fiber = fiber_from_id(parent_fiber_id);
job_fiber_yield(fiber, parent_fiber); job_fiber_yield(fiber, parent_fiber);
} else { } else {
WaitOnAddress(addr, cmp, size, INFINITE); i32 timeout_ms = INFINITE;
if (timeout_seconds != F32_INFINITY) {
timeout_ms = (i32)(timeout_seconds * 1000);
}
WaitOnAddress(addr, cmp, size, timeout_ms);
} }
#endif
} }
void sys_wake_single(void *addr) void sys_wake_single(void *addr)
{ {
ASSERT(false); /* FIXME: Real wake single */
(UNUSED)addr; sys_wake_all(addr);
} }
void sys_wake_all(void *addr) void sys_wake_all(void *addr)
@ -428,6 +429,7 @@ void sys_wake_all(void *addr)
u64 wait_bin_index = (u64)addr % NUM_WAIT_ADDR_BINS; u64 wait_bin_index = (u64)addr % NUM_WAIT_ADDR_BINS;
struct wait_bin *bin = &G.wait_bins[wait_bin_index]; struct wait_bin *bin = &G.wait_bins[wait_bin_index];
i32 num_yielders = 0;
while (atomic_i32_fetch_test_set(&bin->lock, 0, 1) != 0) ix_pause(); while (atomic_i32_fetch_test_set(&bin->lock, 0, 1) != 0) ix_pause();
{ {
struct wait_list *wait_list = NULL; struct wait_list *wait_list = NULL;
@ -437,6 +439,7 @@ void sys_wake_all(void *addr)
} }
} }
if (wait_list && wait_list->num_yielders > 0) { if (wait_list && wait_list->num_yielders > 0) {
num_yielders = wait_list->num_yielders;
struct arena_temp scratch = scratch_begin_no_conflict(); struct arena_temp scratch = scratch_begin_no_conflict();
{ {
/* Separate yielders by queue kind */ /* Separate yielders by queue kind */
@ -444,7 +447,7 @@ void sys_wake_all(void *addr)
struct yielder **queue_yielder_arrays[NUM_JOB_QUEUE_KINDS] = ZI; struct yielder **queue_yielder_arrays[NUM_JOB_QUEUE_KINDS] = ZI;
for (i32 i = 0; i < (i32)countof(queue_yielder_arrays); ++i) { for (i32 i = 0; i < (i32)countof(queue_yielder_arrays); ++i) {
/* NOTE: Each array is conservatively sized as the number of all yielders in the list */ /* NOTE: Each array is conservatively sized as the number of all yielders in the list */
queue_yielder_arrays[i] = arena_push_array_no_zero(scratch.arena, struct yielder *, wait_list->num_yielders); queue_yielder_arrays[i] = arena_push_array_no_zero(scratch.arena, struct yielder *, num_yielders);
} }
for (struct yielder *yielder = wait_list->first_yielder; yielder; yielder = yielder->next) { for (struct yielder *yielder = wait_list->first_yielder; yielder; yielder = yielder->next) {
enum job_queue_kind queue_kind = yielder->job_queue_kind; enum job_queue_kind queue_kind = yielder->job_queue_kind;
@ -502,6 +505,19 @@ void sys_wake_all(void *addr)
/* Wake blocking waiters */ /* Wake blocking waiters */
WakeByAddressAll(addr); WakeByAddressAll(addr);
/* Wake workers */
/* TODO: Only wake necessary amount of workers */
if (num_yielders > 0) {
struct snc_lock lock = snc_lock_e(&G.workers_wake_mutex);
{
if (atomic_i64_fetch(&G.workers_wake_gen) >= 0) {
atomic_i64_fetch_add(&G.workers_wake_gen, 1);
snc_cv_broadcast(&G.workers_wake_cv);
}
}
snc_unlock(&lock);
}
} }
/* ========================== * /* ========================== *
@ -651,6 +667,16 @@ void sys_run(i32 count, sys_job_func *func, void *sig, enum sys_priority priorit
} }
atomic_i32_fetch_set(&queue->lock, 0); atomic_i32_fetch_set(&queue->lock, 0);
} }
/* Wake workers */
/* TODO: Only wake necessary amount of workers */
struct snc_lock lock = snc_lock_e(&G.workers_wake_mutex);
{
if (atomic_i64_fetch(&G.workers_wake_gen) >= 0) {
atomic_i64_fetch_add(&G.workers_wake_gen, 1);
snc_cv_broadcast(&G.workers_wake_cv);
}
}
snc_unlock(&lock);
} }
/* ========================== * /* ========================== *
@ -713,16 +739,17 @@ INTERNAL void job_fiber_entry(void *id_ptr)
} }
/* ========================== * /* ========================== *
* Worker entry * Job worker thread
* ========================== */ * ========================== */
INTERNAL SYS_THREAD_DEF(worker_entry, worker_ctx_arg) INTERNAL SYS_THREAD_DEF(job_worker_entry, worker_ctx_arg)
{ {
struct worker_ctx *ctx = worker_ctx_arg; struct worker_ctx *ctx = worker_ctx_arg;
(UNUSED)ctx; (UNUSED)ctx;
{ {
/* TODO: Heuristic pinning */ /* TODO: Heuristic pinning */
/* TODO: Pin non-worker threads to other cores */
HANDLE thread_handle = GetCurrentThread(); HANDLE thread_handle = GetCurrentThread();
b32 success = false; b32 success = false;
(UNUSED)success; (UNUSED)success;
@ -745,8 +772,13 @@ INTERNAL SYS_THREAD_DEF(worker_entry, worker_ctx_arg)
struct fiber *job_fiber = NULL; struct fiber *job_fiber = NULL;
while (!atomic_i32_fetch(&G.workers_shutdown)) { struct snc_lock wake_lock = snc_lock_s(&G.workers_wake_mutex);
i64 last_seen_wake_gen = atomic_i64_fetch(&G.workers_wake_gen);
while (last_seen_wake_gen >= 0) {
snc_unlock(&wake_lock);
/* Pull job from queue */ /* Pull job from queue */
b32 queues_empty = true;
enum sys_priority job_priority = 0; enum sys_priority job_priority = 0;
enum job_queue_kind job_queue_kind = 0; enum job_queue_kind job_queue_kind = 0;
i32 job_fiber_id = 0; i32 job_fiber_id = 0;
@ -755,7 +787,7 @@ INTERNAL SYS_THREAD_DEF(worker_entry, worker_ctx_arg)
void *job_sig = 0; void *job_sig = 0;
struct snc_counter *job_counter = 0; struct snc_counter *job_counter = 0;
{ {
//__profnc("Pull job", RGB32_F(0.75, 0.75, 0)); __profnc("Pull job", RGB32_F(0.75, 0.75, 0));
for (u32 queue_index = 0; queue_index < countof(queues) && !job_func; ++queue_index) { for (u32 queue_index = 0; queue_index < countof(queues) && !job_func; ++queue_index) {
struct job_queue *queue = queues[queue_index]; struct job_queue *queue = queues[queue_index];
if (queue) { if (queue) {
@ -775,21 +807,25 @@ INTERNAL SYS_THREAD_DEF(worker_entry, worker_ctx_arg)
job_func = info->func; job_func = info->func;
job_sig = info->sig; job_sig = info->sig;
job_counter = info->counter; job_counter = info->counter;
if (info->fiber_id > 0) {
}
if (job_id == (info->count - 1)) { if (job_id == (info->count - 1)) {
/* We're picking up the last dispatch, so dequeue the job */ /* We're picking up the last dispatch, so dequeue the job */
dequeue = true; dequeue = true;
} }
if (!next) {
queues_empty = queue_index >= ((i32)countof(queues) - 1);
}
} }
} else { } else {
/* This job is being resumed from a yield */ /* This job is to be resumed from a yield */
job_fiber_id = info->fiber_id; job_fiber_id = info->fiber_id;
job_id = info->num_dispatched; job_id = info->num_dispatched;
job_func = info->func; job_func = info->func;
job_sig = info->sig; job_sig = info->sig;
job_counter = info->counter; job_counter = info->counter;
dequeue = true; dequeue = true;
if (!next) {
queues_empty = queue_index >= ((i32)countof(queues) - 1);
}
} }
if (dequeue) { if (dequeue) {
if (!next) { if (!next) {
@ -937,11 +973,45 @@ INTERNAL SYS_THREAD_DEF(worker_entry, worker_ctx_arg)
} }
} }
} }
wake_lock = snc_lock_s(&G.workers_wake_mutex);
if (queues_empty) {
i64 new_wake_gen = atomic_i64_fetch(&G.workers_wake_gen);
while (new_wake_gen == last_seen_wake_gen) {
__profnc("Wait for job", RGB32_F(0.75, 0.75, 0));
snc_cv_wait(&G.workers_wake_cv, &wake_lock);
new_wake_gen = atomic_i64_fetch(&G.workers_wake_gen);
}
last_seen_wake_gen = new_wake_gen;
} else {
last_seen_wake_gen = atomic_i64_fetch(&G.workers_wake_gen);
}
}
snc_unlock(&wake_lock);
}
/* ========================== *
* Job scheduler thread
* ========================== */
INTERNAL SYS_THREAD_DEF(job_scheduler_entry, _)
{
HANDLE timer = CreateWaitableTimerExW(NULL, NULL, CREATE_WAITABLE_TIMER_HIGH_RESOLUTION, TIMER_ALL_ACCESS);
if (!timer) {
sys_panic(LIT("Failed to create high resolution timer"));
}
while (atomic_i64_fetch(&G.workers_wake_gen) >= 0) {
__profn("Scheduler");
LARGE_INTEGER due = ZI;
due.QuadPart = -1000;
SetWaitableTimerEx(timer, &due, 0, NULL, NULL, NULL, 0);
//SetWaitableTimerEx(timer, &due, 5000, NULL, NULL, NULL, 0);
WaitForSingleObject(timer, INFINITE);
} }
} }
/* ========================== * /* ========================== *
* Test entry * Test thread
* ========================== */ * ========================== */
INTERNAL SYS_THREAD_DEF(test_entry, _) INTERNAL SYS_THREAD_DEF(test_entry, _)
@ -951,23 +1021,26 @@ INTERNAL SYS_THREAD_DEF(test_entry, _)
/* Start workers */ /* Start workers */
G.num_worker_threads = 6; G.num_worker_threads = 6;
//G.num_worker_threads = 2;
G.worker_threads_arena = arena_alloc(GIBI(64)); G.worker_threads_arena = arena_alloc(GIBI(64));
G.worker_threads = arena_push_array(G.worker_threads_arena, struct sys_thread *, G.num_worker_threads); G.worker_threads = arena_push_array(G.worker_threads_arena, struct sys_thread *, G.num_worker_threads);
G.worker_contexts = arena_push_array(G.worker_threads_arena, struct worker_ctx, G.num_worker_threads); G.worker_contexts = arena_push_array(G.worker_threads_arena, struct worker_ctx, G.num_worker_threads);
for (i32 i = 0; i < G.num_worker_threads; ++i) { for (i32 i = 0; i < G.num_worker_threads; ++i) {
struct worker_ctx *ctx = &G.worker_contexts[i]; struct worker_ctx *ctx = &G.worker_contexts[i];
ctx->id = i; ctx->id = i;
ctx->sleep_timer = CreateWaitableTimerExW(NULL, NULL, CREATE_WAITABLE_TIMER_HIGH_RESOLUTION, TIMER_ALL_ACCESS);
struct string name = string_format(scratch.arena, LIT("Worker #%F"), FMT_SINT(i)); struct string name = string_format(scratch.arena, LIT("Worker #%F"), FMT_SINT(i));
G.worker_threads[i] = sys_thread_alloc(worker_entry, ctx, name, PROF_THREAD_GROUP_WORKERS + i); G.worker_threads[i] = sys_thread_alloc(job_worker_entry, ctx, name, PROF_THREAD_GROUP_WORKERS + i);
} }
/* Start scheduler */
struct sys_thread *scheduler_thread = sys_thread_alloc(job_scheduler_entry, NULL, LIT("Scheduler thread"), PROF_THREAD_GROUP_SCHEDULER);
/* Wait on workers */ /* Wait on workers */
for (i32 i = 0; i < G.num_worker_threads; ++i) { for (i32 i = 0; i < G.num_worker_threads; ++i) {
struct sys_thread *worker_thread = G.worker_threads[i]; struct sys_thread *worker_thread = G.worker_threads[i];
sys_thread_wait_release(worker_thread); sys_thread_wait_release(worker_thread);
} }
/* Wait on scheduler */
sys_thread_wait_release(scheduler_thread);
scratch_end(scratch); scratch_end(scratch);
} }
@ -3045,12 +3118,12 @@ int CALLBACK wWinMain(_In_ HINSTANCE instance, _In_opt_ HINSTANCE prev_instance,
/* Get app thread handle */ /* Get app thread handle */
HANDLE app_thread_handle = 0; HANDLE app_thread_handle = 0;
struct snc_lock lock = snc_lock_s(&G.threads_mutex);
{ {
struct snc_lock lock = snc_lock_s(&G.threads_mutex);
struct win32_thread *wt = (struct win32_thread *)app_thread; struct win32_thread *wt = (struct win32_thread *)app_thread;
app_thread_handle = wt->handle; app_thread_handle = wt->handle;
snc_unlock(&lock);
} }
snc_unlock(&lock);
/* Wait for either app thread exit or panic */ /* Wait for either app thread exit or panic */
@ -3068,7 +3141,12 @@ int CALLBACK wWinMain(_In_ HINSTANCE instance, _In_opt_ HINSTANCE prev_instance,
} }
/* Shutdown test thread */ /* Shutdown test thread */
atomic_i32_fetch_set(&G.workers_shutdown, 1); {
struct snc_lock lock = snc_lock_e(&G.workers_wake_mutex);
atomic_i64_fetch_set(&G.workers_wake_gen, -1);
snc_cv_broadcast(&G.workers_wake_cv);
snc_unlock(&lock);
}
sys_thread_wait_release(test_thread); sys_thread_wait_release(test_thread);
} }