C++线程池如何处理高递归任务？求现代C++实现方案

问题描述 #

我计划编写一个C++线程池，支持用户提交并执行任务。比如在8核机器上，理想状态下8个工作线程可并行执行。

但深入思考后发现一个问题：假设存在计算斐波那契数列的任务：

int fib(int n) {
  if (n <= 1) {
    return 1;
  }
  return fib(n-1) + fib(n-2);
}

若使用线程池处理，我可能会写出如下代码：

int fib(int n) {
  if (n <= 1) {
    return 1;
  }
  // 以可等待的std::future形式返回
  auto fut_n_1 = pool.submit(fib, n-1);
  auto fut_n_2 = pool.submit(fib, n-2);
  return fut_n_1.get() + fut_n_2.get();
}

但当fib(n)在工作线程上执行时，会生成两个新任务提交给其他工作线程，同时必须等待fut_n_1和fut_n_2的结果。如此一来，很快所有8个工作线程都会陷入阻塞状态。即便允许生成更多工作线程，线程过度订阅的开销也会迅速降低系统性能。

我并非询问如何优化斐波那契函数的写法，而是想了解如何设计线程池，使其在处理如快速排序这类高递归任务时，能够保存上下文并让出线程资源。我曾见过类似Clik这类古老系统使用sys_set_jump实现该功能，但不想使用如此底层的细节。是否存在现代C++技术能实现类似效果？

当前线程池实现代码 #

头文件 #

class ThreadPool {
 public:
  // 为主线程保留一个线程
  explicit ThreadPool(int size = std::thread::hardware_concurrency() - 1);

  ~ThreadPool();

  template <typename F, typename... Args>
  decltype(auto) SubmitTask(F &&new_task, Args &&...args);

  void Exit();

  auto GetSize() -> size_t;

 private:
  std::vector<std::thread> threads_;
  std::queue<std::function<void()>> tasks_;
  std::mutex mtx_;
  std::condition_variable cv_;
  std::atomic<bool> exit_{false};
};

template <typename F, typename... Args>
auto ThreadPool::SubmitTask(F &&new_task, Args &&...args) -> decltype(auto) {
  using return_type = std::invoke_result_t<F, Args...>;
  if (exit_) {
    throw std::runtime_error("ThreadPool: SubmitTask() called while already exit_ being true");
  }
  auto packaged_new_task = std::make_shared<std::packaged_task<return_type()>>(
      std::bind(std::forward<F>(new_task), std::forward<Args>(args)...));
  auto fut = packaged_new_task->get_future();
  {
    // 以std::function形式提交至线程池任务队列
    std::unique_lock<std::mutex> lock(mtx_);
    tasks_.emplace([packaged_new_task]() { (*packaged_new_task)(); });
  }
  cv_.notify_one();
  return fut;
}

实现文件 #

ThreadPool::ThreadPool(int size) {
  /* std::thread::hardware_concurrency() might return 0 if sys info not
   * available */
  size = std::max(size, MIN_NUM_THREADS_IN_POOL);
  for (auto i = 0; i < size; i++) {
    threads_.emplace_back([this]() {
      while (true) {
        std::function<void()> next_task;
        {
          std::unique_lock<std::mutex> lock(mtx_);
          cv_.wait(lock, [this]() { return exit_ || !tasks_.empty(); });
          if (exit_ && tasks_.empty()) {
            return;  // 线程生命周期结束
          }
          next_task = tasks_.front();
          tasks_.pop();
        }
        next_task();
      }
    });
  }
}

ThreadPool::~ThreadPool() {
  Exit();
  for (auto &worker : threads_) {
    if (worker.joinable()) {
      worker.join();
    }
  }
}

void ThreadPool::Exit() {
  exit_ = true;
  cv_.notify_all();
}

auto ThreadPool::GetSize() -> size_t { return threads_.size(); }

解决方案 #

核心思路是让线程在等待std::future结果时，不阻塞线程，而是主动从任务队列中取出新任务执行，直到等待的future就绪。以下是三种现代C++实现方案：

方案一：C++20协程（优雅的上下文管理） #

协程允许任务在等待时挂起，自动保存上下文，让出线程给其他任务执行，直到等待条件满足后恢复。

1. 定义协程任务类型：

#include <coroutine>
#include <future>

template<typename T>
struct Task {
    struct promise_type {
        std::promise<T> prom;

        Task get_return_object() {
            return Task{std::coroutine_handle<promise_type>::from_promise(*this)};
        }
        std::suspend_never initial_suspend() { return {}; }
        std::suspend_never final_suspend() noexcept { return {}; }
        void return_value(T value) { prom.set_value(value); }
        void unhandled_exception() { prom.set_exception(std::current_exception()); }
    };

    std::coroutine_handle<promise_type> handle;
    std::future<T> fut;

    Task(std::coroutine_handle<promise_type> h) : handle(h), fut(h.promise().prom.get_future()) {}
    ~Task() { if (handle) handle.destroy(); }

    // 让任务支持等待逻辑，适配线程池调度
    bool await_ready() { return fut.is_ready(); }
    void await_suspend(std::coroutine_handle<> caller) {
        // 简化实现：实际可将caller重新加入线程池队列，待fut就绪后恢复
        fut.wait();
        caller.resume();
    }
    T await_resume() { return fut.get(); }
};

2. 修改线程池适配协程：
   调整任务队列，支持保存挂起的协程句柄，并在future就绪时恢复协程执行，让线程在协程挂起期间处理其他任务。

方案二：协作式等待（无需C++20） #

在等待future时，主动从任务队列中窃取任务执行，直到目标future就绪。自定义wait_with_work函数替换直接调用future.get()：

template<typename T>
T wait_with_work(ThreadPool& pool, std::future<T>& fut) {
    while (!fut.is_ready()) {
        std::function<void()> task;
        {
            std::unique_lock<std::mutex> lock(pool.mtx_);
            if (pool.tasks_.empty()) {
                // 无任务可处理时短暂等待
                pool.cv_.wait_for(lock, std::chrono::milliseconds(10));
                continue;
            }
            task = std::move(pool.tasks_.front());
            pool.tasks_.pop();
        }
        if (task) task();
    }
    return fut.get();
}

修改斐波那契函数调用：

int fib(int n, ThreadPool& pool) {
  if (n <= 1) {
    return 1;
  }
  auto fut_n_1 = pool.SubmitTask(fib, n-1, std::ref(pool));
  auto fut_n_2 = pool.SubmitTask(fib, n-2, std::ref(pool));
  return wait_with_work(pool, fut_n_1) + wait_with_work(pool, fut_n_2);
}

该方案兼容性强，通过在等待时处理其他任务避免线程阻塞，需注意线程池内部成员的访问权限（可将wait_with_work设为线程池成员函数）。

方案三：任务窃取线程池 #

传统固定大小线程池易阻塞，任务窃取线程池（每个线程维护本地任务队列）能更好处理递归任务：线程优先处理本地队列任务，本地队列空时从其他线程队列末尾窃取任务执行，结合协作式等待最大化线程利用率。

实现要点：

• 每个工作线程维护本地任务队列（如std::deque，支持两端操作）
• 线程优先处理本地任务，本地队列空时随机窃取其他线程队列的任务
• 任务等待子任务结果时，主动处理其他任务直到子任务完成

总结 #

• C++20协程是最优雅的方案，自动管理任务挂起与恢复，无需手动调度
• 协作式等待方案兼容性好，无需C++20支持，通过等待时处理其他任务避免阻塞
• 任务窃取线程池+协作式调度，能最大化多核CPU利用率，适配递归型并行任务

内容的提问来源于stack exchange，提问作者Love Cute Shiba