C++ Fork Join 并行阻塞

Question

假设您希望并行运行一个部分，然后合并回主线程，然后返回并行部分，依此类推。类似于童年游戏红灯绿灯。

我已经给出了我正在尝试做的事情的示例，我在其中使用条件变量在开始时阻塞线程，但希望并行启动它们，然后在结束时阻塞它们，以便它们可以连续打印出来。 *= 操作可能是跨越许多秒的更大操作。重用线程也很重要。使用任务队列可能过于繁重。

我需要使用某种阻塞结构，而不仅仅是普通的繁忙循环，因为我知道如何使用繁忙循环解决这个问题。

英语:

1. 线程 1 创建 10 个被阻塞的线程
2. 线程 1 向所有线程发出信号以启动(不会互相阻塞)
3. 线程 2-11 处理它们的独占内存
4. 线程 1 正在等待 2-11 完成(这里可以使用原子来计数)
5. 线程 2-11 完成，必要时每个线程都可以通知 1 以检查其状态
6. 线程 1 检查其条件并打印数组
7. 线程 1 重新通知 2-11 以再次处理，从 2 继续

示例代码(Naive 改编自 cplusplus.com 上的示例):

// condition_variable example
#include <iostream>           // std::cout
#include <thread>             // std::thread
#include <mutex>              // std::mutex, std::unique_lock
#include <condition_variable> // std::condition_variable
#include <atomic>

std::mutex mtx;
std::condition_variable cv;
bool ready = false;
std::atomic<int> count(0);

bool end = false;
int a[10];

void doublea (int id) {
  while(!end) {
    std::unique_lock<std::mutex> lck(mtx);
    while (!ready) cv.wait(lck);
    a[id] *= 2;
    count.fetch_add(1);
  }
}

void go() {
  std::unique_lock<std::mutex> lck(mtx);

  ready = true;
  cv.notify_all();
  ready = false; // Naive

  while (count.load() < 10) sleep(1);
  for(int i = 0; i < 10; i++) {
    std::cout << a[i] << std::endl;
  }

  ready = true;
  cv.notify_all();
  ready = false;
  while (count.load() < 10) sleep(1);
  for(int i = 0; i < 10; i++) {
    std::cout << a[i] << std::endl;
  }

  end = true;
  cv.notify_all();
}

int main () {
  std::thread threads[10];
  // spawn 10 threads:
  for (int i=0; i<10; ++i) {
    a[i] = 0;
    threads[i] = std::thread(doublea,i);
  }

  std::cout << "10 threads ready to race...\n";
  go();                       // go!

  return 0;
}

Answer 1

要有效地实现它并不容易。此外，除非您正在学习这门学科，否则它没有任何意义。条件变量在这里不是一个好的选择，因为它不能很好地扩展。

我建议您查看成熟的运行时库如何实现 fork-join 并行性并从中学习或在您的应用中使用它们。参见 http://www.openmprtl.org/ , http://opentbb.org/ , https://www.cilkplus.org/ - 所有这些都是开源的。

OpenMP 是最接近您正在寻找的模型，它具有最有效的 fork-join 障碍实现。但是，它有其缺点，因为它是为 HPC 设计的并且缺乏动态可组合性。 TBB 和 Cilk 最适合嵌套并行性和在模块和库中的使用，这些模块和库可以在外部并行区域的上下文中使用。