#defineNETWORK_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>

namespacenetwork {namespaceutils {

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;

structthread_pool_pimpl;

structthread_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_poolconst&) =delete;

thread_pool(thread_pool &&other);

thread_pool&operator=(thread_poolconst&) =delete;

thread_pool&operator=(thread_pool &&other);

std::size_tconstthread_count()const;

voidpost(boost::function<void()> f);

~thread_pool();

#ifndef NETWORK_UTILS_THREAD_POOL_IPP_20111021

#defineNETWORK_UTILS_THREAD_POOL_IPP_20111021

#include<vector>

#include<thread>

#include<network/utils/thread_pool.hpp>

namespacenetwork {namespaceutils {

structthread_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(newboost::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_pimplconst &) =delete;

thread_pool_pimpl &operator=(thread_pool_pimplconst &) =delete;

thread_pool_pimpl(thread_pool_pimpl&& other) {

other.swap(*this);

}

std::size_tconstthread_count()const {

return threads_;

}

~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();

}

}

voidswap(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_);

}

std::size_t threads_;

io_service_ptr io_service_;

worker_threads_ptr worker_threads_;

std::vector<std::thread> worker_threads_;

sentinel_ptr sentinel_;

thread_pool_pimpl(thread_pool_pimplconst &);// 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_tconstthread_pool::thread_count()const {

voidthread_pool::swap(thread_pool & other) {

std::swap(other.pimpl,this->pimpl);

}

thread_pool::~thread_pool() {

delete pimpl;

}

}// namespace utils

}// namespace network