From 456543f6a4db06b660015eb760788adf57f23d9f Mon Sep 17 00:00:00 2001 From: Chen Peng Date: Sun, 30 Aug 2026 04:11:22 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=80=E4=BA=9B=E5=9F=BA=E7=A1=80=E7=BB=84?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Foundation/c_LockQueue.c | 325 ++++++++++++++++++++++++++++++++++++ Foundation/c_LockQueue.h | 54 ++++++ Foundation/c_Thread.c | 1 + Foundation/c_Thread.h | 125 ++++++++++++++ Foundation/c_ThreadPool.c | 122 ++++++++++++++ Foundation/c_ThreadPool.h | 44 +++++ Foundation/c_ThreadPool.t.c | 39 +++++ 7 files changed, 710 insertions(+) create mode 100644 Foundation/c_LockQueue.c create mode 100644 Foundation/c_LockQueue.h create mode 100644 Foundation/c_Thread.c create mode 100644 Foundation/c_Thread.h create mode 100644 Foundation/c_ThreadPool.c create mode 100644 Foundation/c_ThreadPool.h create mode 100644 Foundation/c_ThreadPool.t.c diff --git a/Foundation/c_LockQueue.c b/Foundation/c_LockQueue.c new file mode 100644 index 0000000..e042a4b --- /dev/null +++ b/Foundation/c_LockQueue.c @@ -0,0 +1,325 @@ +#include + +#if defined(PLATFORM_POSIX) +#include +#endif + +/* ------------------------------------------------------------------------------------------------------------------ */ +/* */ + +c_err_t c_LockQueue_Init(c_LockQueue_t* queue, c_size_t capacity, c_Allocator_t* allocator) { + if (!queue || capacity==0) return C_ERR_PARAM; + queue->allocator = (allocator != NULL) ? *allocator : c_DefaultAllocator; + queue->data = (void**)c_Allocator_Alloc(&queue->allocator, sizeof(void*) * capacity); + if (!queue->data) { + return C_ERR_NOMEM; + } + queue->capacity = capacity; + queue->write_idx = 0; + queue->read_idx = 0; + queue->size = 0; + queue->is_shutdown = C_FALSE; + + if ((c_Mutex_Init(&queue->lock)!=C_ERR_OK) || + (c_Cond_Init(&queue->not_full)!=C_ERR_OK) || + (c_Cond_Init(&queue->not_empty)!=C_ERR_OK)) + { + if (queue->data) { + c_Allocator_Free(&queue->allocator, queue->data); + queue->data = NULL; + } + } + return C_ERR_OK; +} +void c_LockQueue_Destroy(c_LockQueue_t* queue) { + if (!queue) return; + + c_LockQueue_Shutdown(queue); + + c_Mutex_Lock(&queue->lock); + c_Cond_Destroy(&queue->not_full); + c_Cond_Destroy(&queue->not_empty); + if (queue->data) { + c_Allocator_Free(&queue->allocator, queue->data); + queue->data = NULL; + } + c_Mutex_UnLock(&queue->lock); + + // 销毁跨平台锁 + c_Mutex_Destroy(&queue->lock); +} + +void c_LockQueue_Shutdown(c_LockQueue_t* queue) { + if (!queue) return; + + c_Mutex_Lock(&queue->lock); + queue->is_shutdown = C_TRUE; + + // 广播唤醒所有正在阻塞的生产者和消费者，让他们通过 is_shutdown 状态感知并安全退出 + c_Cond_Broadcast(&queue->not_full); + c_Cond_Broadcast(&queue->not_empty); + c_Mutex_UnLock(&queue->lock); +} + +c_err_t c_LockQueue_Push(c_LockQueue_t* queue, void* data) { + if (!queue) return C_ERR_PARAM; + + c_Mutex_Lock(&queue->lock); + + // 1. 经典工业级设计：使用 while 循环检查条件，完美防御虚假唤醒 + while (queue->size == queue->capacity && !queue->is_shutdown) { + c_Cond_Wait(&queue->not_full, &queue->lock); + } + + // 2. 如果队列中途被关闭，直接拒绝写入并返回 + if (queue->is_shutdown) { + c_Mutex_UnLock(&queue->lock); + return C_ERR_FAIL; + } + + // 3. 循环数组插入数据 + queue->data[queue->write_idx] = data; + queue->write_idx = (queue->write_idx + 1) % queue->capacity; + queue->size++; + + // 4. 唤醒可能正在等待数据的消费者 + c_Cond_Signal(&queue->not_empty); + + c_Mutex_UnLock(&queue->lock); + return C_ERR_OK; +} + +c_err_t c_LockQueue_Pop(c_LockQueue_t* queue, void** item) { + if (!queue ) return C_ERR_PARAM; + + c_Mutex_Lock(&queue->lock); + + // 1. 队空且未关闭时，消费者阻塞等待 + while (queue->size == 0 && !queue->is_shutdown) { + c_Cond_Wait(&queue->not_empty, &queue->lock); + } + + // 2. 如果队列已关闭且数据已被清空，优雅退出 + if (queue->is_shutdown && queue->size == 0) { + c_Mutex_UnLock(&queue->lock); + return C_ERR_FAIL; + } + + // 3. 循环数组取出数据 + if (item) { + *item = queue->data[queue->read_idx]; + } + queue->read_idx = (queue->read_idx + 1) % queue->capacity; + queue->size--; + + // 4. 唤醒可能正在等待空间的生产者 + c_Cond_Signal(&queue->not_full); + + c_Mutex_UnLock(&queue->lock); + return C_ERR_OK; +} + +c_size_t c_LockQueue_Size(c_LockQueue_t* queue) { + if (!queue) return 0; + c_Mutex_Lock(&queue->lock); + const c_size_t size = queue->size; + c_Mutex_UnLock(&queue->lock); + return size; +} + +c_bool_t c_LockQueue_IsEmpty(c_LockQueue_t* queue) { + if (!queue) return true; // 安全檢查：無效隊列視為空 + + c_Mutex_Lock(&queue->lock); + const c_bool_t is_empty = (queue->size == 0); + c_Mutex_UnLock(&queue->lock); + + return is_empty; +} + +c_bool_t c_LockQueue_IsFull(c_LockQueue_t* queue) { + if (!queue) return false; // 安全檢查 + + c_Mutex_Lock(&queue->lock); + const c_bool_t is_full = (queue->size == queue->capacity); + c_Mutex_UnLock(&queue->lock); + return is_full; +} + + +c_bool_t c_LockQueue_TimedPop(c_LockQueue_t* queue, void** item, c_uint_t timeout_ms) { + if (!queue || !item) return C_FALSE; + + c_Mutex_Lock(&queue->lock); + + // 1. Calculate the absolute deadline for POSIX or track elapsed time for Windows + c_uint_t remaining_ms = timeout_ms; + +#if defined(PLATFORM_POSIX) + // POSIX timedwait requires an absolute system calendar time deadline + struct timespec deadline; + struct timeval now; + gettimeofday(&now, NULL); + long long total_ns = (long long)now.tv_usec * 1000 + (long long)timeout_ms * 1000000; + deadline.tv_sec = now.tv_sec + total_ns / 1000000000LL; + deadline.tv_nsec = total_ns % 1000000000LL; +#elif defined(PLATFORM_WINDOWS) + // Windows tracks relative intervals natively via GetTickCount/GetTickCount64 + ULONGLONG start_tick = GetTickCount64(); +#endif + + // 2. Loop to defend against Spurious Wakeups + while (queue->size == 0 && !queue->is_shutdown) { + if (remaining_ms == 0) { + // Out of time before cond wait or remaining time became zero + c_Mutex_UnLock(&queue->lock); + return C_FALSE; + } + + // 3. Atomically release the lock and sleep until signaled or timed out +#if defined(PLATFORM_WINDOWS) + // SleepConditionVariableCS handles relative timeout natively + BOOL wait_success = SleepConditionVariableCS(&queue->not_empty.handle, &queue->lock.handle, remaining_ms); + + if (!wait_success) { + if (GetLastError() == ERROR_TIMEOUT) { + c_Mutex_UnLock(&queue->lock); + return C_FALSE; // Dynamic Windows timeout hit + } + } + + // Recalculate remaining time in case of spurious wakeups + ULONGLONG elapsed = GetTickCount64() - start_tick; + if (elapsed >= timeout_ms) { + remaining_ms = 0; + } else { + remaining_ms = timeout_ms - (c_uint_t)elapsed; + } +#elif defined(PLATFORM_POSIX) + // pthread_cond_timedwait takes the exact calculated deadline + int wait_result = pthread_cond_timedwait(&queue->not_empty.handle, &queue->lock.handle, &deadline); + + if (wait_result != 0) { + // POSIX returns ETIMEDOUT (usually 110) if time limit expires + c_Mutex_UnLock(&queue->lock); + return C_FALSE; + } + + // Re-verify remaining time using current clock just to be precise + gettimeofday(&now, NULL); + long long current_ms = (long long)now.tv_sec * 1000 + now.tv_usec / 1000; + long long deadline_ms = (long long)deadline.tv_sec * 1000 + deadline.tv_nsec / 1000000; + if (current_ms >= deadline_ms) { + remaining_ms = 0; + } else { + remaining_ms = (c_uint_t)(deadline_ms - current_ms); + } +#endif + } + + // 4. Handle exit criteria if queue shut down during wait + if (queue->is_shutdown && queue->size == 0) { + c_Mutex_UnLock(&queue->lock); + return C_FALSE; + } + + // 5. Securely pop data from circular buffer + *item = queue->data[queue->read_idx]; + queue->read_idx = (queue->read_idx + 1) % queue->capacity; + queue->size--; + + // 6. Signal blocked producers that space has cleared up + c_Cond_Signal(&queue->not_full); + + c_Mutex_UnLock(&queue->lock); + return C_TRUE; +} + +c_bool_t c_LockQueue_TimedPush(c_LockQueue_t* queue, void* item, c_uint_t timeout_ms) { + if (!queue || !item) return C_FALSE; + + c_Mutex_Lock(&queue->lock); + + // 1. Calculate the absolute deadline for POSIX or track elapsed time for Windows + c_uint_t remaining_ms = timeout_ms; + +#if defined(PLATFORM_POSIX) + // POSIX timedwait requires an absolute wall-clock calendar deadline + struct timespec deadline; + struct timeval now; + gettimeofday(&now, NULL); + long long total_ns = (long long)now.tv_usec * 1000 + (long long)timeout_ms * 1000000; + deadline.tv_sec = now.tv_sec + total_ns / 1000000000LL; + deadline.tv_nsec = total_ns % 1000000000LL; +#elif defined(PLATFORM_WINDOWS) + // Windows tracks relative intervals natively via clock ticks + ULONGLONG start_tick = GetTickCount64(); +#endif + + // 2. Loop to defend against Spurious Wakeups while the queue is full + while (queue->size == queue->capacity && !queue->is_shutdown) { + if (remaining_ms == 0) { + // Out of time before cond wait or remaining time ticked down to zero + c_Mutex_UnLock(&queue->lock); + return C_FALSE; + } + + // 3. Atomically release the lock and sleep until signaled or timed out +#if defined(PLATFORM_WINDOWS) + // SleepConditionVariableCS handles relative timeout natively + BOOL wait_success = SleepConditionVariableCS(&queue->not_full.handle, &queue->lock.handle, remaining_ms); + + if (!wait_success) { + if (GetLastError() == ERROR_TIMEOUT) { + c_Mutex_UnLock(&queue->lock); + return C_FALSE; // Dynamic Windows timeout expired + } + } + + // Recalculate remaining time in case of a spurious wakeup + ULONGLONG elapsed = GetTickCount64() - start_tick; + if (elapsed >= timeout_ms) { + remaining_ms = 0; + } else { + remaining_ms = timeout_ms - (c_uint_t)elapsed; + } +#elif defined(PLATFORM_POSIX) + // pthread_cond_timedwait takes the absolute calculated deadline + int wait_result = pthread_cond_timedwait(&queue->not_full.handle, &queue->lock.handle, &deadline); + + if (wait_result != 0) { + // POSIX returns ETIMEDOUT (110) if time limit expires + c_Mutex_UnLock(&queue->lock); + return C_FALSE; + } + + // Re-verify remaining time using current clock to stay precise + gettimeofday(&now, NULL); + long long current_ms = (long long)now.tv_sec * 1000 + now.tv_usec / 1000; + long long deadline_ms = (long long)deadline.tv_sec * 1000 + deadline.tv_nsec / 1000000; + if (current_ms >= deadline_ms) { + remaining_ms = 0; + } else { + remaining_ms = (c_uint_t)(deadline_ms - current_ms); + } +#endif + } + + // 4. Handle exit criteria if queue shut down during wait + if (queue->is_shutdown) { + c_Mutex_UnLock(&queue->lock); + return C_FALSE; + } + + // 5. Securely push data into the circular buffer + queue->data[queue->write_idx] = item; + queue->write_idx = (queue->write_idx + 1) % queue->capacity; + queue->size++; + + // 6. Signal blocked consumers that data is ready + c_Cond_Signal(&queue->not_empty); + + c_Mutex_UnLock(&queue->lock); + return C_TRUE; +} + diff --git a/Foundation/c_LockQueue.h b/Foundation/c_LockQueue.h new file mode 100644 index 0000000..b88cff5 --- /dev/null +++ b/Foundation/c_LockQueue.h @@ -0,0 +1,54 @@ +#ifndef INCLUDED_C_LOCKQUEUE_H +#define INCLUDED_C_LOCKQUEUE_H + +#ifndef INCLUDED_C_TYPES_H +#include +#endif /*INCLUDED_C_TYPES_H*/ + +#ifndef INCLUDED_C_MUTEX_H +#include +#endif /*INCLUDED_C_MUTEX_H*/ + +#ifndef INCLUDED_C_COND_H +#include +#endif /*INCLUDED_C_COND_H*/ + +#ifndef INCLUDED_C_ALLOCATOR_H +#include +#endif /*INCLUDED_C_ALLOCATOR_H*/ + + +/* ------------------------------------------------------------------------------------------------------------------ */ +/* */ + +typedef struct { + void** data; + c_size_t capacity; + c_size_t write_idx; + c_size_t read_idx; + c_size_t size; + c_bool_t is_shutdown; + + c_Mutex_t lock; + c_Cond_t not_full; + c_Cond_t not_empty; + c_Allocator_t allocator; +}c_LockQueue_t; + +/* ------------------------------------------------------------------------------------------------------------------ */ +/* */ + +c_err_t c_LockQueue_Init(c_LockQueue_t* queue, c_size_t capacity, c_Allocator_t* allocator); +void c_LockQueue_Destroy(c_LockQueue_t* queue); +void c_LockQueue_Shutdown(c_LockQueue_t* queue); + +c_err_t c_LockQueue_Push(c_LockQueue_t* queue, void* data); +c_err_t c_LockQueue_Pop(c_LockQueue_t* queue, void** data); +c_bool_t c_LockQueue_TimedPop(c_LockQueue_t* queue, void** item, c_uint_t timeout_ms); +c_bool_t c_LockQueue_TimedPush(c_LockQueue_t* queue, void* item, c_uint_t timeout_ms); +c_size_t c_LockQueue_Size(c_LockQueue_t* queue); + +c_bool_t c_LockQueue_IsEmpty(c_LockQueue_t* queue); +c_bool_t c_LockQueue_IsFull(c_LockQueue_t* queue); + +#endif /*INCLUDED_C_LOCKQUEUE_H*/ diff --git a/Foundation/c_Thread.c b/Foundation/c_Thread.c new file mode 100644 index 0000000..3884e29 --- /dev/null +++ b/Foundation/c_Thread.c @@ -0,0 +1 @@ +#include diff --git a/Foundation/c_Thread.h b/Foundation/c_Thread.h new file mode 100644 index 0000000..bebb118 --- /dev/null +++ b/Foundation/c_Thread.h @@ -0,0 +1,125 @@ +#ifndef INCLUDED_C_THREAD_H +#define INCLUDED_C_THREAD_H + +#ifndef INCLUDED_C_TYPES_H +#include +#endif /*INCLUDED_C_TYPES_H*/ + +/* ------------------------------------------------------------------------------------------------------------------ */ +/* */ + +typedef void* c_ThreadResult_t; + +#if defined(_WIN32) || defined(_WIN64) + #include + #include + typedef HANDLE c_Thread_t; + typedef DWORD c_ThreadId_t; // Windows 使用 DWORD 作为线程 ID + #define C_THREAD_FUNC_RETURN_TYPE unsigned __stdcall + #define C_THREAD_FUNC_RETURN_VTYPE unsigned +#else + #include + typedef pthread_t c_Thread_t; + typedef unsigned long c_ThreadId_t; // POSIX 转换为数字 ID + #define C_THREAD_FUNC_RETURN_TYPE void* + #define C_THREAD_FUNC_RETURN_VTYPE C_THREAD_FUNC_RETURN_TYPE +#endif + +// Universal thread creation signature +typedef C_THREAD_FUNC_RETURN_TYPE (*c_ThreadFn_t)(void*); + +C_STATIC_FORCE_INLINE +c_bool_t c_Thread_Create(c_Thread_t* thread, c_ThreadFn_t func, void* arg) { +#if defined(_WIN32) || defined(_WIN64) + // *thread = CreateThread(NULL, 0, func, arg, 0, NULL); + // return (*thread != NULL); + *thread = (HANDLE)_beginthreadex(NULL, 0, (unsigned (__stdcall *)(void*))func, arg, 0, NULL); + return (*thread != NULL); +#else + return (pthread_create(thread, NULL, func, arg) == 0); +#endif +} + +C_STATIC_FORCE_INLINE +void c_Thread_Join(c_Thread_t thread) { +#if defined(_WIN32) || defined(_WIN64) + WaitForSingleObject(thread, INFINITE); + CloseHandle(thread); +#else + pthread_join(thread, NULL); +#endif +} + +C_STATIC_FORCE_INLINE +void c_Thread_Sleep(unsigned int milliseconds) { + #if defined(_WIN32) || defined(_WIN64) + Sleep(milliseconds); + #else + usleep(milliseconds * 1000); + #endif +} + +C_STATIC_FORCE_INLINE +c_ThreadId_t c_Thread_SelfId(void) { +#if defined(_WIN32) || defined(_WIN64) + return GetCurrentThreadId(); +#else + return (c_ThreadId_t)pthread_self(); +#endif +} + +C_STATIC_FORCE_INLINE +c_bool_t c_Thread_Detach(c_Thread_t thread) { +#if defined(_WIN32) || defined(_WIN64) + // Windows 中，关闭线程句柄并不等于终止线程。 + // 它只是减少内核对象的引用计数。线程运行结束时，内核对象会自动销毁。 + // 这与 POSIX 的 detach 行为完全一致。 + if (thread != NULL) { + return (CloseHandle(thread) != 0); + } + return C_FALSE; +#else + // POSIX 直接使用 pthread_detach + return (pthread_detach(thread) == 0); +#endif +} + +C_STATIC_FORCE_INLINE +c_bool_t c_Thread_Equal(c_Thread_t t1, c_Thread_t t2) { +#if defined(_WIN32) || defined(_WIN64) + // Windows 下可以通过比较线程 ID 来判断是否为同一个线程 + // 即使其中一个是 GetCurrentThread() 产生的伪句柄，GetThreadId 也能正确识别 + return (GetThreadId(t1) == GetThreadId(t2)); +#else + // POSIX 提供专用的比较函数 + return (pthread_equal(t1, t2) != 0); +#endif +} + +C_STATIC_FORCE_INLINE +c_bool_t c_Thread_JoinWithResult(c_Thread_t thread, c_ThreadResult_t* out_result) { +#if defined(_WIN32) || defined(_WIN64) + if (WaitForSingleObject(thread, INFINITE) == WAIT_OBJECT_0) { + DWORD exit_code = 0; + if (GetExitCodeThread(thread, &exit_code)) { + if (out_result) { + // 将 DWORD 转换为指针类型输出 + *out_result = (c_ThreadResult_t)(uintptr_t)exit_code; + } + CloseHandle(thread); + return C_TRUE; + } + } + CloseHandle(thread); // 即使失败也尝试关闭句柄 + return C_FALSE; +#else + // POSIX 直接在 join 时传入指针变量的地址 + return (pthread_join(thread, out_result) == 0); +#endif +} + +#define c_Thread_Exit(x) return (C_THREAD_FUNC_RETURN_VTYPE)(x) + +#define c_Thread_Fn(fn) C_THREAD_FUNC_RETURN_TYPE fn + +#endif /*INCLUDED_C_THREAD_H*/ diff --git a/Foundation/c_ThreadPool.c b/Foundation/c_ThreadPool.c new file mode 100644 index 0000000..b2e70f7 --- /dev/null +++ b/Foundation/c_ThreadPool.c @@ -0,0 +1,122 @@ +#include +#include + +// 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); +} + diff --git a/Foundation/c_ThreadPool.h b/Foundation/c_ThreadPool.h new file mode 100644 index 0000000..248ce5c --- /dev/null +++ b/Foundation/c_ThreadPool.h @@ -0,0 +1,44 @@ +#ifndef INCLUDED_C_THREADPOOL_H +#define INCLUDED_C_THREADPOOL_H + +#ifndef INCLUDED_C_LOCKQUEUE_H +#include +#endif /*INCLUDED_C_LOCKQUEUE_H*/ + +#ifndef INCLUDED_C_THREAD_H +#include +#endif /*INCLUDED_C_THREAD_H*/ + +#ifndef INCLUDED_C_POOL_H +#include +#endif /*INCLUDED_C_POOL_H*/ + + + +/* ------------------------------------------------------------------------------------------------------------------ */ +/* */ + +// Struct representing a single executable unit of work +typedef struct { + void (*function)(void*); // Pointer to the user function + void* argument; // Argument passed to the function +} c_ThreadPoolTask_t; + +typedef struct { + c_LockQueue_t task_queue; // Safe queue storing thread_pool_task_t pointers + c_Thread_t* threads; // Array of worker thread handles + int thread_count; // Total number of worker threads + volatile c_bool_t is_shutdown; // Shutdown flag + int check_interval_ms; + c_Pool_t task_pool; +} c_ThreadPool_t; + +// Public Core API +c_err_t c_ThreadPool_Init(c_ThreadPool_t* pool, int thread_count, int queue_capacity, int task_pool_size, int check_interval_ms); + +void c_ThreadPool_Destroy(c_ThreadPool_t* pool); + +c_bool_t c_ThreadPool_Submit(c_ThreadPool_t* pool, void (*function)(void*), void* argument); + + +#endif /*INCLUDED_C_THREADPOOL_H*/ diff --git a/Foundation/c_ThreadPool.t.c b/Foundation/c_ThreadPool.t.c new file mode 100644 index 0000000..be18b81 --- /dev/null +++ b/Foundation/c_ThreadPool.t.c @@ -0,0 +1,39 @@ +#include "c_ThreadPool.h" +#include +#include + +// Sample payload mimicking work +void compute_square(void* arg) { + int val = *(int*)arg; + printf("[Thread Pool Task] Processing square of %d = %d\n", val, val * val); + free(arg); // Free parameter memory passed during submission +} + + +int main(int argc, char** argv){ + // Create a pool with 4 worker threads and a max capacity of 20 pending tasks + c_ThreadPool_t pool = {0}; + c_ThreadPool_Init(&pool, 4, 20, 10, 500); + + printf("--- Thread Pool Initialized with 4 Workers ---\n"); + + // Queue 10 dynamic processing computations + for (int i = 1; i <= 10; i++) { + int* num = (int*)malloc(sizeof(int)); + *num = i; + if (!c_ThreadPool_Submit(&pool, compute_square, num)) { + fprintf(stderr, "[Thread Pool Task] Submit %d failed\n", i); + } + } + + // Force main to simulate work before cleaning up + printf("All tasks submitted. Waiting for processing to settle...\n"); + + c_Thread_Sleep(2000); + + printf("--- Destroying Thread Pool ---\n"); + c_ThreadPool_Destroy(&pool); + printf("Thread pool destroyed cleanly.\n"); + + return 0; +}