多生产者线程通过条件变量通知消费者线程的高效实现是否安全？

问题场景与优化思路 #

我有N个工作线程执行计算任务，另有一个消费者线程等待计算完成。最初采用condition_variable（CV）和初始值为N的原子计数器实现：每个工作线程递减计数器，当计数器归0时向消费者发送CV信号。

但参考常规实现逻辑，修改共享变量需持有mutex，这会导致所有工作线程竞争mutex产生不必要的冲突。因此优化方案为：不借助mutex直接递减原子计数器，仅让最后一个发送信号的工作线程获取mutex。

示例代码 #

constexpr int N_WORKERS = 10;

struct WorkData
{
    int input[N_WORKERS];
    int result[N_WORKERS];
    std::atomic<int> remainingWorkers;
    std::condition_variable cv;
    std::mutex mutex;  
};

void workerFunc(WorkData* wd, int index)
{
    // Do some calculation here
    wd->result[index] = wd->input[index] * wd->input[index];
    //-------------------------
    if (--wd->remainingWorkers == 0)
    {
        wd->mutex.lock();
        wd->mutex.unlock();
        
        wd->cv.notify_one();
    }
}

int main()
{
    WorkData wd;
    wd.remainingWorkers.store(N_WORKERS);

    std::thread workerThreads[N_WORKERS];
    for (int i = 0; i < N_WORKERS; i++)
    {
        wd.input[i] = i;
        wd.result[i] = 0;
        workerThreads[i] = std::thread(workerFunc, &wd, i);
    }
    // Wait for the worker threads to finish unless they already have
    if (wd.remainingWorkers.load() > 0)
    {
        std::unique_lock<std::mutex> lock(wd.mutex);
        while (wd.remainingWorkers.load() > 0)
            wd.cv.wait(lock);
    }
    // Consume result of calculations
    for (int i = 0; i < N_WORKERS; i++)
        std::cout << wd.input[i] << "^2 = " << wd.result[i] << std::endl;
    
    for (std::thread& t : workerThreads)
        t.join();
    return 0;
}

提问 #

请问该实现是否安全，是否存在未发现的竞争条件？

回答 #

这个实现是线程安全的，不存在未发现的竞态条件，核心原因如下：

1. 彻底避免信号丢失 #

消费者线程的等待逻辑是先检查原子计数器，仅当计数器大于0时才上锁进入CV等待。最后一个工作线程的操作逻辑是：先将计数器减至0，再执行一次mutex的加锁解锁，最后发送信号。

• 若消费者还未完成计数器检查、未上锁，工作线程的加锁操作会阻塞，直到消费者完成上锁并进入等待状态，随后解锁并发送信号，确保消费者能被唤醒；
• 若消费者已进入等待状态，工作线程的加锁解锁仅做一次同步操作，后续的信号依然能正常唤醒消费者；
• 若消费者已检查到计数器为0，根本不会进入等待逻辑，自然不存在信号丢失的问题。

2. 原子变量保证内存可见性 #

remainingWorkers是std::atomic类型，所有增减、加载操作都是原子的，且自带内存屏障，确保所有线程对该变量的修改和读取操作是全局同步的，不会出现脏读或指令重排导致的状态不一致。

3. 计算结果无数据竞争 #

每个工作线程仅操作result数组的独立索引，不存在多线程写同一内存地址的竞争；消费者线程仅在所有工作线程完成计算（计数器归0）后才读取结果，此时所有工作线程的写入操作已全部完成，不会出现读未完成写入的情况。

小优化建议 #

可以将工作线程中的手动加锁解锁：

wd->mutex.lock();
wd->mutex.unlock();

替换为RAII风格的锁管理：

std::lock_guard<std::mutex> lock(wd->mutex);

无需手动控制锁的释放，能避免异常场景下的锁泄漏问题。

内容的提问来源于stack exchange，提问作者Morgan