123 lines
3.8 KiB
C
123 lines
3.8 KiB
C
#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_SUCCESS) {
|
|
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_SUCCESS) {
|
|
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);
|
|
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_SUCCESS;
|
|
}
|
|
|
|
/**
|
|
* @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_SUCCESS) {
|
|
C_FREE(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_SUCCESS) {
|
|
// 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);
|
|
}
|
|
|