thread_pool_old.hpp

Go to the documentation of this file.

#pragma once
 
#ifdef CPP_TOOLBOX_OLD_THREAD_POOL
 
#  include <vector>  // 用于存储工作线程/For storing worker threads
// #include <queue>             // 不再需要标准队列/No longer need standard
// queue
#  include <thread>  // C++ 线程库/C++ thread library
// #include <mutex>             // 不再需要互斥锁/No longer need mutex
// #include <condition_variable> // 不再需要条件变量/No longer need condition
// variable
#  include <atomic>  // 用于原子布尔标志/For atomic boolean flags
#  include <functional>  // 用于 std::function, std::bind/For std::function, std::bind
#  include <future>  // 用于异步获取任务结果/For async task results
#  include <iostream>  // 用于日志输出/For logging output
#  include <memory>  // 用于智能指针/For smart pointers
#  include <stdexcept>  // 用于异常处理/For exception handling
#  include <type_traits>  // 用于类型特征/For type traits
#  include <utility>  // 用于移动语义/For move semantics
 
#  include <cpp-toolbox/cpp-toolbox_export.hpp>
#  include <cpp-toolbox/macro.hpp>
 
#  include "cpp-toolbox/container/lock_free_queue.hpp"
 
namespace toolbox::base
{
 
class CPP_TOOLBOX_EXPORT thread_pool_t
{
public:
  explicit thread_pool_t(size_t threads = 0);
 
  ~thread_pool_t();
 
  template<class F, class... Args>
  auto submit(F&& f, Args&&... args)
      -> std::future<typename std::invoke_result_t<F, Args...>>;
 
  CPP_TOOLBOX_DISABLE_COPY(thread_pool_t)
  CPP_TOOLBOX_DISABLE_MOVE(thread_pool_t)
 
private:
  std::vector<std::thread> workers_;  // 工作线程列表/List of worker threads
  toolbox::container::lock_free_queue_t<std::function<void()>>
      tasks_;  // 无锁任务队列/Lock-free task queue
  std::atomic<bool> stop_;  // 停止标志/Stop flag
};
 
template<class F, class... Args>
auto thread_pool_t::submit(F&& f, Args&&... args)
    -> std::future<typename std::invoke_result_t<F, Args...>>
{
  using return_type = typename std::invoke_result_t<F, Args...>;
 
  if (stop_.load(
          std::memory_order_relaxed)) {  // 使用 relaxed 因为后续 enqueue
                                         // 有自己的同步/Use relaxed since
                                         // enqueue has its own synchronization
    throw std::runtime_error(
        "在已停止的线程池上提交任务/Submitting task to stopped thread pool");
  }
 
  auto task = std::make_shared<std::packaged_task<return_type()>>(
      std::bind(std::forward<F>(f), std::forward<Args>(args)...));
 
  std::future<return_type> res = task->get_future();
 
  tasks_.enqueue(
      [task /* Capture shared_ptr */]()
      {
        try {
          (*task)();
        } catch (const std::future_error& fe) {
          std::cerr << "Worker lambda caught future_error for task "
                    << task.get() << ": " << fe.what() << std::endl;
        } catch (const std::exception& e) {
          std::cerr << "Worker lambda caught unexpected exception during "
                       "packaged_task invocation for task "
                    << task.get() << ": " << e.what() << std::endl;
        } catch (...) {
          std::cerr << "Worker lambda caught unknown exception during "
                       "packaged_task invocation for task "
                    << task.get() << std::endl;
        }
      });
 
  return res;
}
 
}  // namespace toolbox::base
 
#endif  // CPP_TOOLBOX_OLD_THREAD_POOL