feat(internal): thread pool

src/sentry_sync.c

@@ -3,6 +3,7 @@
 #include "sentry_core.h"
 #include "sentry_string.h"
 #include "sentry_utils.h"
+#include <limits.h>
 #include <stdio.h>
 #include <string.h>
 
@@ -107,6 +108,353 @@ 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 max_pending;
+    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);
+    }
+
+    pool->committing = false;
+    if (sentry__atomic_fetch(&pool->pending) == 0) {
+        sentry__cond_wake_all(&pool->state_signal);
+        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, size_t max_pending)
+{
+    if (thread_count == 0 || max_pending == 0
+        || max_pending > (size_t)LONG_MAX) {
+        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;
+    pool->max_pending = (long)max_pending;
+    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__mutex_lock(&pool->lock);
+    if (!sentry__atomic_fetch(&pool->running)) {
+        sentry_free(pool->thread_name);
+        pool->thread_name = sentry__string_clone(thread_name);
+    }
+    sentry__mutex_unlock(&pool->lock);
+}
+
+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__atomic_fetch(&pool->pending) >= pool->max_pending) {
+        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__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:
  *