一些基础组件

Foundation/c_ThreadPool.c

@@ -0,0 +1,122 @@
+#include <c_ThreadPool.h>
+#include <c_Memory.h>
+
+// Worker thread routine
+static c_Thread_Fn(thread_pool_worker(void* arg)) {
+    c_ThreadPool_t* pool = (c_ThreadPool_t*)arg;
+    c_ThreadPoolTask_t* task;
+    while (!pool->is_shutdown) {
+        task = NULL;
+
+        // Use a 500ms timeout for timedpop to allow periodic shutdown condition verification
+        if (c_LockQueue_TimedPop(&pool->task_queue, (void**)&task, pool->check_interval_ms)!=C_ERR_OK) {
+            if (task) {
+                if (task->function) {
+                    // Execute user payload safely
+                    task->function(task->argument);
+                }
+                // C_FREE(task); // Free the memory allocated for the task wrapper
+                c_Pool_Free(&pool->task_pool, task);
+            }
+        }
+    }
+    c_Thread_Exit(0);
+}
+
+/**
+ * @brief Initializes a fixed-size thread pool.
+ */
+c_err_t c_ThreadPool_Init(c_ThreadPool_t* pool, int thread_count, int queue_capacity, int task_pool_size, int check_interval_ms) {
+    if (!pool || thread_count <= 0 || queue_capacity <= 0) return C_ERR_PARAM;
+
+    pool->is_shutdown = C_FALSE;
+    pool->thread_count = thread_count;
+    pool->check_interval_ms = check_interval_ms;
+
+    if (c_Pool_Init(&pool->task_pool, sizeof(c_ThreadPoolTask_t), task_pool_size)!=C_ERR_OK) {
+        return C_ERR_FAIL;
+    }
+
+    // Allocate the thread handle array
+    pool->threads = (c_Thread_t*)C_ALLOC(sizeof(c_Thread_t) * thread_count);
+    if (!pool->threads) {
+        return C_ERR_NOMEM;
+    }
+
+    // Initialize our previously constructed cross-platform thread-safe queue
+    c_err_t err = c_LockQueue_Init(&pool->task_queue, queue_capacity, 0);
+    if (err!=C_ERR_OK) {
+        C_FREE(pool->threads);
+        return err;
+    }
+
+    // Spawn the requested worker threads
+    for (int i = 0; i < thread_count; i++) {
+        if (!c_Thread_Create(&pool->threads[i], thread_pool_worker, pool)) {
+            // Rollback strategy on failures
+            pool->is_shutdown = C_TRUE;
+            c_LockQueue_Shutdown(&pool->task_queue);
+            for (int j = 0; j < i; j++) {
+                c_Thread_Join(pool->threads[j]);
+            }
+            c_LockQueue_Destroy(&pool->task_queue);
+            C_FREE(pool->threads);
+            return C_ERR_FAIL;
+        }
+    }
+
+    return C_ERR_OK;
+}
+
+/**
+ * @brief Submits a work payload to the pool.
+ */
+c_bool_t c_ThreadPool_Submit(c_ThreadPool_t* pool, void (*function)(void*), void* argument){
+    if (!pool || !function || pool->is_shutdown) return C_FALSE;
+
+    // c_ThreadPoolTask_t* task = (c_ThreadPoolTask_t*)C_ALLOC(sizeof(*task));
+    c_ThreadPoolTask_t* task = c_Pool_Alloc(&pool->task_pool);
+    if (!task) return C_FALSE;
+
+    task->function = function;
+    task->argument = argument;
+
+    // Push the task into our thread-safe buffer. If full, this blocks the calling thread
+    if (c_LockQueue_Push(&pool->task_queue, task)!=C_ERR_OK) {
+        c_Pool_Free(&pool->task_pool, task);
+        return C_FALSE;
+    }
+
+    return C_TRUE;
+}
+
+/**
+ * @brief Orderly terminates the thread pool, waiting for running jobs to finish.
+ */
+void c_ThreadPool_Destroy(c_ThreadPool_t* pool) {
+    if (!pool) return;
+
+    // 1. Terminate the processing loop
+    pool->is_shutdown = C_TRUE;
+
+    // 2. Shut down the queue to unblock workers waiting indefinitely
+    c_LockQueue_Shutdown(&pool->task_queue);
+
+    // 3. Join all worker threads safely
+    for (int i = 0; i < pool->thread_count; i++) {
+        c_Thread_Join(pool->threads[i]);
+    }
+
+    // 4. Drain any remaining unexecuted tasks to prevent memory leaks
+    // void* unexecuted_task = NULL;
+    // while (c_LockQueue_Pop(&pool->task_queue, &unexecuted_task)==C_ERR_OK) {
+    //     C_FREE(unexecuted_task);
+    // }
+    c_Pool_DryUp(&pool->task_pool);
+    c_Pool_Destroy(&pool->task_pool);
+
+    // 5. Reclaim memory structures
+    c_LockQueue_Destroy(&pool->task_queue);
+    C_FREE(pool->threads);
+}
+