326 lines
11 KiB
C
326 lines
11 KiB
C
#include <c_LockQueue.h>
|
|
|
|
#if defined(PLATFORM_POSIX)
|
|
#include <sys/time.h>
|
|
#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;
|
|
}
|
|
|