diff --git a/src/base/base_threads.c b/src/base/base_threads.c index a8c72585..2c2d1d99 100644 --- a/src/base/base_threads.c +++ b/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); +} + diff --git a/src/base/base_threads.h b/src/base/base_threads.h index b34286cc..3aae2ea6 100644 --- a/src/base/base_threads.h +++ b/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); diff --git a/src/linker/thread_pool/thread_pool.c b/src/linker/thread_pool/thread_pool.c index 71003bb9..0d16213c 100644 --- a/src/linker/thread_pool/thread_pool.c +++ b/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 - U64 task_done = ins_atomic_u64_inc_eval(&pool->task_done); - if (task_done == task_count) { - semaphore_drop(pool->main_semaphore); - } + // update task done count + ins_atomic_u64_inc_eval(&pool->task_done); } + + 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); - pool->exec_semaphore = exec_semaphore; - pool->task_semaphore = task_semaphore; - pool->main_semaphore = main_semaphore; - pool->barrier = barrier_alloc(worker_count); - pool->is_live = 1; - pool->worker_count = worker_count; - pool->worker_arr = push_array(arena, TP_Worker, worker_count); + TP_Context *pool = push_array(arena, TP_Context, 1); + pool->exec_semaphore = exec_semaphore; + pool->barrier = barrier_alloc(worker_count); + pool->is_live = 1; + 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) { + U64 drop_count64 = Min(task_count, pool->worker_count); + U32 drop_count = safe_cast_u32(drop_count64); + semaphore_drop_count(pool->exec_semaphore, drop_count); + } - // if we are in shared mode ping local semaphore - if (pool->exec_semaphore.u64[0] != 0) { - for (U64 worker_idx = 0; worker_idx < drop_count; worker_idx +=1) { - 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]); - - // wait for workers to finish tasks - semaphore_take(pool->main_semaphore, max_U64); + tp_run_tasks(pool, pool->worker_arr); } } diff --git a/src/linux/base/linux_base.c b/src/linux/base/linux_base.c index 44212e92..67c26313 100644 --- a/src/linux/base/linux_base.c +++ b/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; - if(semaphore.u64[0] != 0) - { - err = LNX_RETRY_ON_EINTR(sem_post((sem_t*)semaphore.u64[0])); - Assert(err == 0); + for EachIndex(i, count) { + int err = -1; + if(semaphore.u64[0] != 0) + { + err = LNX_RETRY_ON_EINTR(sem_post((sem_t*)*semaphore.u64)); + Assert(err == 0); + } } } diff --git a/src/win32/base/win32_base.c b/src/win32/base/win32_base.c index f77846c0..d2f9171a 100644 --- a/src/win32/base/win32_base.c +++ b/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, 1, 0); + ReleaseSemaphore((HANDLE)*semaphore.u64, drop_count, 0); } //- rjf: barriers