18、测试多线程代码：使用CUnit测试线程安全函数、处理竞态条件

多线程测试，说实话，是单元测试里最让人头疼的一块。我早年做嵌入式项目时，就吃过竞态条件的大亏——一个全局变量被两个线程同时读写，bug 时隐时现，折腾了两周才定位到。从那以后，我养成了一个习惯：凡是涉及多线程的代码，必须用测试工具把竞态条件“逼”出来。

这一章，我们就聊聊怎么用 CUnit 来测试线程安全函数，以及如何设计用例来暴露竞态条件。

18.1 线程安全函数长什么样？

先明确一个概念：线程安全函数，指的是多个线程同时调用它，不会产生数据竞争或逻辑错误。常见的实现手段有：

• 加锁保护：用互斥锁（mutex）保护共享资源
• 无锁设计：使用原子操作或线程局部存储
• 不可变数据：函数内部不修改全局状态

举个例子，一个简单的计数器：

// counter.h
typedef struct {
    int value;
    pthread_mutex_t lock;
} SafeCounter;

void counter_init(SafeCounter *c);
void counter_increment(SafeCounter *c);
int counter_get(SafeCounter *c);

// counter.c
void counter_init(SafeCounter *c) {
    c->value = 0;
    pthread_mutex_init(&c->lock, NULL);
}

void counter_increment(SafeCounter *c) {
    pthread_mutex_lock(&c->lock);
    c->value++;
    pthread_mutex_unlock(&c->lock);
}

int counter_get(SafeCounter *c) {
    pthread_mutex_lock(&c->lock);
    int val = c->value;
    pthread_mutex_unlock(&c->lock);
    return val;
}

这个实现加了锁，理论上线程安全。但理论归理论，我们得用测试来验证。

18.2 用 CUnit 测试线程安全函数

CUnit 本身不直接提供多线程测试框架，但我们可以利用 POSIX 线程（pthread）来创建多个线程，在测试函数中并发调用目标函数。

核心思路：

• 启动 N 个线程，每个线程循环调用目标函数 M 次
• 主线程等待所有线程结束
• 检查最终结果是否符合预期（比如计数器值是否等于 N * M）

来看一个完整的测试用例：

#include <CUnit/CUnit.h>
#include <pthread.h>
#include "counter.h"

#define THREAD_NUM 4
#define ITER_PER_THREAD 10000

SafeCounter g_counter;

typedef struct {
    int thread_id;
    int result;
} ThreadArg;

void* thread_func(void *arg) {
    ThreadArg *targ = (ThreadArg*)arg;
    for (int i = 0; i < ITER_PER_THREAD; i++) {
        counter_increment(&g_counter);
    }
    targ->result = counter_get(&g_counter);
    return NULL;
}

void test_counter_thread_safe(void) {
    counter_init(&g_counter);

    pthread_t threads[THREAD_NUM];
    ThreadArg args[THREAD_NUM];

    // 创建线程
    for (int i = 0; i < THREAD_NUM; i++) {
        args[i].thread_id = i;
        pthread_create(&threads[i], NULL, thread_func, &args[i]);
    }

    // 等待所有线程结束
    for (int i = 0; i < THREAD_NUM; i++) {
        pthread_join(threads[i], NULL);
    }

    // 验证最终结果
    int final_val = counter_get(&g_counter);
    CU_ASSERT_EQUAL(final_val, THREAD_NUM * ITER_PER_THREAD);
}

18.3 处理竞态条件：如何设计“破坏性”测试

竞态条件有个特点：它不一定每次都能复现。你跑 10 次可能只出现 1 次。所以测试设计要刻意制造“碰撞”。

我个人习惯用以下几种手法：

1. 高并发 + 短操作：让线程密集执行短小操作，增加锁竞争概率
2. 线程数多于 CPU 核心数：迫使线程频繁切换，暴露时序问题
3. 随机延迟：在线程内部插入 usleep(rand()%10)，打乱执行顺序
4. 重复运行：把测试用例放在循环里跑 100 次，直到出现失败

举个例子，给上面的测试加上随机延迟：

void* thread_func_with_delay(void *arg) {
    ThreadArg *targ = (ThreadArg*)arg;
    for (int i = 0; i < ITER_PER_THREAD; i++) {
        counter_increment(&g_counter);
        // 随机延迟 0~5 微秒，增加竞态暴露概率
        usleep(rand() % 5);
    }
    targ->result = counter_get(&g_counter);
    return NULL;
}

18.4 测试无锁实现：一个反面教材

有些同学喜欢用无锁编程来提升性能。但无锁代码的测试更难。来看一个“看似安全”的无锁计数器：

// 错误示例：无锁但非原子操作
void unsafe_increment(int *counter) {
    int temp = *counter;  // 读
    temp++;               // 改
    *counter = temp;      // 写
    // 这三步不是原子的，线程切换时数据会丢失
}

用我们上面的多线程测试去跑这个函数，几乎每次都会失败。但如果你只跑 2 个线程、每个线程 100 次，可能偶尔能通过。这就是竞态条件的狡猾之处。

正确的无锁实现应该使用原子操作（C11 的 atomic_fetch_add 或 GCC 的 __sync_fetch_and_add）：

#include <stdatomic.h>

atomic_int g_atomic_counter = 0;

void atomic_increment() {
    atomic_fetch_add(&g_atomic_counter, 1);
}

用 CUnit 测试原子操作时，同样可以用多线程并发调用的方式验证。

18.5 测试中的常见陷阱

嗯，这里要提醒几个容易踩的坑：

陷阱	说明	建议
测试本身不是线程安全的	多个线程同时调用 CU_ASSERT 宏，可能导致断言结果混乱	只在主线程中做断言检查，工作线程只负责执行操作
忘记初始化互斥锁	静态分配的 mutex 如果没有初始化，行为未定义	统一用 pthread_mutex_init 初始化
测试时间过长	高并发测试可能跑几秒甚至几分钟	设置超时机制，或者把压力测试单独放在一个 suite 里
忽略线程创建失败	pthread_create 可能返回错误	检查返回值，失败时跳过测试或报错

18.6 知识体系图

下面这张图总结了多线程测试的核心流程和关键点：

18.7 总结

测试多线程代码，说白了就是和“不确定性”做斗争。你没法保证一次测试就能覆盖所有时序，但可以通过精心设计的并发场景，把竞态条件逼到墙角。

我个人总结了三句话：

• 加锁的代码不一定安全——锁的粒度、顺序、遗漏都可能引入 bug
• 无锁的代码更难测——原子操作虽然快，但心智负担大
• 测试本身也要线程安全——断言留在主线程，工作线程只管干活

最后送大家一句话：多线程测试不是一次性的工作，而是持续的过程。每次修改共享资源的访问逻辑，都应该跑一遍并发测试。别问我怎么知道的——我曾经因为少跑一次测试，在线上环境排查了三天。