Update thread_pool to use std::thread instead of boost::thread.

deanberris · deanberris · commit b7617ff8592a · 2012-11-04T23:17:36.000+11:00
diff --git a/include/network/utils/thread_pool.hpp b/include/network/utils/thread_pool.hpp
@@ -8,7 +8,8 @@
 #define NETWORK_UTILS_THREAD_POOL_HPP_20101020
 
 #include <cstddef>
-#include <boost/thread/thread.hpp>
+#include <vector>
+#include <thread>
 #include <boost/shared_ptr.hpp>
 #include <boost/function.hpp>
 #include <boost/asio/io_service.hpp>
@@ -17,15 +18,18 @@
 namespace network { namespace utils {
 
 typedef boost::shared_ptr<boost::asio::io_service> io_service_ptr;
-typedef boost::shared_ptr<boost::thread_group> worker_threads_ptr;
 typedef boost::shared_ptr<boost::asio::io_service::work> sentinel_ptr;
 
 struct thread_pool_pimpl;
 
 struct thread_pool {
   thread_pool(std::size_t threads = 1,
               io_service_ptr io_service = io_service_ptr(),
-              worker_threads_ptr worker_threads = worker_threads_ptr());
+              std::vector<std::thread> worker_threads = {});
+  thread_pool(thread_pool const&) = delete;
+  thread_pool(thread_pool &&other);
+  thread_pool& operator=(thread_pool const&) = delete;
+  thread_pool& operator=(thread_pool &&other);
   std::size_t const thread_count() const;
   void post(boost::function<void()> f);
   ~thread_pool();
diff --git a/include/network/utils/thread_pool.ipp b/include/network/utils/thread_pool.ipp
@@ -7,57 +7,50 @@
 #ifndef NETWORK_UTILS_THREAD_POOL_IPP_20111021
 #define NETWORK_UTILS_THREAD_POOL_IPP_20111021
 
+#include <vector>
+#include <thread>
 #include <network/utils/thread_pool.hpp>
 
 namespace network { namespace utils {
 
   struct thread_pool_pimpl {
-    thread_pool_pimpl(
-                      std::size_t threads = 1,
+    thread_pool_pimpl(std::size_t threads = 1,
                       io_service_ptr io_service = io_service_ptr(),
-                      worker_threads_ptr worker_threads = worker_threads_ptr()
-                      )
+                      std::vector<std::thread> worker_threads = {})
     : threads_(threads)
     , io_service_(io_service)
-    , worker_threads_(worker_threads)
+    , worker_threads_(std::move(worker_threads))
     , sentinel_()
     {
       bool commit = false;
       BOOST_SCOPE_EXIT((&commit)(&io_service_)(&worker_threads_)(&sentinel_)) {
         if (!commit) {
           sentinel_.reset();
           io_service_.reset();
-          if (worker_threads_.get()) {
-            worker_threads_->interrupt_all();
-            worker_threads_->join_all();
-          }
-          worker_threads_.reset();
+          for (auto& thread : worker_threads_)
+            if (thread.joinable()) thread.join();
+          worker_threads_.clear();
         }
       } BOOST_SCOPE_EXIT_END
 
-      if (!io_service_.get()) {
-        io_service_.reset(new boost::asio::io_service);
-      }
-
-      if (!worker_threads_.get()) {
-        worker_threads_.reset(new boost::thread_group);
-      }
-
-      if (!sentinel_.get()) {
+      if (!io_service_.get()) io_service_.reset(new boost::asio::io_service);
+      if (!sentinel_.get())
         sentinel_.reset(new boost::asio::io_service::work(*io_service_));
-      }
-
+      auto local_io_service = io_service_;
       for (std::size_t counter = 0; counter < threads_; ++counter)
-        worker_threads_->create_thread(
-                                       boost::bind(
-                                                   &boost::asio::io_service::run,
-                                                   io_service_
-                                                   )
-                                       );
+        worker_threads_.emplace_back([local_io_service](){
+          local_io_service->run();});
 
       commit = true;
     }
 
+    thread_pool_pimpl(thread_pool_pimpl const &) = delete;
+    thread_pool_pimpl & operator=(thread_pool_pimpl const &) = delete;
+
+    thread_pool_pimpl(thread_pool_pimpl&& other) {
+      other.swap(*this);
+    }
+
     std::size_t const thread_count() const {
       return threads_;
     }
@@ -69,34 +62,32 @@ namespace network { namespace utils {
     ~thread_pool_pimpl() {
       sentinel_.reset();
       try {
-        worker_threads_->join_all();
+        for (auto& thread : worker_threads_) thread.join();
       } catch (...) {
         BOOST_ASSERT(false && "A handler was not supposed to throw, but one did.");
         std::abort();
       }
     }
 
     void swap(thread_pool_pimpl & other) {
-      std::swap(other.threads_, threads_);
-      std::swap(other.io_service_, io_service_);
-      std::swap(other.worker_threads_, worker_threads_);
-      std::swap(other.sentinel_, sentinel_);
+      using std::swap;
+      swap(other.threads_, threads_);
+      swap(other.io_service_, io_service_);
+      swap(other.worker_threads_, worker_threads_);
+      swap(other.sentinel_, sentinel_);
     }
+
   protected:
     std::size_t threads_;
     io_service_ptr io_service_;
-    worker_threads_ptr worker_threads_;
+    std::vector<std::thread> worker_threads_;
     sentinel_ptr sentinel_;
-
-  private:
-    thread_pool_pimpl(thread_pool_pimpl const &); // no copies please
-    thread_pool_pimpl & operator=(thread_pool_pimpl); // no assignment please
   };
 
   thread_pool::thread_pool(std::size_t threads,
                            io_service_ptr io_service,
-                           worker_threads_ptr worker_threads)
-  : pimpl(new (std::nothrow) thread_pool_pimpl(threads, io_service, worker_threads))
+                           std::vector<std::thread> worker_threads)
+  : pimpl(new (std::nothrow) thread_pool_pimpl(threads, io_service, std::move(worker_threads)))
   {}
 
   std::size_t const thread_pool::thread_count() const {
@@ -110,11 +101,11 @@ namespace network { namespace utils {
   void thread_pool::swap(thread_pool & other) {
     std::swap(other.pimpl, this->pimpl);
   }
-  
+
   thread_pool::~thread_pool() {
     delete pimpl;
   }
-  
+
 }  // namespace utils
 }  // namespace network
 
