diff options
Diffstat (limited to 'src/thread_pool.cpp')
| -rw-r--r-- | src/thread_pool.cpp | 55 |
1 files changed, 28 insertions, 27 deletions
diff --git a/src/thread_pool.cpp b/src/thread_pool.cpp index 727cdcdda..6df991d7d 100644 --- a/src/thread_pool.cpp +++ b/src/thread_pool.cpp @@ -1,14 +1,25 @@ // thread_pool.cpp +struct WorkerTask; +struct ThreadPool; + #define WORKER_TASK_PROC(name) isize name(void *data) typedef WORKER_TASK_PROC(WorkerTaskProc); +gb_internal void thread_pool_init(ThreadPool *pool, gbAllocator const &a, isize thread_count, char const *worker_name); +gb_internal void thread_pool_destroy(ThreadPool *pool); +gb_internal bool thread_pool_add_task(ThreadPool *pool, WorkerTaskProc *proc, void *data); +gb_internal void thread_pool_wait(ThreadPool *pool); + + struct WorkerTask { WorkerTask * next; WorkerTaskProc *do_work; void * data; + ThreadPool * pool; }; + struct ThreadPool { gbAllocator allocator; BlockingMutex mutex; @@ -23,9 +34,9 @@ struct ThreadPool { }; -THREAD_PROC(thread_pool_thread_proc); +gb_internal THREAD_PROC(thread_pool_thread_proc); -void thread_pool_init(ThreadPool *pool, gbAllocator const &a, isize thread_count, char const *worker_name) { +gb_internal void thread_pool_init(ThreadPool *pool, gbAllocator const &a, isize thread_count, char const *worker_name) { pool->allocator = a; pool->stop = false; mutex_init(&pool->mutex); @@ -34,16 +45,11 @@ void thread_pool_init(ThreadPool *pool, gbAllocator const &a, isize thread_count slice_init(&pool->threads, a, thread_count); for_array(i, pool->threads) { Thread *t = &pool->threads[i]; - thread_init(t); - } - - for_array(i, pool->threads) { - Thread *t = &pool->threads[i]; - thread_start(t, thread_pool_thread_proc, pool); + thread_init_and_start(t, thread_pool_thread_proc, pool); } } -void thread_pool_destroy(ThreadPool *pool) { +gb_internal void thread_pool_destroy(ThreadPool *pool) { mutex_lock(&pool->mutex); pool->stop = true; condition_broadcast(&pool->task_cond); @@ -51,65 +57,60 @@ void thread_pool_destroy(ThreadPool *pool) { for_array(i, pool->threads) { Thread *t = &pool->threads[i]; - thread_join(t); + thread_join_and_destroy(t); } - - for_array(i, pool->threads) { - Thread *t = &pool->threads[i]; - thread_destroy(t); - } - + gb_free(pool->allocator, pool->threads.data); mutex_destroy(&pool->mutex); condition_destroy(&pool->task_cond); } -bool thread_pool_queue_empty(ThreadPool *pool) { +gb_internal bool thread_pool_queue_empty(ThreadPool *pool) { return pool->task_queue == nullptr; } -WorkerTask *thread_pool_queue_pop(ThreadPool *pool) { +gb_internal WorkerTask *thread_pool_queue_pop(ThreadPool *pool) { GB_ASSERT(pool->task_queue != nullptr); WorkerTask *task = pool->task_queue; pool->task_queue = task->next; return task; } -void thread_pool_queue_push(ThreadPool *pool, WorkerTask *task) { +gb_internal void thread_pool_queue_push(ThreadPool *pool, WorkerTask *task) { GB_ASSERT(task != nullptr); task->next = pool->task_queue; pool->task_queue = task; } -bool thread_pool_add_task(ThreadPool *pool, WorkerTaskProc *proc, void *data) { +gb_internal bool thread_pool_add_task(ThreadPool *pool, WorkerTaskProc *proc, void *data) { GB_ASSERT(proc != nullptr); - mutex_lock(&pool->mutex); WorkerTask *task = gb_alloc_item(permanent_allocator(), WorkerTask); if (task == nullptr) { - mutex_unlock(&pool->mutex); GB_PANIC("Out of memory"); return false; } + task->pool = pool; task->do_work = proc; task->data = data; + mutex_lock(&pool->mutex); thread_pool_queue_push(pool, task); GB_ASSERT(pool->ready >= 0); - pool->ready++; + pool->ready.fetch_add(1); condition_broadcast(&pool->task_cond); mutex_unlock(&pool->mutex); return true; } -void thread_pool_do_task(WorkerTask *task) { +gb_internal void thread_pool_do_task(WorkerTask *task) { task->do_work(task->data); } -void thread_pool_wait(ThreadPool *pool) { +gb_internal void thread_pool_wait(ThreadPool *pool) { if (pool->threads.count == 0) { while (!thread_pool_queue_empty(pool)) { thread_pool_do_task(thread_pool_queue_pop(pool)); - --pool->ready; + pool->ready.fetch_sub(1); } GB_ASSERT(pool->ready == 0); return; @@ -138,7 +139,7 @@ void thread_pool_wait(ThreadPool *pool) { } -THREAD_PROC(thread_pool_thread_proc) { +gb_internal THREAD_PROC(thread_pool_thread_proc) { ThreadPool *pool = cast(ThreadPool *)thread->user_data; for (;;) { |