Dec 18, 2013 · Jun 23, 2013 · Jun 23, 2013 · Jun 23, 2013 · Jun 23, 2013 · Jun 23, 2013
diff --git a/concurrency/src/network/concurrency/thread_pool.hpp b/concurrency/src/network/concurrency/thread_pool.hpp
 #include <boost/asio/io_service.hpp>

 namespace network {
 namespace concurrency {

 typedef std::shared_ptr<boost::asio::io_service> io_service_ptr;
 typedef std::shared_ptr<std::vector<std::thread>> worker_threads_ptr;
 typedef std::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(),
      std::vector<std::thread> worker_threads = std::vector<std::thread>());
 #if !defined(BOOST_NO_CXX11_DELETED_FUNCTIONS)
  thread_pool(thread_pool const&) = delete;
 #endif  // !defined(BOOST_NO_CXX11_DELETEED_FUNCTIONS)
  thread_pool(thread_pool && other);
 #if !defined(BOOST_NO_CXX11_DELETED_FUNCTIONS)
  thread_pool& operator=(thread_pool const&) = delete;
 #endif  // !defined(BOOST_NO_CXX11_DELETEED_FUNCTIONS)
  thread_pool& operator=(thread_pool && other);
  std::size_t const thread_count() const;
  void post(std::function<void()> f);
  ~thread_pool();
  void swap(thread_pool& other);
 protected:
  thread_pool_pimpl* pimpl;
 };

 inline void swap(thread_pool& l, thread_pool& r) { l.swap(r); }

 }  // namespace concurrency
  namespace concurrency {

    typedef std::shared_ptr<boost::asio::io_service> io_service_ptr;
    typedef std::shared_ptr<std::vector<std::thread>> worker_threads_ptr;
    typedef std::shared_ptr<boost::asio::io_service::work> sentinel_ptr;

    struct thread_pool {
      thread_pool(std::size_t threads = 1,
  io_service_ptr io_service = io_service_ptr(),
  std::vector<std::thread> worker_threads = std::vector<std::thread>());
      thread_pool(thread_pool const&) = delete;
      thread_pool(thread_pool && other);
      ~thread_pool();

      thread_pool& operator=(thread_pool const&) = delete;
      thread_pool& operator=(thread_pool && other);

      std::size_t const thread_count() const;
      void post(std::function<void()> f);
      void swap(thread_pool& other);

    private:

      struct pimpl;
      pimpl* pimpl_;

    };

    inline
    void swap(thread_pool& l, thread_pool& r) {
      l.swap(r);
    }

  }  // namespace concurrency
 }  // namespace network

 #endif /* NETWORK_CONCURRENCY_THREAD_POOL_HPP_20101020 */
diff --git a/concurrency/src/network/concurrency/thread_pool.ipp b/concurrency/src/network/concurrency/thread_pool.ipp
 #include <boost/scope_exit.hpp>

 namespace network {
 namespace concurrency {

 struct thread_pool_pimpl {
  thread_pool_pimpl(
      std::size_t threads = 1,
      io_service_ptr io_service = io_service_ptr(),
      std::vector<std::thread> worker_threads = std::vector<std::thread>())
      : threads_(threads),
        io_service_(io_service),
        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();
        for (auto& thread : worker_threads_)
          if (thread.joinable())
            thread.join();
        worker_threads_.clear();
  namespace concurrency {

    struct thread_pool::pimpl {
      pimpl(std::size_t threads = 1,
    io_service_ptr io_service = io_service_ptr(),
    std::vector<std::thread> worker_threads = std::vector<std::thread>())
 : threads_(threads),
  io_service_(io_service),
  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();
    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 (!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_.emplace_back([local_io_service]() {
      local_io_service->run();
    });
 }

 commit = true;
      }
    }
    BOOST_SCOPE_EXIT_END 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_.emplace_back([local_io_service]() {
        local_io_service->run();
      });

    commit = true;
  }

 #if !defined(BOOST_NO_CXX11_DEFAULTED_FUNCTIONS)
  thread_pool_pimpl(thread_pool_pimpl const&) = delete;
  thread_pool_pimpl& operator=(thread_pool_pimpl const&) = delete;
 #endif  // !defined(BOOST_NO_CXX11_DEFAULTED_FUNCTIONS)
      ~pimpl() {
 sentinel_.reset();
 try {
  for (auto& thread : worker_threads_)
    thread.join();
 }
 catch (...) {
  BOOST_ASSERT(false &&
       "A handler was not supposed to throw, but one did.");
  std::abort();
 }
      }

  thread_pool_pimpl(thread_pool_pimpl && other) { other.swap(*this); }
      std::size_t threads_;
      io_service_ptr io_service_;
      std::vector<std::thread> worker_threads_;
      sentinel_ptr sentinel_;

 std::size_t const thread_count() const { return threads_; }
  };

  void post(std::function<void()> f) { io_service_->post(f); }
  thread_pool::thread_pool(std::size_t threads,
   io_service_ptr io_service,
   std::vector<std::thread> worker_threads)
    : pimpl_(new (std::nothrow)
     pimpl(threads, io_service, std::move(worker_threads))) {

  ~thread_pool_pimpl() {
    sentinel_.reset();
    try {
      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) {
    using std::swap;
    swap(other.threads_, threads_);
    swap(other.io_service_, io_service_);
    swap(other.worker_threads_, worker_threads_);
    swap(other.sentinel_, sentinel_);
  std::size_t const thread_pool::thread_count() const {
    return pimpl_->threads_;
  }

 protected:
  std::size_t threads_;
  io_service_ptr io_service_;
  std::vector<std::thread> worker_threads_;
  sentinel_ptr sentinel_;
 };

 thread_pool::thread_pool(std::size_t threads,
                         io_service_ptr io_service,
                         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 {
  return pimpl->thread_count();
 }

 void thread_pool::post(std::function<void()> f) { pimpl->post(f); }

 void thread_pool::swap(thread_pool& other) {
  std::swap(other.pimpl, this->pimpl);
 }
  void thread_pool::post(std::function<void()> f) {
    pimpl_->io_service_->post(f);
  }

 thread_pool::~thread_pool() { delete pimpl; }
  void thread_pool::swap(thread_pool& other) {
    std::swap(other.pimpl_, this->pimpl_);
  }

  thread_pool::~thread_pool() {
    delete pimpl_;
  }
 }  // namespace concurrency
 }  // namespace network
Original file line number	Diff line number	Diff line change
Expand Up		@@ -16,38 +16,40 @@
		#include <boost/asio/io_service.hpp>

		namespace network {
		namespace concurrency {

		typedef std::shared_ptr<boost::asio::io_service> io_service_ptr;
		typedef std::shared_ptr<std::vector<std::thread>> worker_threads_ptr;
		typedef std::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(),
		std::vector<std::thread> worker_threads = std::vector<std::thread>());
		#if !defined(BOOST_NO_CXX11_DELETED_FUNCTIONS)
		thread_pool(thread_pool const&) = delete;
		#endif // !defined(BOOST_NO_CXX11_DELETEED_FUNCTIONS)
		thread_pool(thread_pool && other);
		#if !defined(BOOST_NO_CXX11_DELETED_FUNCTIONS)
		thread_pool& operator=(thread_pool const&) = delete;
		#endif // !defined(BOOST_NO_CXX11_DELETEED_FUNCTIONS)
		thread_pool& operator=(thread_pool && other);
		std::size_t const thread_count() const;
		void post(std::function<void()> f);
		~thread_pool();
		void swap(thread_pool& other);
		protected:
		thread_pool_pimpl* pimpl;
		};

		inline void swap(thread_pool& l, thread_pool& r) { l.swap(r); }

		} // namespace concurrency
		namespace concurrency {

		typedef std::shared_ptr<boost::asio::io_service> io_service_ptr;
		typedef std::shared_ptr<std::vector<std::thread>> worker_threads_ptr;
		typedef std::shared_ptr<boost::asio::io_service::work> sentinel_ptr;

		struct thread_pool {
		thread_pool(std::size_t threads = 1,
		io_service_ptr io_service = io_service_ptr(),
		std::vector<std::thread> worker_threads = std::vector<std::thread>());
		thread_pool(thread_pool const&) = delete;
		thread_pool(thread_pool && other);
		~thread_pool();

		thread_pool& operator=(thread_pool const&) = delete;
		thread_pool& operator=(thread_pool && other);

		std::size_t const thread_count() const;
		void post(std::function<void()> f);
		void swap(thread_pool& other);

		private:

		struct pimpl;
		pimpl* pimpl_;

		};

		inline
		void swap(thread_pool& l, thread_pool& r) {
		l.swap(r);
		}

		} // namespace concurrency
		} // namespace network

		#endif /* NETWORK_CONCURRENCY_THREAD_POOL_HPP_20101020 */
Original file line number	Diff line number	Diff line change
Expand Up		@@ -14,99 +14,93 @@
		#include <boost/scope_exit.hpp>

		namespace network {
		namespace concurrency {

		struct thread_pool_pimpl {
		thread_pool_pimpl(
		std::size_t threads = 1,
		io_service_ptr io_service = io_service_ptr(),
		std::vector<std::thread> worker_threads = std::vector<std::thread>())
		: threads_(threads),
		io_service_(io_service),
		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();
		for (auto& thread : worker_threads_)
		if (thread.joinable())
		thread.join();
		worker_threads_.clear();
		namespace concurrency {

		struct thread_pool::pimpl {
		pimpl(std::size_t threads = 1,
		io_service_ptr io_service = io_service_ptr(),
		std::vector<std::thread> worker_threads = std::vector<std::thread>())
		: threads_(threads),
		io_service_(io_service),
		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();
		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 (!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_.emplace_back([local_io_service]() {
		local_io_service->run();
		});
		}

		commit = true;
		}
		}
		BOOST_SCOPE_EXIT_END 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_.emplace_back([local_io_service]() {
		local_io_service->run();
		});

		commit = true;
		}

		#if !defined(BOOST_NO_CXX11_DEFAULTED_FUNCTIONS)
		thread_pool_pimpl(thread_pool_pimpl const&) = delete;
		thread_pool_pimpl& operator=(thread_pool_pimpl const&) = delete;
		#endif // !defined(BOOST_NO_CXX11_DEFAULTED_FUNCTIONS)
		~pimpl() {
		sentinel_.reset();
		try {
		for (auto& thread : worker_threads_)
		thread.join();
		}
		catch (...) {
		BOOST_ASSERT(false &&
		"A handler was not supposed to throw, but one did.");
		std::abort();
		}
		}

		thread_pool_pimpl(thread_pool_pimpl && other) { other.swap(*this); }
		std::size_t threads_;
		io_service_ptr io_service_;
		std::vector<std::thread> worker_threads_;
		sentinel_ptr sentinel_;

		std::size_t const thread_count() const { return threads_; }
		};

		void post(std::function<void()> f) { io_service_->post(f); }
		thread_pool::thread_pool(std::size_t threads,
		io_service_ptr io_service,
		std::vector<std::thread> worker_threads)
		: pimpl_(new (std::nothrow)
		pimpl(threads, io_service, std::move(worker_threads))) {

		~thread_pool_pimpl() {
		sentinel_.reset();
		try {
		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) {
		using std::swap;
		swap(other.threads_, threads_);
		swap(other.io_service_, io_service_);
		swap(other.worker_threads_, worker_threads_);
		swap(other.sentinel_, sentinel_);
		std::size_t const thread_pool::thread_count() const {
		return pimpl_->threads_;
		}

		protected:
		std::size_t threads_;
		io_service_ptr io_service_;
		std::vector<std::thread> worker_threads_;
		sentinel_ptr sentinel_;
		};

		thread_pool::thread_pool(std::size_t threads,
		io_service_ptr io_service,
		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 {
		return pimpl->thread_count();
		}

		void thread_pool::post(std::function<void()> f) { pimpl->post(f); }

		void thread_pool::swap(thread_pool& other) {
		std::swap(other.pimpl, this->pimpl);
		}
		void thread_pool::post(std::function<void()> f) {
		pimpl_->io_service_->post(f);
		}

		thread_pool::~thread_pool() { delete pimpl; }
		void thread_pool::swap(thread_pool& other) {
		std::swap(other.pimpl_, this->pimpl_);
		}

		thread_pool::~thread_pool() {
		delete pimpl_;
		}
		} // namespace concurrency
		} // namespace network

Expand Down