Linux C實現單生產者多消費者環形緩沖區

使用C11里的原子變量實現，沒有用互斥鎖，效率更高。

ring_buffer.h:

/*** @file ring_buffer.h* @author tl* @brief 單生產者多消費者環形緩沖區，每條數據被所有消費者讀后才釋放。讀線程安全，寫僅單線程。* @version* @date 2025-08-06** @copyright Copyright (c) 2025**/#ifndef _RINGBUFFER_H_
#define _RINGBUFFER_H_#ifdef __cplusplus
extern "C"
{
#endif#include <stdint.h>
#include <stddef.h>/*** @brief 環形緩沖區數據包頭結構*/typedef struct __attribute__((packed)){uint32_t length;   ///< 負載數據長度uint16_t type;     ///< 類型uint16_t reserved; ///< 預留擴展} mu_ring_buffer_pkt_hdr;/*** @brief 環形緩沖區句柄*/typedef struct mu_ring_buffer mu_ring_buffer;/*** @brief 環形緩沖區讀者句柄*/typedef struct mu_ring_buffer_reader mu_ring_buffer_reader;/*** @brief 創建環形緩沖區* @param capacity 緩沖區總大小* @return mu_ring_buffer* 成功返回指針，失敗返回NULL*/mu_ring_buffer *mu_ring_buffer_create(size_t capacity);/*** @brief 銷毀環形緩沖區** @param pp_rb 緩沖區指針的地址*/void mu_ring_buffer_destroy(mu_ring_buffer **pp_rb);/*** @brief 添加一個新的讀者** @param rb 環形緩沖區指針* @return mu_ring_buffer_reader* 成功返回讀者指針，失敗返回NULL*/mu_ring_buffer_reader *mu_ring_buffer_add_reader(mu_ring_buffer *rb);/*** @brief 刪除讀者** @param rb 環形緩沖區指針* @param pp_reader 讀者指針的地址* @note 本函數不保證多線程并發安全，調用前請確保該讀者沒有在讀取數據*/void mu_ring_buffer_remove_reader(mu_ring_buffer *rb, mu_ring_buffer_reader **pp_reader);/*** @brief 寫入數據到環形緩沖區** @param rb 環形緩沖區指針* @param hdr 包頭* @param data 負載數據* @return int 成功返回 0，失敗返回負數（-1參數錯誤或內部錯誤，-2空間不足）* @note 僅支持單線程，不支持多線程并發寫入*/int mu_ring_buffer_write(mu_ring_buffer *rb, const mu_ring_buffer_pkt_hdr *hdr, const void *data);/*** @brief 從環形緩沖區讀取數據** @param reader 讀者指針* @param hdr 包頭* @param data 負載數據緩沖區* @param data_size data緩沖區大小* @return int 成功返回 0，失敗返回負數（-1參數或內部錯誤，-2無數據，-3緩沖區空間不足）* @note 線程安全，支持多線程并發讀取*/int mu_ring_buffer_read(mu_ring_buffer_reader *reader, mu_ring_buffer_pkt_hdr *hdr, void *data, size_t data_size);#ifdef __cplusplus
}
#endif#endif // _RINGBUFFER_H_

ring_buffer.c:

#include "ring_buffer.h"
#include <stdlib.h>
#include <string.h>
#include <stdatomic.h>
#include <assert.h>#define PKT_HDR_SIZE (sizeof(mu_ring_buffer_pkt_hdr))
#define MAX_READERS 32struct mu_ring_buffer_reader
{unsigned index;struct mu_ring_buffer *rb;
};struct mu_ring_buffer
{uint8_t *buffer;size_t capacity;                    // 緩沖區容量atomic_size_t tail;                 // 寫指針atomic_size_t *reader_heads;        // 每個讀者的讀指針(head)atomic_uint_fast32_t reader_bitmap; // 讀者分配bitmap
};static inline unsigned find_first_zero_bit(uint32_t v)
{for (unsigned i = 0; i < MAX_READERS; ++i)if (!(v & (1u << i)))return i;return MAX_READERS;
}mu_ring_buffer *mu_ring_buffer_create(size_t capacity)
{if (capacity < PKT_HDR_SIZE + 1)return NULL;mu_ring_buffer *rb = (mu_ring_buffer *)calloc(1, sizeof(mu_ring_buffer));if (!rb)return NULL;rb->buffer = (uint8_t *)malloc(capacity);if (!rb->buffer){free(rb);return NULL;}rb->capacity = capacity;rb->reader_heads = (atomic_size_t *)calloc(MAX_READERS, sizeof(atomic_size_t));if (!rb->reader_heads){free(rb->buffer);free(rb);return NULL;}atomic_store(&rb->tail, 0);atomic_store(&rb->reader_bitmap, 0);// 初始化所有reader_heads為(size_t)-1for (unsigned i = 0; i < MAX_READERS; ++i)atomic_store(&rb->reader_heads[i], (size_t)-1);return rb;
}void mu_ring_buffer_destroy(mu_ring_buffer **pp_rb)
{if (!pp_rb || !*pp_rb)return;mu_ring_buffer *rb = *pp_rb;if (rb->reader_heads)free(rb->reader_heads);if (rb->buffer)free(rb->buffer);free(rb);*pp_rb = NULL;
}mu_ring_buffer_reader *mu_ring_buffer_add_reader(mu_ring_buffer *rb)
{if (!rb)return NULL;uint_fast32_t old, newval;do{old = atomic_load(&rb->reader_bitmap);unsigned idx = find_first_zero_bit(old);if (idx >= MAX_READERS)return NULL;newval = old | (1u << idx);if (atomic_compare_exchange_weak(&rb->reader_bitmap, &old, newval)){mu_ring_buffer_reader *r = (mu_ring_buffer_reader *)malloc(sizeof(mu_ring_buffer_reader));if (!r)return NULL;r->index = idx;r->rb = rb;// 新讀者head指針初始化為tailatomic_store(&rb->reader_heads[idx], atomic_load(&rb->tail));return r;}} while (1);
}void mu_ring_buffer_remove_reader(mu_ring_buffer *rb, mu_ring_buffer_reader **pp_reader)
{if (!rb || !pp_reader || !*pp_reader)return;mu_ring_buffer_reader *reader = *pp_reader;unsigned idx = reader->index;atomic_store(&rb->reader_heads[idx], (size_t)-1); // 標記為無效uint32_t mask = ~(1u << idx);atomic_fetch_and(&rb->reader_bitmap, mask);free(reader);*pp_reader = NULL;
}// 返回所有有效reader head的最小值
static inline size_t min_reader_head(mu_ring_buffer *rb)
{size_t min = atomic_load(&rb->tail);uint32_t bm = atomic_load(&rb->reader_bitmap);int found = 0;for (unsigned i = 0; i < MAX_READERS; ++i){if (bm & (1u << i)){size_t h = atomic_load_explicit(&rb->reader_heads[i], memory_order_acquire);if (h == (size_t)-1)continue; // 已注銷if (!found || ((h + rb->capacity - min) % rb->capacity < (min + rb->capacity - min) % rb->capacity)){min = h;found = 1;}}}return min;
}static inline void ring_buffer_write_data(mu_ring_buffer *rb, size_t offset, const void *src, size_t length)
{if (offset + length <= rb->capacity){memcpy(rb->buffer + offset, src, length);}else{size_t first = rb->capacity - offset;memcpy(rb->buffer + offset, src, first);memcpy(rb->buffer, (const uint8_t *)src + first, length - first);}
}static inline void ring_buffer_read_data(mu_ring_buffer *rb, size_t offset, void *dst, size_t length)
{if (offset + length <= rb->capacity){memcpy(dst, rb->buffer + offset, length);}else{size_t first = rb->capacity - offset;memcpy(dst, rb->buffer + offset, first);memcpy((uint8_t *)dst + first, rb->buffer, length - first);}
}int mu_ring_buffer_write(mu_ring_buffer *rb, const mu_ring_buffer_pkt_hdr *hdr, const void *data)
{if (!rb || !hdr || !data || hdr->length == 0)return -1;size_t total = PKT_HDR_SIZE + hdr->length;if (total > rb->capacity - 1)return -2;size_t tail = atomic_load_explicit(&rb->tail, memory_order_relaxed);size_t min_head = min_reader_head(rb);size_t used = (tail + rb->capacity - min_head) % rb->capacity;size_t free_space = rb->capacity - used - 1;if (free_space < total)return -2;// 寫包頭ring_buffer_write_data(rb, tail, hdr, PKT_HDR_SIZE);// 寫數據size_t data_offset = (tail + PKT_HDR_SIZE) % rb->capacity;ring_buffer_write_data(rb, data_offset, data, hdr->length);// 更新tailatomic_store_explicit(&rb->tail, (tail + total) % rb->capacity, memory_order_release);return 0;
}int mu_ring_buffer_read(mu_ring_buffer_reader *reader, mu_ring_buffer_pkt_hdr *hdr, void *data, size_t data_size)
{if (!reader || !hdr || !data)return -1;mu_ring_buffer *rb = reader->rb;unsigned idx = reader->index;size_t head = atomic_load_explicit(&rb->reader_heads[idx], memory_order_relaxed);size_t tail = atomic_load_explicit(&rb->tail, memory_order_acquire);if (head == (size_t)-1) // 已刪除return -1;if (head == tail) // 沒有新數據return -2;// 讀取包頭ring_buffer_read_data(rb, head, hdr, PKT_HDR_SIZE);if (hdr->length == 0)return -1;// 檢查數據完整性size_t pkt_total = PKT_HDR_SIZE + hdr->length;size_t avail = (tail + rb->capacity - head) % rb->capacity; // 計算可讀數據量if (avail < pkt_total)return -1;if (data_size < hdr->length)return -3;// 讀取數據size_t data_offset = (head + PKT_HDR_SIZE) % rb->capacity;ring_buffer_read_data(rb, data_offset, data, hdr->length);// 更新自己的headatomic_store_explicit(&rb->reader_heads[idx], (head + pkt_total) % rb->capacity, memory_order_release);return 0;
}

單元測試 main.c:

#include "ring_buffer.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <assert.h>
#include <time.h>
#include <errno.h>// 測試統計
static int tests_run = 0;
static int tests_passed = 0;#define TEST_ASSERT(condition, message) \do                                  \{                                   \tests_run++;                    \if (condition)                  \{                               \tests_passed++;             \printf("? %s\n", message);  \}                               \else                            \{                               \printf("? %s\n", message);  \}                               \} while (0)// 測試用例1: 基本創建和銷毀
void test_basic_create_destroy()
{printf("\n=== 測試基本創建和銷毀 ===\n");// 正常創建mu_ring_buffer *rb = mu_ring_buffer_create(1024);TEST_ASSERT(rb != NULL, "正常大小緩沖區創建");// 銷毀mu_ring_buffer_destroy(&rb);TEST_ASSERT(rb == NULL, "緩沖區銷毀后指針為NULL");// 邊界條件測試mu_ring_buffer *rb_small = mu_ring_buffer_create(sizeof(mu_ring_buffer_pkt_hdr));TEST_ASSERT(rb_small == NULL, "過小容量創建失敗");mu_ring_buffer *rb_min = mu_ring_buffer_create(sizeof(mu_ring_buffer_pkt_hdr) + 1);TEST_ASSERT(rb_min != NULL, "最小容量創建成功");mu_ring_buffer_destroy(&rb_min);// 重復銷毀mu_ring_buffer_destroy(&rb);TEST_ASSERT(rb == NULL, "重復銷毀安全");// NULL指針銷毀mu_ring_buffer **null_ptr = NULL;mu_ring_buffer_destroy(null_ptr);TEST_ASSERT(1, "NULL指針銷毀安全");
}// 測試用例2: 讀者管理
void test_reader_management()
{printf("\n=== 測試讀者管理 ===\n");mu_ring_buffer *rb = mu_ring_buffer_create(1024);TEST_ASSERT(rb != NULL, "創建緩沖區");// 添加讀者mu_ring_buffer_reader *reader1 = mu_ring_buffer_add_reader(rb);TEST_ASSERT(reader1 != NULL, "添加第一個讀者");mu_ring_buffer_reader *reader2 = mu_ring_buffer_add_reader(rb);TEST_ASSERT(reader2 != NULL, "添加第二個讀者");// 添加最大數量的讀者mu_ring_buffer_reader *readers[32];int added_count = 2;for (int i = 2; i < 32; i++){readers[i] = mu_ring_buffer_add_reader(rb);if (readers[i] != NULL){added_count++;}}TEST_ASSERT(added_count <= 32, "最多添加32個讀者");// 嘗試添加超出限制的讀者mu_ring_buffer_reader *extra_reader = mu_ring_buffer_add_reader(rb);TEST_ASSERT(extra_reader == NULL, "超出限制的讀者添加失敗");// 刪除讀者mu_ring_buffer_remove_reader(rb, &reader1);TEST_ASSERT(reader1 == NULL, "刪除讀者后指針為NULL");// 刪除后可以重新添加reader1 = mu_ring_buffer_add_reader(rb);TEST_ASSERT(reader1 != NULL, "刪除后重新添加讀者");// 清理mu_ring_buffer_remove_reader(rb, &reader1);mu_ring_buffer_remove_reader(rb, &reader2);for (int i = 2; i < added_count; i++){if (readers[i]){mu_ring_buffer_remove_reader(rb, &readers[i]);}}mu_ring_buffer_destroy(&rb);
}// 測試用例3: 基本讀寫功能
void test_basic_read_write()
{printf("\n=== 測試基本讀寫功能 ===\n");mu_ring_buffer *rb = mu_ring_buffer_create(1024);mu_ring_buffer_reader *reader = mu_ring_buffer_add_reader(rb);// 寫入數據mu_ring_buffer_pkt_hdr write_hdr = {10, 1, 0};char write_data[10] = "hello";int ret = mu_ring_buffer_write(rb, &write_hdr, write_data);TEST_ASSERT(ret == 0, "寫入數據成功");// 讀取數據mu_ring_buffer_pkt_hdr read_hdr;char read_data[20] = {0};ret = mu_ring_buffer_read(reader, &read_hdr, read_data, sizeof(read_data));TEST_ASSERT(ret == 0, "讀取數據成功");TEST_ASSERT(read_hdr.length == write_hdr.length, "讀取包頭長度正確");TEST_ASSERT(read_hdr.type == write_hdr.type, "讀取包頭類型正確");TEST_ASSERT(strcmp(read_data, write_data) == 0, "讀取數據內容正確");// 再次讀取應該無數據ret = mu_ring_buffer_read(reader, &read_hdr, read_data, sizeof(read_data));TEST_ASSERT(ret == -2, "重復讀取返回無數據");mu_ring_buffer_remove_reader(rb, &reader);mu_ring_buffer_destroy(&rb);
}// 測試用例4: 多個讀者讀取
void test_multiple_readers()
{printf("\n=== 測試多個讀者讀取 ===\n");mu_ring_buffer *rb = mu_ring_buffer_create(1024);mu_ring_buffer_reader *reader1 = mu_ring_buffer_add_reader(rb);mu_ring_buffer_reader *reader2 = mu_ring_buffer_add_reader(rb);// 寫入數據mu_ring_buffer_pkt_hdr hdr = {5, 1, 0};char data[5] = "test";int ret = mu_ring_buffer_write(rb, &hdr, data);TEST_ASSERT(ret == 0, "寫入數據成功");// 兩個讀者都能讀到數據mu_ring_buffer_pkt_hdr read_hdr1, read_hdr2;char read_data1[10] = {0}, read_data2[10] = {0};ret = mu_ring_buffer_read(reader1, &read_hdr1, read_data1, sizeof(read_data1));TEST_ASSERT(ret == 0, "讀者1讀取成功");ret = mu_ring_buffer_read(reader2, &read_hdr2, read_data2, sizeof(read_data2));TEST_ASSERT(ret == 0, "讀者2讀取成功");TEST_ASSERT(strcmp(read_data1, read_data2) == 0, "兩個讀者讀取數據一致");mu_ring_buffer_remove_reader(rb, &reader1);mu_ring_buffer_remove_reader(rb, &reader2);mu_ring_buffer_destroy(&rb);
}// 測試用例5: 環形緩沖區繞行
void test_ring_wraparound()
{printf("\n=== 測試環形緩沖區繞行 ===\n");// 創建較小的緩沖區便于測試繞行size_t capacity = 100;mu_ring_buffer *rb = mu_ring_buffer_create(capacity);mu_ring_buffer_reader *reader = mu_ring_buffer_add_reader(rb);// 寫入多個數據包，使其繞行char test_data[20];mu_ring_buffer_pkt_hdr hdr = {20, 1, 0};// 寫入足夠多的數據使其繞行for (int i = 0; i < 5; i++){snprintf(test_data, sizeof(test_data), "packet_%d", i);int ret = mu_ring_buffer_write(rb, &hdr, test_data);if (ret != 0){break; // 緩沖區滿了}}// 讀取所有數據int read_count = 0;mu_ring_buffer_pkt_hdr read_hdr;char read_data[30];while (1){int ret = mu_ring_buffer_read(reader, &read_hdr, read_data, sizeof(read_data));if (ret != 0){break;}read_count++;printf("  讀取到: %s\n", read_data);}TEST_ASSERT(read_count > 0, "成功讀取數據包");mu_ring_buffer_remove_reader(rb, &reader);mu_ring_buffer_destroy(&rb);
}// 測試用例6: 緩沖區滿的處理
void test_buffer_full()
{printf("\n=== 測試緩沖區滿的處理 ===\n");size_t capacity = 64; // 較小的緩沖區mu_ring_buffer *rb = mu_ring_buffer_create(capacity);mu_ring_buffer_reader *reader = mu_ring_buffer_add_reader(rb);// 寫入大量數據直到緩沖區滿mu_ring_buffer_pkt_hdr hdr = {10, 1, 0};char data[10] = "testdata";int write_count = 0;while (1){int ret = mu_ring_buffer_write(rb, &hdr, data);if (ret != 0){TEST_ASSERT(ret == -2, "緩沖區滿時返回-2");break;}write_count++;}printf("  成功寫入 %d 個數據包\n", write_count);TEST_ASSERT(write_count > 0, "至少能寫入一些數據");// 讀取一些數據后應該能繼續寫入mu_ring_buffer_pkt_hdr read_hdr;char read_data[15];int ret = mu_ring_buffer_read(reader, &read_hdr, read_data, sizeof(read_data));TEST_ASSERT(ret == 0, "讀取數據成功");// 現在應該能再寫入一個包ret = mu_ring_buffer_write(rb, &hdr, data);TEST_ASSERT(ret == 0, "讀取后能繼續寫入");mu_ring_buffer_remove_reader(rb, &reader);mu_ring_buffer_destroy(&rb);
}// 測試用例7: 錯誤參數處理
void test_error_handling()
{printf("\n=== 測試錯誤參數處理 ===\n");mu_ring_buffer *rb = mu_ring_buffer_create(1024);mu_ring_buffer_reader *reader = mu_ring_buffer_add_reader(rb);mu_ring_buffer_pkt_hdr hdr = {10, 1, 0};char data[10] = "test";char read_data[10];// 測試NULL參數TEST_ASSERT(mu_ring_buffer_write(NULL, &hdr, data) == -1, "寫入NULL緩沖區返回錯誤");TEST_ASSERT(mu_ring_buffer_write(rb, NULL, data) == -1, "寫入NULL包頭返回錯誤");TEST_ASSERT(mu_ring_buffer_write(rb, &hdr, NULL) == -1, "寫入NULL數據返回錯誤");mu_ring_buffer_pkt_hdr zero_hdr = {0, 1, 0};TEST_ASSERT(mu_ring_buffer_write(rb, &zero_hdr, data) == -1, "寫入零長度數據返回錯誤");// 測試讀取錯誤參數TEST_ASSERT(mu_ring_buffer_read(NULL, &hdr, read_data, sizeof(read_data)) == -1, "NULL讀者返回錯誤");TEST_ASSERT(mu_ring_buffer_read(reader, NULL, read_data, sizeof(read_data)) == -1, "NULL包頭返回錯誤");TEST_ASSERT(mu_ring_buffer_read(reader, &hdr, NULL, sizeof(read_data)) == -1, "NULL數據緩沖區返回錯誤");// 測試數據緩沖區太小mu_ring_buffer_write(rb, &hdr, data);char small_buf[5];TEST_ASSERT(mu_ring_buffer_read(reader, &hdr, small_buf, sizeof(small_buf)) == -3, "數據緩沖區太小返回-3");mu_ring_buffer_remove_reader(rb, &reader);mu_ring_buffer_destroy(&rb);
}// 多線程測試結構
typedef struct
{mu_ring_buffer *rb;int thread_id;int iterations;int *success_count;pthread_mutex_t *mutex;
} thread_data_t;// 寫線程函數
void *writer_thread(void *arg)
{thread_data_t *data = (thread_data_t *)arg;for (int i = 0; i < data->iterations; i++){mu_ring_buffer_pkt_hdr hdr = {16, (uint16_t)data->thread_id, 0};char write_data[16];snprintf(write_data, sizeof(write_data), "T%d_MSG_%d", data->thread_id, i);int ret = mu_ring_buffer_write(data->rb, &hdr, write_data);if (ret == 0){pthread_mutex_lock(data->mutex);(*data->success_count)++;pthread_mutex_unlock(data->mutex);}usleep(1000); // 1ms延遲}return NULL;
}// 讀線程函數
void *reader_thread(void *arg)
{thread_data_t *data = (thread_data_t *)arg;mu_ring_buffer_reader *reader = mu_ring_buffer_add_reader(data->rb);if (!reader){return NULL;}int read_count = 0;time_t start_time = time(NULL);while (time(NULL) - start_time < 5){ // 運行5秒mu_ring_buffer_pkt_hdr hdr;char read_data[32];int ret = mu_ring_buffer_read(reader, &hdr, read_data, sizeof(read_data));if (ret == 0){read_count++;printf("  讀者%d讀取: %s\n", data->thread_id, read_data);}else if (ret == -2){usleep(10000); // 10ms延遲}}pthread_mutex_lock(data->mutex);(*data->success_count) += read_count;pthread_mutex_unlock(data->mutex);mu_ring_buffer_remove_reader(data->rb, &reader);return NULL;
}// 讀者管理線程數據結構
typedef struct
{mu_ring_buffer *rb;int thread_id;volatile int *stop_flag;pthread_mutex_t *mutex;int *operation_count;
} reader_manager_data_t;// 讀者管理線程函數
void *reader_manager_thread(void *arg)
{reader_manager_data_t *data = (reader_manager_data_t *)arg;mu_ring_buffer_reader *managed_readers[5] = {0}; // 最多管理5個讀者int reader_count = 0;int cycle = 0;printf("  讀者管理線程啟動\n");while (!(*data->stop_flag)){cycle++;// 每個周期添加1-3個讀者int readers_to_add = 1 + (cycle % 3);for (int i = 0; i < readers_to_add && reader_count < 5; i++){managed_readers[reader_count] = mu_ring_buffer_add_reader(data->rb);if (managed_readers[reader_count]){reader_count++;printf("    管理線程添加讀者 #%d (總數: %d)\n", reader_count, reader_count);pthread_mutex_lock(data->mutex);(*data->operation_count)++;pthread_mutex_unlock(data->mutex);}}// 讓讀者工作一段時間 (200-500ms)int work_time = 200 + (cycle % 300);usleep(work_time * 1000);// 讀取一些數據for (int i = 0; i < reader_count; i++){if (managed_readers[i]){mu_ring_buffer_pkt_hdr hdr;char read_data[32];// 嘗試讀取幾次for (int j = 0; j < 3; j++){int ret = mu_ring_buffer_read(managed_readers[i], &hdr, read_data, sizeof(read_data));if (ret == 0){printf("    管理的讀者#%d讀取: %s\n", i + 1, read_data);pthread_mutex_lock(data->mutex);(*data->operation_count)++;pthread_mutex_unlock(data->mutex);break;}else if (ret == -2){break; // 無數據}}}}// 刪除一些讀者 (保留至少1個)int readers_to_remove = (reader_count > 2) ? 1 + (cycle % 2) : 0;for (int i = 0; i < readers_to_remove && reader_count > 1; i++){// 從末尾刪除讀者int idx = reader_count - 1;if (managed_readers[idx]){mu_ring_buffer_remove_reader(data->rb, &managed_readers[idx]);managed_readers[idx] = NULL;reader_count--;printf("    管理線程刪除讀者 (剩余: %d)\n", reader_count);pthread_mutex_lock(data->mutex);(*data->operation_count)++;pthread_mutex_unlock(data->mutex);}}// 周期間隔usleep(100000); // 100ms}// 清理剩余的讀者printf("  讀者管理線程清理剩余讀者...\n");for (int i = 0; i < reader_count; i++){if (managed_readers[i]){mu_ring_buffer_remove_reader(data->rb, &managed_readers[i]);printf("    清理讀者 #%d\n", i + 1);}}printf("  讀者管理線程結束\n");return NULL;
}void test_multithreading()
{printf("\n=== 測試多線程并發 ===\n");mu_ring_buffer *rb = mu_ring_buffer_create(8192); // 增大緩沖區pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;volatile int stop_flag = 0;// 創建寫線程數據int write_success = 0;thread_data_t writer_data = {rb, 0, 200, &write_success, &mutex}; // 增加寫入次數// 創建固定讀線程數據int read_success = 0;thread_data_t reader_data1 = {rb, 1, 0, &read_success, &mutex};thread_data_t reader_data2 = {rb, 2, 0, &read_success, &mutex};// 創建讀者管理線程數據int manager_operations = 0;reader_manager_data_t manager_data = {rb, 99, &stop_flag, &mutex, &manager_operations};// 創建線程pthread_t writer_tid, reader_tid1, reader_tid2, manager_tid;printf("  啟動固定讀者線程...\n");pthread_create(&reader_tid1, NULL, reader_thread, &reader_data1);pthread_create(&reader_tid2, NULL, reader_thread, &reader_data2);printf("  啟動讀者管理線程...\n");pthread_create(&manager_tid, NULL, reader_manager_thread, &manager_data);usleep(200000); // 讓讀者和管理線程先啟動printf("  啟動寫入線程...\n");pthread_create(&writer_tid, NULL, writer_thread, &writer_data);// 等待寫線程完成pthread_join(writer_tid, NULL);printf("  寫入線程完成\n");// 讓系統再運行2秒處理剩余數據sleep(2);// 通知停止stop_flag = 1;// 等待所有線程完成pthread_join(reader_tid1, NULL);pthread_join(reader_tid2, NULL);pthread_join(manager_tid, NULL);printf("  所有線程已完成\n");printf("  寫入成功: %d\n", write_success);printf("  固定讀者讀取成功: %d\n", read_success);printf("  讀者管理操作數: %d\n", manager_operations);TEST_ASSERT(write_success > 0, "寫入成功");TEST_ASSERT(read_success > 0, "讀取成功");TEST_ASSERT(manager_operations > 0, "讀者管理操作成功");// 驗證最終狀態printf("  驗證最終緩沖區狀態...\n");pthread_mutex_destroy(&mutex);mu_ring_buffer_destroy(&rb);printf("  多線程并發測試完成\n");
}// 測試用例9: 大數據包測試
void test_large_packets()
{printf("\n=== 測試大數據包 ===\n");size_t capacity = 8192;mu_ring_buffer *rb = mu_ring_buffer_create(capacity);mu_ring_buffer_reader *reader = mu_ring_buffer_add_reader(rb);// 測試接近最大大小的數據包size_t large_size = capacity / 2;char *large_data = malloc(large_size);memset(large_data, 'A', large_size);large_data[large_size - 1] = '\0';mu_ring_buffer_pkt_hdr hdr = {(uint32_t)large_size, 1, 0};int ret = mu_ring_buffer_write(rb, &hdr, large_data);TEST_ASSERT(ret == 0, "寫入大數據包成功");// 讀取大數據包char *read_buffer = malloc(large_size + 100);mu_ring_buffer_pkt_hdr read_hdr;ret = mu_ring_buffer_read(reader, &read_hdr, read_buffer, large_size + 100);TEST_ASSERT(ret == 0, "讀取大數據包成功");TEST_ASSERT(read_hdr.length == hdr.length, "大數據包長度正確");// 測試超大數據包（應該失敗）size_t too_large = capacity;mu_ring_buffer_pkt_hdr huge_hdr = {(uint32_t)too_large, 1, 0};ret = mu_ring_buffer_write(rb, &huge_hdr, large_data);TEST_ASSERT(ret == -2, "超大數據包寫入失敗");free(large_data);free(read_buffer);mu_ring_buffer_remove_reader(rb, &reader);mu_ring_buffer_destroy(&rb);
}// 測試用例10: 性能測試
void test_performance()
{printf("\n=== 性能測試 ===\n");mu_ring_buffer *rb = mu_ring_buffer_create(1024 * 1024); // 1MBmu_ring_buffer_reader *reader = mu_ring_buffer_add_reader(rb);const int iterations = 10000;mu_ring_buffer_pkt_hdr hdr = {64, 1, 0};char data[64];memset(data, 'X', sizeof(data));// 寫入性能測試clock_t start = clock();int write_success = 0;for (int i = 0; i < iterations; i++){if (mu_ring_buffer_write(rb, &hdr, data) == 0){write_success++;}}clock_t write_time = clock() - start;printf("  寫入 %d/%d 個數據包，耗時: %ld ms\n",write_success, iterations, write_time * 1000 / CLOCKS_PER_SEC);// 讀取性能測試start = clock();int read_success = 0;mu_ring_buffer_pkt_hdr read_hdr;char read_data[100];for (int i = 0; i < write_success; i++){if (mu_ring_buffer_read(reader, &read_hdr, read_data, sizeof(read_data)) == 0){read_success++;}}clock_t read_time = clock() - start;printf("  讀取 %d 個數據包，耗時: %ld ms\n",read_success, read_time * 1000 / CLOCKS_PER_SEC);TEST_ASSERT(write_success > iterations * 0.8, "寫入成功率 > 80%");TEST_ASSERT(read_success == write_success, "讀取數量與寫入數量一致");mu_ring_buffer_remove_reader(rb, &reader);mu_ring_buffer_destroy(&rb);
}int main()
{printf("開始環形緩沖區測試...\n");test_basic_create_destroy();test_reader_management();test_basic_read_write();test_multiple_readers();test_ring_wraparound();test_buffer_full();test_error_handling();test_multithreading();test_large_packets();test_performance();printf("\n=== 測試結果 ===\n");printf("總測試數: %d\n", tests_run);printf("通過測試: %d\n", tests_passed);printf("失敗測試: %d\n", tests_run - tests_passed);printf("通過率: %.2f%%\n", (float)tests_passed / tests_run * 100);return (tests_passed == tests_run) ? 0 : 1;
}