Boost线程循环同步方案咨询：并行任务与后续任务同步问题

我有多个对象，它们都具备相同的func方法，需要并行调用所有对象的func；等所有func执行完成后，再调用另一个对象的anotherFunc方法，该逻辑需循环执行。核心约束：

• 所有执行func的线程，必须等待anotherFunc完成后才能进入下一轮
• 执行anotherFunc的线程，必须等待所有func执行完成后才能运行

当前代码实现 #

for(int i = 0; i < n; ++i)
{  
    // 这些线程需要等待anotherFunc完成后再进入下一轮，线程间无数据交互
    boost::asio::post(pool, boost::bind(&obj::func, &instance1));
    boost::asio::post(pool, boost::bind(&obj::func, &instance2));
    ...
    boost::asio::post(pool, boost::bind(&obj::func, &instancek));

    // 这个任务需要等待上面所有func执行完成后，再调用对象方法更新成员变量
    boost::asio::post(pool, boost::bind(&anotherObj::anotherFunc, &anotherInstance));
}

疑问与尝试 #

我想明确哪种同步机制更适用：

• 使用条件变量是否过于繁琐？如何实现多条件等待？
• 考虑用barrier让所有线程等待最后一个完成，但要确保anotherFunc在所有func完成后才执行

我曾尝试在anotherFunc末尾添加cond.notify_all()，并在func中加入以下代码：

boost::mutex::scoped_lock lock(mutex);      
cond.wait(lock);
lock.unlock(); // 原代码中unock为笔误，应为lock

但运行数百次迭代后出现死锁，寻求成熟的解决方案。

成熟解决方案 #

方案1：使用boost::barrier分阶段同步 #

boost::barrier是专门用于一组线程互相等待到达同步点的工具，完美适配你的两轮同步需求：

1. 第一阶段：所有func线程等待彼此完成，允许anotherFunc启动
2. 第二阶段：所有线程（含anotherFunc）等待彼此完成，再进入下一轮循环

具体实现

// 全局或类成员变量，初始化时设置对应计数
boost::barrier func_barrier(k); // k为func对象的数量
boost::barrier round_barrier(k + 1); // k个func线程 + 1个anotherFunc线程

// 修改后的obj::func逻辑
void obj::func() {
    // 执行原有业务逻辑
    do_your_business();

    // 等待所有func线程完成
    func_barrier.wait();

    // 等待anotherFunc执行完毕，开启下一轮
    round_barrier.wait();
}

// 修改后的anotherObj::anotherFunc逻辑
void anotherObj::anotherFunc() {
    // 等待所有func线程完成
    func_barrier.wait();

    // 执行原有业务逻辑（更新成员变量等）
    update_member_values();

    // 等待所有线程到达同步点，触发下一轮
    round_barrier.wait();
}

// 循环提交任务的代码保持不变
for(int i = 0; i < n; ++i)
{  
    boost::asio::post(pool, boost::bind(&obj::func, &instance1));
    boost::asio::post(pool, boost::bind(&obj::func, &instance2));
    ...
    boost::asio::post(pool, boost::bind(&obj::func, &instancek));

    boost::asio::post(pool, boost::bind(&anotherObj::anotherFunc, &anotherInstance));
}

方案2：使用boost::latch+条件变量（适配动态场景） #

如果func的数量是动态变化的，barrier的固定计数就不适用了，此时可以用boost::latch等待所有func完成，再配合条件变量等待anotherFunc结束：

// 全局或类成员变量
boost::mutex round_mutex;
boost::condition_variable round_cond;
bool round_completed = false;

for(int i = 0; i < n; ++i)
{  
    boost::latch func_latch(k); // 本轮需要等待的func数量

    // 提交instance1的func任务
    boost::asio::post(pool, [&](){
        instance1.func();
        func_latch.count_down(); // 完成后减少latch计数

        // 等待本轮anotherFunc执行完成
        boost::unique_lock<boost::mutex> lock(round_mutex);
        round_cond.wait(lock, [](){ return round_completed; });
    });

    // 同理提交instance2到instancek的func任务...

    // 提交anotherFunc任务
    boost::asio::post(pool, [&](){
        func_latch.wait(); // 阻塞直到所有func完成
        anotherInstance.anotherFunc();

        // 通知所有func线程本轮已完成
        boost::lock_guard<boost::mutex> lock(round_mutex);
        round_completed = true;
        round_cond.notify_all();
    });

    // 重置本轮标志，准备下一轮循环
    boost::lock_guard<boost::mutex> lock(round_mutex);
    round_completed = false;
}

之前代码死锁的原因 #

你之前的实现存在两个核心问题：

1. 无轮次区分：条件变量的notify_all会唤醒所有等待线程，若下一轮的线程已提前进入等待状态，会导致信号被错误消耗，旧轮次线程无法被唤醒
2. 无等待谓词：cond.wait(lock)未添加谓词判断，会出现虚假唤醒，导致线程提前执行，破坏同步逻辑

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