让线程池的所有线程执行给定的函数

Posted 2023-04-12

【中文标题】让线程池的所有线程执行给定的函数

【英文标题】：Have all threads of a thread pool execute a given function

【发布时间】：2020-01-31 02:46:13

【问题描述】：

我使用了一个稍微定制的 progschj 线程池实现版本 (https://github.com/progschj/ThreadPool)。到目前为止，它适用于我的用例，除了后面描述的那个。

我希望能够让池中的每个线程执行一次给定函数，例如让每个线程设置一些线程局部变量。

thread_pool.execute_all([] ()  do some stuff; );

这个调用应该是阻塞的，等待池中的所有线程都执行完这个函数。

我试图修改线程的主循环来实现这一点，但我无法确保池中的每个线程都只执行一次给定的函数，并且主线程一直等待直到它们完成。

谁能提出这个功能的工作版本，或者给出如何实现这个功能的提示？

【问题讨论】：

thread_pool好像没有这样的功能。在问我们如何添加它？如果是这样，您要求我们为您修改库。这会使您的问题脱离本网站的主题。

是的，我实际上尝试将此功能添加到线程池中，但无法获得有效的解决方案。我希望有人能认为这是一个有趣的问题，并且至少可以给出关于如何制作它的一些提示（使用这个库或任何其他线程池实现）。

看来你得玩一下线程的join方法。

join不是在等待线程停止吗？我不想停止线程，只等待它们每个都执行给定的函数。然后他们应该可以继续执行新任务。

出于好奇，您要解决什么问题？编写关心线程池工作人员身份的代码似乎有点可疑。您现在使用的 thread_pool 实现是微不足道的，但如果您切换到更复杂的实现，那么您需要注意线程池在需求低时杀死工作人员并创建新的常见做法需求高涨时的工人。

【参考方案1】：

写一个计数信号屏障。

struct counting_barrier 
  explicit counting_barrier( std::ptrdiff_t c ):count(c) 
  void operator--(int)& 
    this->operator--();
  
  void operator--()& 
    auto l = lock();
    --count;
    cv.wait( l, [&]
      return count<=0;
     );
    cv.notify_all();
  
private:
  std::unique_lock<std::mutex> lock()& 
    return std::unique_lock<std::mutex>(m);
  
  std::condition_variable cv;
  std::mutex m;
  std::ptrdiff_t count = 0;
;

现在我们可以使用现有的原语来实现它。

template<class F>
void execute_all( F&& f ) 
  counting_barrier barrier(workers.size()+1);
  for (std::size_t i = 0; i < workers.size(); ++i) 
    enqueue([&]
      f();
      --barrier;
    );
  
  --barrier;

应该这样做。在workers.size() 击中之前，不允许工人越过障碍；当最后一个工人释放屏障时，workers.size()-1 必须被阻止。因此，每个工作人员都会在函数返回之前调用f()。

代码未经测试。

【讨论】：

非常感谢。它可以通过 2 处修改正常工作：wait_for 应该是 wait in counting_barrier::operator--。由于我希望 execute_all 函数等待线程全部调用该函数，因此我存储了 enqueue 返回的 futures 并等待它们。

@PierreKraemer 当障碍减少到 0 时，所有的 f()s 都被调用了。等待期货是多余的。哦等等，你是对的，给障碍加一个；在那里，现在execute_all 也阻挡了障碍物。无需两种同步！

是的。在主函数中使用workers.size()+1 计数初始化和最后一个--barrier，就可以了:-)