Пул потоков на основе boost::lockfree::queue - C++ - Ответ 6457442

class function_wrapper
{
   struct impl_base {
      virtual void call() = 0;
      virtual ~impl_base() {}
   };
 
   std::shared_ptr<impl_base> impl;
 
   template <typename F>
   struct impl_type : impl_base
   {
      F f;
      impl_type(F&& f_) : f(std::move(f_)) {}
      void call() override { f(); }
   };
 
public:
   template <typename F>
   function_wrapper(F&& f) : impl(new impl_type<F>(std::move(f)))
   {}
 
   function_wrapper() = default;
   function_wrapper(const function_wrapper&) = default;
   function_wrapper(function_wrapper&) = default;
   ~function_wrapper() = default;
 
   void operator ()() { impl->call(); }
};
 
class join_thread
{
   std::vector<std::thread>& threads;
public:
   explicit join_thread(std::vector<std::thread>& threads_) :
      threads(threads_)
   {}
   ~join_thread()
   {
      for (auto& thread : threads) {
         if (thread.joinable()) thread.join();
      }
   }
};
 
boost::lockfree::spsc_queue<function_wrapper> work_queue(128);
 
class thread_pool
{
   std::atomic<bool> done;
   std::vector<std::thread> threads;
   join_thread joiner;
 
   void worker_thread()
   {
      while (!done)
      {
         function_wrapper task;
         if (work_queue.pop(task))
            task();
         else
            std::this_thread::yield();
      }
   }
 
public:
   thread_pool() : done(false), joiner(threads)
   {
      size_t const thread_count = std::thread::hardware_concurrency();
 
      if (thread_count <= 1 || !work_queue.is_lock_free())
         throw std::logic_error("not compatible system for thread_pool");
 
      try {
         for (size_t i = 0; i < thread_count; ++i)
            threads.emplace_back(&thread_pool::worker_thread, this);
      }
      catch(...) {
         done = true;
         throw;
      }
   }
 
   ~thread_pool() { done = true; }
 
   template <typename FunctionType>
   std::future<typename std::result_of<FunctionType()>::type>
   submit(FunctionType f)
   {
      typedef typename std::result_of<FunctionType()>::type
            result_type;
      std::packaged_task<result_type()> task(std::move(f));
      std::future<result_type> res(task.get_future());
      work_queue.push(std::move(task));
      return res;
   }
};