diff --git a/src/sentry_sync.c b/src/sentry_sync.c index b63698396..b486aaafd 100644 --- a/src/sentry_sync.c +++ b/src/sentry_sync.c @@ -107,6 +107,346 @@ 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; + long 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) +{ + sentry__cond_wake_all(&pool->work_signal); +} + +static bool +is_pooled_thread(sentry_threadpool_t *pool) +{ + const sentry_threadid_t current = sentry__current_thread(); + for (size_t i = 0; i < pool->started_threads; i++) { + if (sentry__threadid_equal(current, pool->threads[i])) { + return true; + } + } + return false; +} + +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_thread(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) { + return 1; + } + + sentry__mutex_lock(&pool->lock); + while (sentry__atomic_fetch(&pool->running) && pool->stopping) { + sentry__cond_wait(&pool->state_signal, &pool->lock); + } + if (sentry__atomic_fetch(&pool->running)) { + sentry__mutex_unlock(&pool->lock); + return 0; + } + + sentry__atomic_store(&pool->running, 1); + pool->stopping = false; + for (size_t i = 0; i < pool->thread_count; i++) { + if (sentry__thread_spawn(&pool->threads[i], threadpool_thread, pool) + != 0) { + pool->stopping = true; + const size_t started_threads = pool->started_threads; + threadpool_wake_all(pool); + sentry__mutex_unlock(&pool->lock); + + for (size_t j = 0; j < started_threads; j++) { + sentry__thread_join(pool->threads[j]); + } + + sentry__mutex_lock(&pool->lock); + for (size_t j = 0; j < started_threads; j++) { + sentry__thread_free(&pool->threads[j]); + } + pool->started_threads = 0; + pool->index = 0; + sentry__atomic_store(&pool->running, 0); + sentry__cond_wake_all(&pool->state_signal); + sentry__mutex_unlock(&pool->lock); + return 1; + } + pool->started_threads++; + } + sentry__mutex_unlock(&pool->lock); + 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 (!sentry__atomic_fetch(&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 || !sentry__atomic_fetch(&pool->running)) { + return; + } + + sentry__mutex_lock(&pool->lock); + if (is_pooled_thread(pool)) { + sentry__mutex_unlock(&pool->lock); + SENTRY_WARN("cannot flush thread pool from a pooled thread"); + return; + } + 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 || !sentry__atomic_fetch(&pool->running)) { + return; + } + + sentry__mutex_lock(&pool->lock); + if (is_pooled_thread(pool)) { + sentry__mutex_unlock(&pool->lock); + SENTRY_WARN("cannot shut down thread pool from a pooled thread"); + return; + } + if (pool->stopping) { + while (sentry__atomic_fetch(&pool->running)) { + sentry__cond_wait(&pool->state_signal, &pool->lock); + } + sentry__mutex_unlock(&pool->lock); + return; + } + pool->stopping = true; + const size_t started_threads = pool->started_threads; + threadpool_wake_all(pool); + sentry__cond_wake(&pool->state_signal); + sentry__mutex_unlock(&pool->lock); + + for (size_t i = 0; i < started_threads; i++) { + sentry__thread_join(pool->threads[i]); + } + + sentry__mutex_lock(&pool->lock); + for (size_t i = 0; i < started_threads; i++) { + sentry__thread_free(&pool->threads[i]); + } + pool->started_threads = 0; + pool->index = 0; + sentry__atomic_store(&pool->running, 0); + sentry__cond_wake_all(&pool->state_signal); + sentry__mutex_unlock(&pool->lock); +} + +void +sentry__threadpool_free(sentry_threadpool_t *pool) +{ + if (!pool) { + return; + } + if (sentry__atomic_fetch(&pool->running)) { + sentry__mutex_lock(&pool->lock); + if (is_pooled_thread(pool)) { + sentry__mutex_unlock(&pool->lock); + SENTRY_WARN("cannot free thread pool from a pooled thread"); + return; + } + sentry__mutex_unlock(&pool->lock); + } + 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 1a9c48a6c..46b51c458 100644 --- a/src/sentry_sync.h +++ b/src/sentry_sync.h @@ -492,8 +492,67 @@ 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 configured with `thread_count` threads. Tasks execute + * in parallel, while completion callbacks run in submission order. + */ +sentry_threadpool_t *sentry__threadpool_new(size_t thread_count); + +/** + * Sets a name for pooled threads. Each thread is named `-`, + * where `index` starts at 0. + * + * Should be executed before thread pool start. + */ +void sentry__threadpool_setname( + sentry_threadpool_t *pool, const char *thread_name); + +/** + * Starts the pooled threads. Calling this on a running pool succeeds without + * effect. A pool can be restarted after shutdown. + * + * Returns 0 on success, or a non-zero value if `pool` is `NULL` or the pooled + * threads cannot be started. + */ +int sentry__threadpool_start(sentry_threadpool_t *pool); + +/** + * Submits a task for execution. `exec_func` runs on a pooled thread. The + * optional `complete_func` runs after execution, in submission order, followed + * by the optional `cleanup_func`. + * + * Takes ownership of `task_data` on every call. If the task is rejected, + * `cleanup_func` is called immediately when provided. + * + * Returns 0 if the task was accepted, or a non-zero value if the arguments are + * invalid, the pool is not running or is stopping, or allocation fails. + */ +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); + +/** + * Blocks until all accepted tasks and their completion and cleanup callbacks + * have finished. Does nothing when called from one of the pool's threads. + */ +void sentry__threadpool_flush(sentry_threadpool_t *pool); + +/** + * Stops accepting tasks, drains the queue, and joins all pooled threads. The + * pool can be started again after shutdown. + */ +void sentry__threadpool_shutdown(sentry_threadpool_t *pool); + +/** + * Shuts down the pool if necessary and releases its resources. + */ +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 be9e4b657..5af41c66f 100644 --- a/tests/unit/test_sync.c +++ b/tests/unit/test_sync.c @@ -582,6 +582,486 @@ SENTRY_TEST(bgworker_delayed_shutdown) sentry__bgworker_decref(bgw); } +struct threadpool_test_state { + sentry_mutex_t lock; + sentry_cond_t cond; + volatile long first_started; + volatile long release_first; + volatile long second_ran; + 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; + + if (task->id == 0) { + sentry__mutex_lock(&state->lock); + sentry__atomic_store(&state->first_started, 1); + sentry__cond_wake(&state->cond); + while (!sentry__atomic_fetch(&state->release_first)) { + sentry__cond_wait(&state->cond, &state->lock); + } + sentry__mutex_unlock(&state->lock); + } else { + sentry__mutex_lock(&state->lock); + while (!sentry__atomic_fetch(&state->first_started)) { + sentry__cond_wait(&state->cond, &state->lock); + } + sentry__atomic_store(&state->second_ran, 1); + sentry__atomic_store(&state->release_first, 1); + sentry__cond_wake(&state->cond); + sentry__mutex_unlock(&state->lock); + } +} + +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); + sentry__mutex_init(&state.lock); + sentry__cond_init(&state.cond); + 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_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); +#ifndef SENTRY_PLATFORM_WINDOWS + pthread_cond_destroy(&state.cond); +#endif + sentry__mutex_free(&state.lock); +} + +struct threadpool_restart_state { + int executed; + int completed; + int cleaned_up; +}; + +static void +threadpool_restart_exec(void *data) +{ + struct threadpool_restart_state *state = data; + state->executed++; +} + +static void +threadpool_restart_complete(void *data) +{ + struct threadpool_restart_state *state = data; + state->completed++; +} + +static void +threadpool_restart_cleanup(void *data) +{ + struct threadpool_restart_state *state = data; + state->cleaned_up++; +} + +SENTRY_TEST(threadpool_restart) +{ + struct threadpool_restart_state state = { 0 }; + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + + for (int i = 0; i < 2; i++) { + TEST_ASSERT(sentry__threadpool_start(pool) == 0); + TEST_ASSERT( + sentry__threadpool_submit(pool, threadpool_restart_exec, + threadpool_restart_complete, threadpool_restart_cleanup, &state) + == 0); + sentry__threadpool_flush(pool); + sentry__threadpool_shutdown(pool); + } + + TEST_CHECK_INT_EQUAL(state.executed, 2); + TEST_CHECK_INT_EQUAL(state.completed, 2); + TEST_CHECK_INT_EQUAL(state.cleaned_up, 2); + + sentry__threadpool_free(pool); +} + +struct threadpool_count_state { + volatile long executed; + volatile long completed; + volatile long cleaned_up; +}; + +static void +threadpool_count_exec(void *data) +{ + struct threadpool_count_state *state = data; + sentry__atomic_fetch_and_add(&state->executed, 1); +} + +static void +threadpool_count_complete(void *data) +{ + struct threadpool_count_state *state = data; + sentry__atomic_fetch_and_add(&state->completed, 1); +} + +static void +threadpool_count_cleanup(void *data) +{ + struct threadpool_count_state *state = data; + sentry__atomic_fetch_and_add(&state->cleaned_up, 1); +} + +struct threadpool_reentry_state { + sentry_mutex_t lock; + sentry_cond_t cond; + bool first_complete_entered; + bool second_exec_done; + bool first_complete_left; + volatile long second_complete_ran; + volatile long second_complete_before_first_left; + volatile long completion_count; + volatile long cleanup_count; + int completion_order[2]; +}; + +struct threadpool_reentry_task { + struct threadpool_reentry_state *state; + int id; +}; + +static void +threadpool_reentry_exec(void *data) +{ + struct threadpool_reentry_task *task = data; + struct threadpool_reentry_state *state = task->state; + + if (task->id != 1) { + return; + } + + sentry__mutex_lock(&state->lock); + while (!state->first_complete_entered) { + sentry__cond_wait(&state->cond, &state->lock); + } + state->second_exec_done = true; + sentry__cond_wake(&state->cond); + sentry__mutex_unlock(&state->lock); +} + +static void +threadpool_reentry_complete(void *data) +{ + struct threadpool_reentry_task *task = data; + struct threadpool_reentry_state *state = task->state; + long pos = sentry__atomic_fetch_and_add(&state->completion_count, 1); + if (pos < 2) { + state->completion_order[pos] = task->id; + } + + if (task->id == 0) { + sentry__mutex_lock(&state->lock); + state->first_complete_entered = true; + sentry__cond_wake(&state->cond); + while (!state->second_exec_done) { + sentry__cond_wait(&state->cond, &state->lock); + } + state->first_complete_left = true; + sentry__mutex_unlock(&state->lock); + } else { + sentry__mutex_lock(&state->lock); + bool first_complete_left = state->first_complete_left; + sentry__mutex_unlock(&state->lock); + if (!first_complete_left) { + sentry__atomic_store(&state->second_complete_before_first_left, 1); + } + } + + if (task->id == 1) { + sentry__atomic_store(&state->second_complete_ran, 1); + } +} + +static void +threadpool_reentry_cleanup(void *data) +{ + struct threadpool_reentry_task *task = data; + sentry__atomic_fetch_and_add(&task->state->cleanup_count, 1); +} + +SENTRY_TEST(threadpool_commit_reentry) +{ + struct threadpool_reentry_state state = { 0 }; + struct threadpool_reentry_task tasks[] = { + { &state, 0 }, + { &state, 1 }, + }; + sentry_threadpool_t *pool = sentry__threadpool_new(2); + TEST_ASSERT(!!pool); + sentry__mutex_init(&state.lock); + sentry__cond_init(&state.cond); + TEST_ASSERT(sentry__threadpool_start(pool) == 0); + + for (size_t i = 0; i < 2; i++) { + TEST_ASSERT(sentry__threadpool_submit(pool, threadpool_reentry_exec, + threadpool_reentry_complete, threadpool_reentry_cleanup, + &tasks[i]) + == 0); + } + sentry__threadpool_flush(pool); + + TEST_CHECK(sentry__atomic_fetch(&state.second_complete_ran)); + TEST_CHECK(!sentry__atomic_fetch(&state.second_complete_before_first_left)); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&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(sentry__atomic_fetch(&state.cleanup_count), 2); + + sentry__threadpool_shutdown(pool); + sentry__threadpool_free(pool); +#ifndef SENTRY_PLATFORM_WINDOWS + pthread_cond_destroy(&state.cond); +#endif + sentry__mutex_free(&state.lock); +} + +struct callback_pool_state { + sentry_threadpool_t *pool; + volatile long flush_calls; + volatile long shutdown_calls; + volatile long free_calls; +}; + +static void +callback_flush(void *data) +{ + struct callback_pool_state *state = data; + sentry__threadpool_flush(state->pool); + sentry__atomic_fetch_and_add(&state->flush_calls, 1); +} + +static void +callback_shutdown(void *data) +{ + struct callback_pool_state *state = data; + sentry__threadpool_shutdown(state->pool); + sentry__atomic_fetch_and_add(&state->shutdown_calls, 1); +} + +static void +threadpool_noop_exec(void *UNUSED(data)) +{ +} + +static void +callback_free(void *data) +{ + struct callback_pool_state *state = data; + sentry__threadpool_free(state->pool); + sentry__atomic_fetch_and_add(&state->free_calls, 1); +} + +SENTRY_TEST(threadpool_guard) +{ + struct callback_pool_state state = { 0 }; + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + state.pool = pool; + TEST_ASSERT(sentry__threadpool_start(pool) == 0); + TEST_ASSERT(sentry__threadpool_submit(pool, callback_flush, callback_flush, + callback_flush, &state) + == 0); + TEST_ASSERT(sentry__threadpool_submit( + pool, threadpool_noop_exec, callback_shutdown, NULL, &state) + == 0); + TEST_ASSERT(sentry__threadpool_submit( + pool, threadpool_noop_exec, callback_free, NULL, &state) + == 0); + + sentry__threadpool_flush(pool); + + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.flush_calls), 3); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.shutdown_calls), 1); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.free_calls), 1); + + sentry__threadpool_shutdown(pool); + sentry__threadpool_free(pool); +} + +SENTRY_TEST(threadpool_invalid_args) +{ + struct threadpool_count_state state = { 0 }; + + TEST_CHECK_PTR_EQUAL(sentry__threadpool_new(0), NULL); + if (sizeof(sentry_threadid_t) > 1) { + TEST_CHECK_PTR_EQUAL( + sentry__threadpool_new(SIZE_MAX / sizeof(sentry_threadid_t) + 1), + NULL); + } + + TEST_CHECK(sentry__threadpool_start(NULL) != 0); + TEST_CHECK(sentry__threadpool_submit(NULL, threadpool_count_exec, + threadpool_count_complete, threadpool_count_cleanup, &state) + != 0); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.cleaned_up), 1); + sentry__threadpool_setname(NULL, "ignored"); + sentry__threadpool_flush(NULL); + sentry__threadpool_shutdown(NULL); + sentry__threadpool_free(NULL); + + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + TEST_ASSERT(sentry__threadpool_start(pool) == 0); + TEST_CHECK_INT_EQUAL(sentry__threadpool_start(pool), 0); + TEST_CHECK(sentry__threadpool_submit(pool, NULL, threadpool_count_complete, + threadpool_count_cleanup, &state) + != 0); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.cleaned_up), 2); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.completed), 0); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.executed), 0); + + sentry__threadpool_shutdown(pool); + sentry__threadpool_shutdown(pool); + sentry__threadpool_free(pool); +} + +SENTRY_TEST(threadpool_no_completion_callback_cleans_up) +{ + struct threadpool_count_state state = { 0 }; + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + TEST_ASSERT(sentry__threadpool_start(pool) == 0); + + TEST_ASSERT(sentry__threadpool_submit(pool, threadpool_count_exec, NULL, + threadpool_count_cleanup, &state) + == 0); + sentry__threadpool_flush(pool); + + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.executed), 1); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.completed), 0); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.cleaned_up), 1); + + sentry__threadpool_shutdown(pool); + sentry__threadpool_free(pool); +} + +SENTRY_TEST(threadpool_shutdown_drains) +{ + enum { TASKS = 16 }; + struct threadpool_count_state state = { 0 }; + sentry_threadpool_t *pool = sentry__threadpool_new(2); + TEST_ASSERT(!!pool); + TEST_ASSERT(sentry__threadpool_start(pool) == 0); + + for (int i = 0; i < TASKS; i++) { + TEST_ASSERT( + sentry__threadpool_submit(pool, threadpool_count_exec, + threadpool_count_complete, threadpool_count_cleanup, &state) + == 0); + } + sentry__threadpool_shutdown(pool); + + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.executed), TASKS); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.completed), TASKS); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.cleaned_up), TASKS); + sentry__threadpool_free(pool); +} + +#if defined(SENTRY_PLATFORM_LINUX) && !defined(SENTRY_PLATFORM_ANDROID) +struct threadpool_name_state { + char thread_name[16]; + volatile long captured; +}; + +static void +threadpool_name_exec(void *data) +{ + struct threadpool_name_state *state = data; + pthread_getname_np( + pthread_self(), state->thread_name, sizeof(state->thread_name)); + sentry__atomic_store(&state->captured, 1); +} +#endif + +SENTRY_TEST(threadpool_thread_name) +{ +#if !defined(SENTRY_PLATFORM_LINUX) || defined(SENTRY_PLATFORM_ANDROID) + SKIP_TEST(); +#else + struct threadpool_name_state state = { 0 }; + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + + sentry__threadpool_setname(pool, "tp"); + TEST_ASSERT(sentry__threadpool_start(pool) == 0); + TEST_ASSERT(sentry__threadpool_submit( + pool, threadpool_name_exec, NULL, NULL, &state) + == 0); + sentry__threadpool_flush(pool); + + TEST_CHECK(sentry__atomic_fetch(&state.captured)); + TEST_CHECK_STRING_EQUAL(state.thread_name, "tp-0"); + + sentry__threadpool_shutdown(pool); + sentry__threadpool_free(pool); +#endif +} + +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); +} + #define COND_WAKE_ALL_THREADS 2 struct cond_wake_all_state { diff --git a/tests/unit/tests.inc b/tests/unit/tests.inc index bc50aec76..2fdc5eeb5 100644 --- a/tests/unit/tests.inc +++ b/tests/unit/tests.inc @@ -387,6 +387,15 @@ XX(stringbuilder_reserve_overflow) XX(symbolizer) XX(task_queue) XX(thread_without_name_still_valid) +XX(threadpool_commit_reentry) +XX(threadpool_guard) +XX(threadpool_invalid_args) +XX(threadpool_no_completion_callback_cleans_up) +XX(threadpool_ordered_parallel) +XX(threadpool_rejected_submit_cleans_up) +XX(threadpool_restart) +XX(threadpool_shutdown_drains) +XX(threadpool_thread_name) XX(trace_continuation_truth_table) XX(trace_finish) XX(traceparent_header_disabled_by_default)