如何在 C++ 中同步和合并多个线程的结果？

Question

我有一个数据源不断地输入数据包。有 5 个线程（A、B、C、D、E）处理数据包。请注意，5 个线程的速度完全不同，并且它们为每个传入数据包生成 5 个不同的特征（每个线程生成 1 个特征）。

5 个线程的进度不同：当 A 分析完前 10 个包时，B 可能只完成了包 1、包 2，而 C 可能根本没有完成单个包。

我的任务是匹配5个线程的结果，当前10个数据包的所有5个特征都可用时开始最终分析。

我的问题是： - 如何组合来自不同线程的结果，确保仅在有一定数量的结果可用时才触发分析线程？ - 我似乎需要一个聚合器线程来检查不同缓冲区的可用性。我正在考虑锁定/条件。我如何实现涉及不同缓冲区的这种条件？

多线程方面完全是新手。欢迎任何建议。

我正在使用带有 Boost 库的 GNU C++。

Answer 1

拥有一个“聚合器”线程：该线程将从工作线程获取输入（我建议通过非阻塞线程安全队列），一旦“批次”准备就绪，将其推送到“分析器”线程。

队列具有不阻塞任何工作人员的优点：“聚合器”只需轮询工作人员队列（通过条件部分）。您可以根据自己的喜好控制轮询速率。

该解决方案解决了“同步所有”情况的问题。

Answer 2

您可能需要检查生产者-消费者问题

Answer 3

使用信号量和额外的布尔“完成”变量。每次线程完成时，它首先写入其答案，然后写入其“完成”变量，然后调用“检查”函数来检查所有线程的“完成”变量，如果它们全部为真，则触发分析线程。

根据您的性能权衡，您可能只需要最慢的“工作”线程来调用“检查”函数，因此快速线程不会继续锁定其“完成”变量以进行读取。当然，这取决于知道哪个是最慢的。

我不知道你的重置策略：你想每次等待10个新输入还是连续分析最近10个？

Answer 4

一些伪代码：

worker thread: 
   -> do work, 
     -> publish result to queue
   -> 10 reached, signal my condvar 

aggregator thread: 
   -> wait on all condvars. 
   -> lock all result queues, swap in new empty ones.  
   -> do aggregation processing. 

创建新队列的原因是，如果删除项目，聚合处理可能会导致大量锁定和失效 - 如果将新队列放在工作线程上，则无需担心锁定（特别是当聚合器不需要与工作人员共享其结果）。

Answer 5

屏障是规范的“同步所有”操作。

但是，听起来您希望在关键部分中有一个“计数结果”变量，当完成一定数量时该变量会递增。然后，您想要执行“阻止直到变量等于 x”。这可以通过针对计数结果变量的自旋锁来完成。

Answer 6

拥有一个存储结果的容器和这样的函数（伪代码）：

void storeResult(Result result) {
    Mutex m("ResultContainerMutex");

    container.push_back(result);
    if(container.size() > ANALYSIS_SIZE) {
        StartAnalysisThread(container.copy());
        container.clear();
    }
}

由于互斥体仅保护添加到容器的操作，该操作相当快，因此不应导致过度序列化。

Answer 7

根据您当前的设计，您受到最慢计算的限制，其他线程将无法充分使用。

如果你想处理很多数据包，我会像这样分割工作：

将数据包分发到 N 个相同的线程，该线程按顺序计算它们收到的数据包的 5 个结果。

每个线程将其结果数据包放入线程安全的 fifo 中。

您的主线程读取结果，并在需要时使用数据包编号对它们重新排序