C 11中的线程池

Question

有关问题：

关于C 11：

• C 11： std::线程池？
• C 11中的异步（启动::a同步）会使线程池过时，以避免昂贵的线程创建吗？

关于升压：

• C提升线程重用线程
• 提升::线程并创建一个线程池！

我如何获得发送任务的线程池，而不必一次又一次地创建和删除它们？这意味着持久性线程可以在不加入的情况下重新同步。

我有如下代码：

namespace {
  std::vector<std::thread> workers;

  int total = 4;
  int arr[4] = {0};

  void each_thread_does(int i) {
    arr[i] += 2;
  }
}

int main(int argc, char *argv[]) {
  for (int i = 0; i < 8; ++i) { // for 8 iterations,
    for (int j = 0; j < 4; ++j) {
      workers.push_back(std::thread(each_thread_does, j));
    }
    for (std::thread &t: workers) {
      if (t.joinable()) {
        t.join();
      }
    }
    arr[4] = std::min_element(arr, arr+4);
  }
  return 0;
}

与每次迭代都创建和加入线程不同，我更愿意在每次迭代中将任务发送给我的工作线程，并且只创建一次。

Answer 1

线程池意味着所有线程都在运行，也就是说，线程函数永远不会返回。为了给线程一些有意义的事情去做，您必须设计一个线程间通信系统，既可以告诉线程有事情要做，也可以通信实际的工作数据。

通常，这将涉及某种并发数据结构，每个线程可能会在某种条件变量上Hibernate，当有工作要做时，会通知这些条件变量。收到通知后，一个或多个线程将唤醒，从并发数据结构中恢复任务，对其进行处理，并以类似的方式存储结果。

然后，线程会继续检查是否有更多的工作要做，如果没有，就继续睡觉。

结果是你必须自己设计所有这些，因为没有一个普遍适用的“工作”的自然概念。这是相当多的工作，有些微妙的问题你必须解决。（如果你喜欢一个在幕后为你处理线程管理的系统，你可以在Go中编程。）

Answer 2

您可以使用C线程池库，https://github.com/vit-vit/ctpl.

然后您编写的代码可以替换为以下代码

#include <ctpl.h>  // or <ctpl_stl.h> if ou do not have Boost library

int main (int argc, char *argv[]) {
    ctpl::thread_pool p(2 /* two threads in the pool */);
    int arr[4] = {0};
    std::vector<std::future<void>> results(4);
    for (int i = 0; i < 8; ++i) { // for 8 iterations,
        for (int j = 0; j < 4; ++j) {
            results[j] = p.push([&arr, j](int){ arr[j] +=2; });
        }
        for (int j = 0; j < 4; ++j) {
            results[j].get();
        }
        arr[4] = std::min_element(arr, arr + 4);
    }
}

您将获得所需的线程数，并且不会在迭代中一遍又一遍地创建和删除它们。

Answer 3

这是从我的回答复制到另一个非常类似的帖子：

>

从系统可以支持的最大线程数开始：

int num_threads = std::thread::hardware_concurrency();

为了实现高效的线程池，一旦根据num_threads创建了线程，最好不要创建新线程，也不要破坏旧线程（通过加入）。这将导致性能下降，甚至可能使应用程序运行速度比串行版本慢。

每个C11线程应该以无限循环的方式在其函数中运行，不断等待新任务的抓取和运行。

下面是如何将这样一个函数附加到线程池：

int num_threads = std::thread::hardware_concurrency();
std::vector<std::thread> threads;
for (int i = 0; i < num_threads; i++)
{
    pool.push_back(std::thread(Infinite_loop_function));
}

void Pool::Infinite_loop_function()
{
    while (true)
    {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);

            condition.wait(lock, [this](){
                return !queue.empty() || terminate_pool;
            });
            Job = queue.front();
            queue.pop();
        }

        Job(); // function<void()> type
    }
};

void Pool::Add_Job(function<void()> New_Job)
{
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        queue.push(New_Job);
    }

    condition.notify_one();
}

Pool_Obj.Add_Job(std::bind(&Some_Class::Some_Method, &Some_object));

一旦您集成了这些成分，您就拥有了自己的动态线程池。这些线程总是运行，等待作业完成。

如果有一些语法错误，我道歉，我打了这个代码，我有一个坏的记忆。很抱歉，我不能提供完整的线程池代码；这将违反我的工作完整性。

编辑：要终止池，请调用Shutdown（）方法：

Pool::shutdown()
{
    {
        std::unique_lock<std::mutex> lock(threadpool_mutex);
        terminate_pool = true; // use this flag in condition.wait
    }

    condition.notify_all(); // wake up all threads.

    // Join all threads.
    for (std::thread &th : threads)
    {
        th.join();
    }

    pool.clear();  
    stopped = true; // use this flag in destructor, if not set, call shutdown() 
}