From d1682dfc3cf767fb75465a3f69275fc126579559 Mon Sep 17 00:00:00 2001 From: J-P Nurmi Date: Fri, 17 Jul 2026 16:51:35 +0200 Subject: [PATCH] feat(internal): add thread pool Add a bounded thread pool that runs tasks in parallel and invokes completion callbacks in submission order. Add unit coverage for ordered parallel execution. --- src/sentry_sync.c | 275 +++++++++++++++++++++++++++++++++++++++++ src/sentry_sync.h | 52 ++++++++ tests/unit/test_sync.c | 127 +++++++++++++++++++ tests/unit/tests.inc | 3 + 4 files changed, 457 insertions(+) diff --git a/src/sentry_sync.c b/src/sentry_sync.c index b63698396a..17d4ba01a6 100644 --- a/src/sentry_sync.c +++ b/src/sentry_sync.c @@ -1,6 +1,7 @@ #include "sentry_sync.h" #include "sentry_alloc.h" #include "sentry_core.h" +#include "sentry_cpu_relax.h" #include "sentry_string.h" #include "sentry_utils.h" #include @@ -107,6 +108,280 @@ sentry__thread_setname(sentry_threadid_t thread_id, const char *thread_name) return thread_setname(thread_id, thread_name); } +typedef struct sentry_threadpool_task_s { + struct sentry_threadpool_task_s *next; + void (*exec_func)(void *task_data); + void (*complete_func)(void *task_data); + void (*cleanup_func)(void *task_data); + void *task_data; + bool done; +} sentry_threadpool_task_t; + +struct sentry_threadpool_s { + sentry_threadid_t *threads; + char *thread_name; + size_t thread_count; + size_t started_threads; + sentry_mutex_t lock; + sentry_cond_t work_signal; + sentry_cond_t state_signal; + sentry_threadpool_task_t *first_task; + sentry_threadpool_task_t *last_task; + sentry_threadpool_task_t *next_task; + long pending; + long index; + bool running; + bool stopping; + bool committing; +}; + +static void +threadpool_task_free(sentry_threadpool_task_t *task) +{ + if (task->cleanup_func) { + task->cleanup_func(task->task_data); + } + sentry_free(task); +} + +static void +threadpool_wake_all(sentry_threadpool_t *pool) +{ + for (size_t i = 0; i < pool->started_threads; i++) { + sentry__cond_wake(&pool->work_signal); + } +} + +static void +threadpool_commit_ready(sentry_threadpool_t *pool) +{ + if (pool->committing) { + return; + } + pool->committing = true; + + while (pool->first_task && pool->first_task->done) { + sentry_threadpool_task_t *task = pool->first_task; + pool->first_task = task->next; + if (!pool->first_task) { + pool->last_task = NULL; + } + + sentry__mutex_unlock(&pool->lock); + if (task->complete_func) { + task->complete_func(task->task_data); + } + threadpool_task_free(task); + sentry__mutex_lock(&pool->lock); + + sentry__atomic_fetch_and_add(&pool->pending, -1); + sentry__cond_wake(&pool->state_signal); + } + + pool->committing = false; + if (sentry__atomic_fetch(&pool->pending) == 0) { + threadpool_wake_all(pool); + } +} + +SENTRY_THREAD_FN +threadpool_worker(void *data) +{ + sentry_threadpool_t *pool = data; + if (pool->thread_name) { + const long index = sentry__atomic_fetch_and_add(&pool->index, 1); + char thread_name[16]; + snprintf(thread_name, sizeof(thread_name), "%s-%ld", pool->thread_name, + index); + sentry__thread_setname(sentry__current_thread(), thread_name); + } + + while (true) { + sentry__mutex_lock(&pool->lock); + sentry_threadpool_task_t *task = pool->next_task; + while (!task) { + if (pool->stopping && sentry__atomic_fetch(&pool->pending) == 0) { + sentry__mutex_unlock(&pool->lock); + return 0; + } + sentry__cond_wait(&pool->work_signal, &pool->lock); + task = pool->next_task; + } + pool->next_task = task->next; + sentry__mutex_unlock(&pool->lock); + + task->exec_func(task->task_data); + + sentry__mutex_lock(&pool->lock); + task->done = true; + threadpool_commit_ready(pool); + sentry__cond_wake(&pool->work_signal); + sentry__mutex_unlock(&pool->lock); + } +} + +sentry_threadpool_t * +sentry__threadpool_new(size_t thread_count) +{ + if (thread_count == 0) { + return NULL; + } + sentry_threadpool_t *pool = SENTRY_MAKE(sentry_threadpool_t); + if (!pool) { + return NULL; + } + pool->threads = sentry__calloc(thread_count, sizeof(sentry_threadid_t)); + if (!pool->threads) { + sentry_free(pool); + return NULL; + } + pool->thread_count = thread_count; + sentry__mutex_init(&pool->lock); + sentry__cond_init(&pool->work_signal); + sentry__cond_init(&pool->state_signal); + for (size_t i = 0; i < thread_count; i++) { + sentry__thread_init(&pool->threads[i]); + } + return pool; +} + +void +sentry__threadpool_setname(sentry_threadpool_t *pool, const char *thread_name) +{ + if (!pool) { + return; + } + sentry_free(pool->thread_name); + pool->thread_name = sentry__string_clone(thread_name); +} + +int +sentry__threadpool_start(sentry_threadpool_t *pool) +{ + if (!pool || pool->running) { + return pool ? 0 : 1; + } + pool->running = true; + pool->stopping = false; + for (size_t i = 0; i < pool->thread_count; i++) { + if (sentry__thread_spawn(&pool->threads[i], threadpool_worker, pool) + != 0) { + sentry__mutex_lock(&pool->lock); + pool->stopping = true; + threadpool_wake_all(pool); + sentry__mutex_unlock(&pool->lock); + for (size_t j = 0; j < pool->started_threads; j++) { + sentry__thread_join(pool->threads[j]); + } + pool->started_threads = 0; + pool->running = false; + return 1; + } + pool->started_threads++; + } + return 0; +} + +int +sentry__threadpool_submit(sentry_threadpool_t *pool, + void (*exec_func)(void *task_data), void (*complete_func)(void *task_data), + void (*cleanup_func)(void *task_data), void *task_data) +{ + if (!pool || !exec_func) { + if (cleanup_func) { + cleanup_func(task_data); + } + return 1; + } + sentry_threadpool_task_t *task = SENTRY_MAKE(sentry_threadpool_task_t); + if (!task) { + if (cleanup_func) { + cleanup_func(task_data); + } + return 1; + } + task->exec_func = exec_func; + task->complete_func = complete_func; + task->cleanup_func = cleanup_func; + task->task_data = task_data; + + sentry__mutex_lock(&pool->lock); + if (!pool->running || pool->stopping) { + sentry__mutex_unlock(&pool->lock); + threadpool_task_free(task); + return 1; + } + + if (pool->last_task) { + pool->last_task->next = task; + } else { + pool->first_task = task; + } + pool->last_task = task; + if (!pool->next_task) { + pool->next_task = task; + } + sentry__atomic_fetch_and_add(&pool->pending, 1); + sentry__cond_wake(&pool->work_signal); + sentry__mutex_unlock(&pool->lock); + return 0; +} + +void +sentry__threadpool_flush(sentry_threadpool_t *pool) +{ + if (!pool || !pool->running) { + return; + } + sentry__mutex_lock(&pool->lock); + while (sentry__atomic_fetch(&pool->pending) > 0) { + sentry__cond_wait(&pool->state_signal, &pool->lock); + } + sentry__mutex_unlock(&pool->lock); +} + +void +sentry__threadpool_shutdown(sentry_threadpool_t *pool) +{ + if (!pool || !pool->running) { + return; + } + sentry__mutex_lock(&pool->lock); + pool->stopping = true; + threadpool_wake_all(pool); + sentry__cond_wake(&pool->state_signal); + sentry__mutex_unlock(&pool->lock); + + for (size_t i = 0; i < pool->started_threads; i++) { + sentry__thread_join(pool->threads[i]); + } + pool->started_threads = 0; + pool->running = false; + pool->index = 0; +} + +void +sentry__threadpool_free(sentry_threadpool_t *pool) +{ + if (!pool) { + return; + } + sentry__threadpool_shutdown(pool); + sentry_threadpool_task_t *task = pool->first_task; + while (task) { + sentry_threadpool_task_t *next = task->next; + threadpool_task_free(task); + task = next; + } + for (size_t i = 0; i < pool->thread_count; i++) { + sentry__thread_free(&pool->threads[i]); + } + sentry_free(pool->thread_name); + sentry__mutex_free(&pool->lock); + sentry_free(pool->threads); + sentry_free(pool); +} + /** * Queue operations, locking and Reference counting: * diff --git a/src/sentry_sync.h b/src/sentry_sync.h index 62028777d4..aaa9fab411 100644 --- a/src/sentry_sync.h +++ b/src/sentry_sync.h @@ -3,6 +3,7 @@ #include "sentry_boot.h" #include "sentry_core.h" +#include "sentry_cpu_relax.h" #include #include @@ -431,6 +432,35 @@ sentry__atomic_compare_swap(volatile long *val, long expected, long desired) #endif } +typedef bool (*sentry_spin_wait_func_t)(int attempt, void *data); + +static inline void +sentry__spin_lock(volatile long *lock) +{ + while (!sentry__atomic_compare_swap(lock, 0, 1)) { + sentry__cpu_relax(); + } +} + +static inline bool +sentry__spin_lock_wait( + volatile long *lock, sentry_spin_wait_func_t wait_func, void *data) +{ + int attempts = 0; + while (!sentry__atomic_compare_swap(lock, 0, 1)) { + if (!wait_func || !wait_func(++attempts, data)) { + return false; + } + } + return true; +} + +static inline void +sentry__spin_unlock(volatile long *lock) +{ + sentry__atomic_store(lock, 0); +} + /** * 64-bit variants of the atomic helpers above. The `long`-based helpers are * only 32 bits wide on Windows and 32-bit POSIX targets, so callers that need @@ -471,8 +501,30 @@ int sentry__thread_setname( struct sentry_bgworker_s; typedef struct sentry_bgworker_s sentry_bgworker_t; +struct sentry_threadpool_s; +typedef struct sentry_threadpool_s sentry_threadpool_t; + typedef void (*sentry_task_exec_func_t)(void *task_data, void *state); +/** + * Creates a thread pool. Tasks execute in parallel, while completion callbacks + * run in submission order. + */ +sentry_threadpool_t *sentry__threadpool_new(size_t thread_count); +void sentry__threadpool_setname( + sentry_threadpool_t *pool, const char *thread_name); +int sentry__threadpool_start(sentry_threadpool_t *pool); +/** + * Takes ownership of `task_data`, freeing it using `cleanup_func` when the + * task is completed, cancelled, or rejected. + */ +int sentry__threadpool_submit(sentry_threadpool_t *pool, + void (*exec_func)(void *task_data), void (*complete_func)(void *task_data), + void (*cleanup_func)(void *task_data), void *task_data); +void sentry__threadpool_flush(sentry_threadpool_t *pool); +void sentry__threadpool_shutdown(sentry_threadpool_t *pool); +void sentry__threadpool_free(sentry_threadpool_t *pool); + /** * Creates a new background worker thread. * diff --git a/tests/unit/test_sync.c b/tests/unit/test_sync.c index 98c5eeef91..4d78b70ed8 100644 --- a/tests/unit/test_sync.c +++ b/tests/unit/test_sync.c @@ -581,6 +581,133 @@ SENTRY_TEST(bgworker_delayed_shutdown) sentry__bgworker_decref(bgw); } +struct threadpool_test_state { + volatile long first_started; + volatile long release_first; + volatile long second_ran; + bool first_timed_out; + int completion_order[2]; + int completion_count; + int cleanup_count; +}; + +struct threadpool_test_task { + struct threadpool_test_state *state; + int id; +}; + +static void +threadpool_test_exec(void *data) +{ + struct threadpool_test_task *task = data; + struct threadpool_test_state *state = task->state; + const uint64_t deadline = sentry__monotonic_time() + 1000; + + if (task->id == 0) { + sentry__atomic_store(&state->first_started, 1); + while (!sentry__atomic_fetch(&state->release_first)) { + if (sentry__monotonic_time() >= deadline) { + state->first_timed_out = true; + break; + } + sleep_ms(1); + } + } else { + while (!sentry__atomic_fetch(&state->first_started) + && sentry__monotonic_time() < deadline) { + sleep_ms(1); + } + sentry__atomic_store(&state->second_ran, 1); + sentry__atomic_store(&state->release_first, 1); + } +} + +static void +threadpool_test_complete(void *data) +{ + struct threadpool_test_task *task = data; + struct threadpool_test_state *state = task->state; + state->completion_order[state->completion_count++] = task->id; +} + +static void +threadpool_test_cleanup(void *data) +{ + struct threadpool_test_task *task = data; + task->state->cleanup_count++; +} + +SENTRY_TEST(threadpool_ordered_parallel) +{ + struct threadpool_test_state state = { 0 }; + struct threadpool_test_task tasks[] = { + { &state, 0 }, + { &state, 1 }, + }; + sentry_threadpool_t *pool = sentry__threadpool_new(2); + TEST_ASSERT(!!pool); + TEST_ASSERT(sentry__threadpool_start(pool) == 0); + + for (size_t i = 0; i < 2; i++) { + TEST_ASSERT( + sentry__threadpool_submit(pool, threadpool_test_exec, + threadpool_test_complete, threadpool_test_cleanup, &tasks[i]) + == 0); + } + sentry__threadpool_flush(pool); + + TEST_CHECK(sentry__atomic_fetch(&state.second_ran)); + TEST_CHECK(!state.first_timed_out); + TEST_CHECK_INT_EQUAL(state.completion_count, 2); + TEST_CHECK_INT_EQUAL(state.completion_order[0], 0); + TEST_CHECK_INT_EQUAL(state.completion_order[1], 1); + TEST_CHECK_INT_EQUAL(state.cleanup_count, 2); + + sentry__threadpool_shutdown(pool); + sentry__threadpool_free(pool); +} + +SENTRY_TEST(threadpool_rejected_submit_cleans_up) +{ + struct threadpool_test_state state = { 0 }; + struct threadpool_test_task task = { &state, 0 }; + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + + TEST_CHECK(sentry__threadpool_submit(pool, threadpool_test_exec, + threadpool_test_complete, threadpool_test_cleanup, &task) + != 0); + TEST_CHECK_INT_EQUAL(state.cleanup_count, 1); + TEST_CHECK_INT_EQUAL(state.completion_count, 0); + + sentry__threadpool_free(pool); +} + +static long g_spin_waits = 0; + +static bool +spin_wait(int UNUSED(attempt), void *UNUSED(data)) +{ + sentry__atomic_fetch_and_add(&g_spin_waits, 1); + return sentry__atomic_fetch(&g_spin_waits) < 2; +} + +SENTRY_TEST(spin_lock) +{ + long lock = 0; + sentry__atomic_store(&g_spin_waits, 0); + + sentry__spin_lock(&lock); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&lock), 1); + TEST_CHECK(!sentry__spin_lock_wait(&lock, spin_wait, NULL)); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&g_spin_waits), 2); + sentry__spin_unlock(&lock); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&lock), 0); + TEST_CHECK(sentry__spin_lock_wait(&lock, spin_wait, NULL)); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&g_spin_waits), 2); + sentry__spin_unlock(&lock); +} + SENTRY_TEST(cond_wait_timeout_overflow) { #if !(defined(SENTRY_PLATFORM_MACOS) \ diff --git a/tests/unit/tests.inc b/tests/unit/tests.inc index 1237f487a3..6666b5580b 100644 --- a/tests/unit/tests.inc +++ b/tests/unit/tests.inc @@ -370,6 +370,7 @@ XX(span_data_n) XX(span_tagging) XX(span_tagging_n) XX(spans_on_scope) +XX(spin_lock) XX(stack_guarantee) XX(stack_guarantee_auto_init) XX(strict_continuation_asymmetric_lenient_continues) @@ -384,6 +385,8 @@ XX(stringbuilder_reserve_overflow) XX(symbolizer) XX(task_queue) XX(thread_without_name_still_valid) +XX(threadpool_ordered_parallel) +XX(threadpool_rejected_submit_cleans_up) XX(trace_continuation_truth_table) XX(trace_finish) XX(traceparent_header_disabled_by_default)