diff --git a/example/executor.cpp b/example/executor.cpp index d1a7c033..119904e5 100644 --- a/example/executor.cpp +++ b/example/executor.cpp @@ -17,7 +17,7 @@ #include #include #include -#include +#include #include #include #include @@ -115,12 +115,12 @@ int test_executor_adaptor() // std::cout << BOOST_CONTEXTOF << std::endl; { boost::basic_thread_pool tp; - boost::serial_executor e1(tp); - boost::serial_executor e2 = e1; + boost::generic_serial_executor e1(tp); + boost::generic_serial_executor e2 = e1; } { boost::executor_adaptor < boost::basic_thread_pool > ea1(4); - boost::executor_adaptor < boost::serial_executor > ea2(ea1); + boost::executor_adaptor < boost::generic_serial_executor > ea2(ea1); submit_some(ea2); } #endif diff --git a/example/generic_serial_executor.cpp b/example/generic_serial_executor.cpp new file mode 100644 index 00000000..149d5cac --- /dev/null +++ b/example/generic_serial_executor.cpp @@ -0,0 +1,112 @@ +// Copyright (C) 2015 Vicente J. Botet Escriba +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) + +#include +#if ! defined BOOST_NO_CXX11_DECLTYPE +#define BOOST_RESULT_OF_USE_DECLTYPE +#endif + +#define BOOST_THREAD_VERSION 4 +#define BOOST_THREAD_PROVIDES_EXECUTORS +//#define BOOST_THREAD_USES_LOG +#define BOOST_THREAD_USES_LOG_THREAD_ID +#define BOOST_THREAD_QUEUE_DEPRECATE_OLD + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +void p1() +{ + std::cout << BOOST_CONTEXTOF << std::endl; + boost::this_thread::sleep_for(boost::chrono::milliseconds(30)); + std::cout << BOOST_CONTEXTOF << std::endl; +} + +void p2() +{ + std::cout << BOOST_CONTEXTOF << std::endl; + boost::this_thread::sleep_for(boost::chrono::milliseconds(10)); + std::cout << BOOST_CONTEXTOF << std::endl; +} + +int f1() +{ + // std::cout << BOOST_CONTEXTOF << std::endl; + boost::this_thread::sleep_for(boost::chrono::seconds(1)); + return 1; +} +int f2(int i) +{ + // std::cout << BOOST_CONTEXTOF << std::endl; + boost::this_thread::sleep_for(boost::chrono::seconds(2)); + return i + 1; +} + +void submit_some(boost::generic_serial_executor& tp) +{ + std::cout << BOOST_CONTEXTOF << std::endl; + for (int i = 0; i < 3; ++i) { + std::cout << BOOST_CONTEXTOF << std::endl; + tp.submit(&p2); + } + for (int i = 0; i < 3; ++i) { + std::cout << BOOST_CONTEXTOF << std::endl; + tp.submit(&p1); + } + std::cout << BOOST_CONTEXTOF << std::endl; + +} + + +void at_th_entry(boost::basic_thread_pool& ) +{ + +} + +int test_executor_adaptor() +{ + // std::cout << BOOST_CONTEXTOF << std::endl; + { + try + { + +#if ! defined(BOOST_NO_CXX11_RVALUE_REFERENCES) + // std::cout << BOOST_CONTEXTOF << std::endl; + { + boost::basic_thread_pool ea1(4); + boost::generic_serial_executor ea2(ea1); + submit_some(ea2); + } +#endif + // std::cout << BOOST_CONTEXTOF << std::endl; + } + catch (std::exception& ex) + { + std::cout << "ERROR= " << ex.what() << "" << std::endl; + return 1; + } + catch (...) + { + std::cout << " ERROR= exception thrown" << std::endl; + return 2; + } + } + // std::cout << BOOST_CONTEXTOF << std::endl; + return 0; +} + + +int main() +{ + return test_executor_adaptor(); +} diff --git a/example/generic_serial_executor_cont.cpp b/example/generic_serial_executor_cont.cpp new file mode 100644 index 00000000..eb3bbc81 --- /dev/null +++ b/example/generic_serial_executor_cont.cpp @@ -0,0 +1,112 @@ +// Copyright (C) 2015 Vicente J. Botet Escriba +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) + +#include +#if ! defined BOOST_NO_CXX11_DECLTYPE +#define BOOST_RESULT_OF_USE_DECLTYPE +#endif + +#define BOOST_THREAD_VERSION 4 +#define BOOST_THREAD_PROVIDES_EXECUTORS +//#define BOOST_THREAD_USES_LOG +#define BOOST_THREAD_USES_LOG_THREAD_ID +#define BOOST_THREAD_QUEUE_DEPRECATE_OLD + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +void p1() +{ + std::cout << BOOST_CONTEXTOF << std::endl; + boost::this_thread::sleep_for(boost::chrono::milliseconds(30)); + std::cout << BOOST_CONTEXTOF << std::endl; +} + +void p2() +{ + std::cout << BOOST_CONTEXTOF << std::endl; + boost::this_thread::sleep_for(boost::chrono::milliseconds(10)); + std::cout << BOOST_CONTEXTOF << std::endl; +} + +int f1() +{ + // std::cout << BOOST_CONTEXTOF << std::endl; + boost::this_thread::sleep_for(boost::chrono::seconds(1)); + return 1; +} +int f2(int i) +{ + // std::cout << BOOST_CONTEXTOF << std::endl; + boost::this_thread::sleep_for(boost::chrono::seconds(2)); + return i + 1; +} + +void submit_some(boost::generic_serial_executor_cont& tp) +{ + std::cout << BOOST_CONTEXTOF << std::endl; + for (int i = 0; i < 3; ++i) { + std::cout << BOOST_CONTEXTOF << std::endl; + tp.submit(&p2); + } + for (int i = 0; i < 3; ++i) { + std::cout << BOOST_CONTEXTOF << std::endl; + tp.submit(&p1); + } + std::cout << BOOST_CONTEXTOF << std::endl; + +} + + +void at_th_entry(boost::basic_thread_pool& ) +{ + +} + +int test_executor_adaptor() +{ + // std::cout << BOOST_CONTEXTOF << std::endl; + { + try + { + +#if ! defined(BOOST_NO_CXX11_RVALUE_REFERENCES) + // std::cout << BOOST_CONTEXTOF << std::endl; + { + boost::basic_thread_pool ea1(4); + boost::generic_serial_executor_cont ea2(ea1); + submit_some(ea2); + } +#endif + // std::cout << BOOST_CONTEXTOF << std::endl; + } + catch (std::exception& ex) + { + std::cout << "ERROR= " << ex.what() << "" << std::endl; + return 1; + } + catch (...) + { + std::cout << " ERROR= exception thrown" << std::endl; + return 2; + } + } + // std::cout << BOOST_CONTEXTOF << std::endl; + return 0; +} + + +int main() +{ + return test_executor_adaptor(); +} diff --git a/example/serial_executor.cpp b/example/serial_executor.cpp index 967c1ea7..f64ac4b9 100644 --- a/example/serial_executor.cpp +++ b/example/serial_executor.cpp @@ -51,8 +51,8 @@ int f2(int i) boost::this_thread::sleep_for(boost::chrono::seconds(2)); return i + 1; } - -void submit_some(boost::serial_executor& tp) +template +void submit_some(boost::serial_executor& tp) { std::cout << BOOST_CONTEXTOF << std::endl; for (int i = 0; i < 3; ++i) { @@ -84,7 +84,7 @@ int test_executor_adaptor() // std::cout << BOOST_CONTEXTOF << std::endl; { boost::basic_thread_pool ea1(4); - boost::serial_executor ea2(ea1); + boost::serial_executor ea2(ea1); submit_some(ea2); } #endif diff --git a/example/serial_executor_cont.cpp b/example/serial_executor_cont.cpp index 89842798..4c452e11 100644 --- a/example/serial_executor_cont.cpp +++ b/example/serial_executor_cont.cpp @@ -52,7 +52,8 @@ int f2(int i) return i + 1; } -void submit_some(boost::serial_executor_cont& tp) +template < class Executor> +void submit_some(boost::serial_executor_cont& tp) { std::cout << BOOST_CONTEXTOF << std::endl; for (int i = 0; i < 3; ++i) { @@ -84,7 +85,7 @@ int test_executor_adaptor() // std::cout << BOOST_CONTEXTOF << std::endl; { boost::basic_thread_pool ea1(4); - boost::serial_executor_cont ea2(ea1); + boost::serial_executor_cont ea2(ea1); submit_some(ea2); } #endif diff --git a/include/boost/thread/executors/generic_serial_executor.hpp b/include/boost/thread/executors/generic_serial_executor.hpp new file mode 100644 index 00000000..2d7d3cb7 --- /dev/null +++ b/include/boost/thread/executors/generic_serial_executor.hpp @@ -0,0 +1,271 @@ +// Copyright (C) 2013,2015 Vicente J. Botet Escriba +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// 2013/11 Vicente J. Botet Escriba +// first implementation of a simple serial scheduler. + +#ifndef BOOST_THREAD_GENERIC_SERIAL_EXECUTOR_HPP +#define BOOST_THREAD_GENERIC_SERIAL_EXECUTOR_HPP + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace boost +{ +namespace executors +{ + class generic_serial_executor + { + public: + /// type-erasure to store the works to do + typedef executors::work work; + private: + + struct shared_state { + typedef executors::work work; + typedef scoped_thread<> thread_t; + + /// the thread safe work queue + concurrent::sync_queue work_queue; + generic_executor ex; + thread_t thr; + + struct try_executing_one_task { + work& task; + boost::promise &p; + try_executing_one_task(work& task, boost::promise &p) + : task(task), p(p) {} + void operator()() { + try { + task(); + p.set_value(); + } catch (...) + { + p.set_exception(current_exception()); + } + } + }; + public: + /** + * \par Returns + * The underlying executor wrapped on a generic executor reference. + */ + generic_executor& underlying_executor() BOOST_NOEXCEPT { return ex; } + + private: + + /** + * The main loop of the worker thread + */ + void worker_thread() + { + try + { + for(;;) + { + work task; + queue_op_status st = work_queue.wait_pull(task); + if (st == queue_op_status::closed) return; + + boost::promise p; + try_executing_one_task tmp(task,p); + ex.submit(tmp); + p.get_future().wait(); + } + } + catch (...) + { + std::terminate(); + return; + } + } + + public: + /// shared_state is not copyable. + BOOST_THREAD_NO_COPYABLE(shared_state) + + /** + * \b Effects: creates a thread pool that runs closures using one of its closure-executing methods. + * + * \b Throws: Whatever exception is thrown while initializing the needed resources. + */ + template + shared_state(Executor& ex) + : ex(ex), thr(&shared_state::worker_thread, this) + { + } + /** + * \b Effects: Destroys the thread pool. + * + * \b Synchronization: The completion of all the closures happen before the completion of the \c shared_state destructor. + */ + ~shared_state() + { + // signal to the worker thread that there will be no more submissions. + close(); + } + + /** + * \b Effects: close the \c generic_serial_executor for submissions. + * The loop will work until there is no more closures to run. + */ + void close() + { + work_queue.close(); + } + + /** + * \b Returns: whether the pool is closed for submissions. + */ + bool closed() + { + return work_queue.closed(); + } + + /** + * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible. + * + * \b Effects: The specified \c closure will be scheduled for execution at some point in the future. + * If invoked closure throws an exception the \c generic_serial_executor will call \c std::terminate, as is the case with threads. + * + * \b Synchronization: completion of \c closure on a particular thread happens before destruction of thread's thread local variables. + * + * \b Throws: \c sync_queue_is_closed if the thread pool is closed. + * Whatever exception that can be throw while storing the closure. + */ + + #if defined(BOOST_NO_CXX11_RVALUE_REFERENCES) + template + void submit(Closure & closure) + { + work_queue.push(work(closure)); + } + #endif + void submit(void (*closure)()) + { + work_queue.push(work(closure)); + } + + template + void submit(BOOST_THREAD_RV_REF(Closure) closure) + { + work_queue.push(work(boost::forward(closure))); + } + + }; + + public: + + /** + * \b Effects: creates a thread pool that runs closures using one of its closure-executing methods. + * + * \b Throws: Whatever exception is thrown while initializing the needed resources. + */ + template + generic_serial_executor(Executor& ex) + : pimpl(make_shared(ex)) + { + } + /** + * \b Effects: Destroys the thread pool. + * + * \b Synchronization: The completion of all the closures happen before the completion of the \c generic_serial_executor destructor. + */ + ~generic_serial_executor() + { + } + + /** + * \par Returns + * The underlying executor wrapped on a generic executor reference. + */ + generic_executor& underlying_executor() BOOST_NOEXCEPT + { + return pimpl->underlying_executor(); + } + + /** + * \b Returns: always false as a serial executor can not re-enter. + * Remark: A serial executor can not execute one of its pending tasks as the tasks depends on the other tasks. + */ + bool try_executing_one() + { + return false; + } + + /** + * \b Effects: close the \c generic_serial_executor for submissions. + * The loop will work until there is no more closures to run. + */ + void close() + { + pimpl->close(); + } + + /** + * \b Returns: whether the pool is closed for submissions. + */ + bool closed() + { + return pimpl->closed(); + } + + /** + * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible. + * + * \b Effects: The specified \c closure will be scheduled for execution at some point in the future. + * If invoked closure throws an exception the \c generic_serial_executor will call \c std::terminate, as is the case with threads. + * + * \b Synchronization: completion of \c closure on a particular thread happens before destruction of thread's thread local variables. + * + * \b Throws: \c sync_queue_is_closed if the thread pool is closed. + * Whatever exception that can be throw while storing the closure. + */ + +#if defined(BOOST_NO_CXX11_RVALUE_REFERENCES) + template + void submit(Closure & closure) + { + pimpl->submit(closure); + } +#endif + void submit(void (*closure)()) + { + pimpl->submit(closure); + } + + template + void submit(BOOST_THREAD_RV_REF(Closure) closure) + { + pimpl->submit(boost::forward(closure)); + } + + /** + * \b Returns: always false as a serial executor can not re-enter. + * Remark: A serial executor can not execute one of its pending tasks as the tasks depends on the other tasks. + */ + template + bool reschedule_until(Pred const& pred) + { + return false; + } + private: + shared_ptr pimpl; + }; +} +using executors::generic_serial_executor; +} + +#include + +#endif diff --git a/include/boost/thread/executors/generic_serial_executor_cont.hpp b/include/boost/thread/executors/generic_serial_executor_cont.hpp new file mode 100644 index 00000000..d3ebd32b --- /dev/null +++ b/include/boost/thread/executors/generic_serial_executor_cont.hpp @@ -0,0 +1,257 @@ +// Copyright (C) 2015 Vicente J. Botet Escriba +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// 2013/11 Vicente J. Botet Escriba +// first implementation of a simple serial scheduler. + +#ifndef BOOST_THREAD_GENERIC_SERIAL_EXECUTOR_CONT_HPP +#define BOOST_THREAD_GENERIC_SERIAL_EXECUTOR_CONT_HPP + +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace boost +{ +namespace executors +{ + class generic_serial_executor_cont + { + public: + /// type-erasure to store the works to do + typedef executors::work work; + private: + + struct shared_state { + typedef executors::work work; + + generic_executor ex_; + future fut_; // protected by mtx_ + bool closed_; // protected by mtx_ + mutex mtx_; + + struct continuation { + work task; + template + struct result { + typedef void type; + }; + continuation(BOOST_THREAD_RV_REF(work) tsk) + : task(boost::move(tsk)) {} + void operator()(future f) + { + try { + task(); + } catch (...) { + std::terminate(); + } + } + }; + + bool closed(lock_guard&) const + { + return closed_; + } + public: + /** + * \par Returns + * The underlying executor wrapped on a generic executor reference. + */ + generic_executor& underlying_executor() BOOST_NOEXCEPT { return ex_; } + + /// shared_state is not copyable. + BOOST_THREAD_NO_COPYABLE(shared_state) + + /** + * \b Effects: creates a serial executor that runs closures in fifo order using one the associated executor. + * + * \b Throws: Whatever exception is thrown while initializing the needed resources. + * + * \b Notes: + * * The lifetime of the associated executor must outlive the serial executor. + * * The current implementation doesn't support submission from synchronous continuation, that is, + * - the executor must execute the continuation asynchronously or + * - the continuation can not submit to this serial executor. + */ + template + shared_state(Executor& ex) + : ex_(ex), fut_(make_ready_future()), closed_(false) + { + } + /** + * \b Effects: Destroys the thread pool. + * + * \b Synchronization: The completion of all the closures happen before the completion of the \c generic_serial_executor_cont destructor. + */ + ~shared_state() + { + // signal to the worker thread that there will be no more submissions. + close(); + } + + /** + * \b Effects: close the \c generic_serial_executor_cont for submissions. + * The loop will work until there is no more closures to run. + */ + void close() + { + lock_guard lk(mtx_); + closed_ = true;; + } + + /** + * \b Returns: whether the pool is closed for submissions. + */ + bool closed() + { + lock_guard lk(mtx_); + return closed(lk); + } + + /** + * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible. + * + * \b Effects: The specified \c closure will be scheduled for execution after the last submitted closure finish. + * If the invoked closure throws an exception the \c generic_serial_executor_cont will call \c std::terminate, as is the case with threads. + * + * \b Throws: \c sync_queue_is_closed if the executor is closed. + * Whatever exception that can be throw while storing the closure. + * + */ + +#if defined(BOOST_NO_CXX11_RVALUE_REFERENCES) + template + void submit(Closure & closure) + { + lock_guard lk(mtx_); + if (closed(lk)) BOOST_THROW_EXCEPTION( sync_queue_is_closed() ); + fut_ = fut_.then(ex_, continuation(work(closure))); + } +#endif + void submit(void (*closure)()) + { + lock_guard lk(mtx_); + if (closed(lk)) BOOST_THROW_EXCEPTION( sync_queue_is_closed() ); + fut_ = fut_.then(ex_, continuation(work(closure))); + } + + template + void submit(BOOST_THREAD_RV_REF(Closure) closure) + { + lock_guard lk(mtx_); + if (closed(lk)) BOOST_THROW_EXCEPTION( sync_queue_is_closed() ); + fut_ = fut_.then(ex_, continuation(work(boost::forward(closure)))); + } + }; + public: + /** + * \b Effects: creates a thread pool that runs closures using one of its closure-executing methods. + * + * \b Throws: Whatever exception is thrown while initializing the needed resources. + */ + template + generic_serial_executor_cont(Executor& ex) + : pimpl(make_shared(ex)) + { + } + /** + * \b Effects: Destroys the thread pool. + * + * \b Synchronization: The completion of all the closures happen before the completion of the \c serial_executor destructor. + */ + ~generic_serial_executor_cont() + { + } + + /** + * \par Returns + * The underlying executor wrapped on a generic executor reference. + */ + generic_executor& underlying_executor() BOOST_NOEXCEPT + { + return pimpl->underlying_executor(); + } + + /** + * \b Returns: always false as a serial executor can not re-enter. + * Remark: A serial executor can not execute one of its pending tasks as the tasks depends on the other tasks. + */ + bool try_executing_one() + { + return false; + } + + /** + * \b Effects: close the \c serial_executor for submissions. + * The loop will work until there is no more closures to run. + */ + void close() + { + pimpl->close(); + } + + /** + * \b Returns: whether the pool is closed for submissions. + */ + bool closed() + { + return pimpl->closed(); + } + + /** + * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible. + * + * \b Effects: The specified \c closure will be scheduled for execution at some point in the future. + * If invoked closure throws an exception the \c serial_executor will call \c std::terminate, as is the case with threads. + * + * \b Synchronization: completion of \c closure on a particular thread happens before destruction of thread's thread local variables. + * + * \b Throws: \c sync_queue_is_closed if the thread pool is closed. + * Whatever exception that can be throw while storing the closure. + */ + +#if defined(BOOST_NO_CXX11_RVALUE_REFERENCES) + template + void submit(Closure & closure) + { + pimpl->submit(closure); + } +#endif + void submit(void (*closure)()) + { + pimpl->submit(closure); + } + + template + void submit(BOOST_THREAD_RV_REF(Closure) closure) + { + pimpl->submit(boost::forward(closure)); + } + + /** + * \b Returns: always false as a serial executor can not re-enter. + * Remark: A serial executor can not execute one of its pending tasks as the tasks depends on the other tasks. + */ + template + bool reschedule_until(Pred const& pred) + { + return false; + } + private: + shared_ptr pimpl; + }; +} +using executors::generic_serial_executor_cont; +} + +#include + +#endif diff --git a/include/boost/thread/executors/serial_executor.hpp b/include/boost/thread/executors/serial_executor.hpp index c3651d6e..0021cae3 100644 --- a/include/boost/thread/executors/serial_executor.hpp +++ b/include/boost/thread/executors/serial_executor.hpp @@ -24,6 +24,7 @@ namespace boost { namespace executors { + template class serial_executor { public: @@ -37,7 +38,7 @@ namespace executors /// the thread safe work queue concurrent::sync_queue work_queue; - generic_executor ex; + Executor ex; thread_t thr; struct try_executing_one_task { @@ -60,7 +61,7 @@ namespace executors * \par Returns * The underlying executor wrapped on a generic executor reference. */ - generic_executor& underlying_executor() BOOST_NOEXCEPT { return ex; } + Executor& underlying_executor() BOOST_NOEXCEPT { return ex; } private: @@ -99,7 +100,6 @@ namespace executors * * \b Throws: Whatever exception is thrown while initializing the needed resources. */ - template shared_state(Executor& ex) : ex(ex), thr(&shared_state::worker_thread, this) { @@ -171,7 +171,6 @@ namespace executors * * \b Throws: Whatever exception is thrown while initializing the needed resources. */ - template serial_executor(Executor& ex) : pimpl(make_shared(ex)) { @@ -189,7 +188,7 @@ namespace executors * \par Returns * The underlying executor wrapped on a generic executor reference. */ - generic_executor& underlying_executor() BOOST_NOEXCEPT + Executor& underlying_executor() BOOST_NOEXCEPT { return pimpl->underlying_executor(); } @@ -257,7 +256,6 @@ namespace executors template bool reschedule_until(Pred const& pred) { - //return pimpl->reschedule_until(pred); return false; } private: diff --git a/include/boost/thread/executors/serial_executor_cont.hpp b/include/boost/thread/executors/serial_executor_cont.hpp index 3557a35b..f5a561ec 100644 --- a/include/boost/thread/executors/serial_executor_cont.hpp +++ b/include/boost/thread/executors/serial_executor_cont.hpp @@ -12,9 +12,7 @@ #include #include #include -#include #include -#include #include #include @@ -24,6 +22,7 @@ namespace boost { namespace executors { + template class serial_executor_cont { public: @@ -34,123 +33,122 @@ namespace executors struct shared_state { typedef executors::work work; - generic_executor ex_; - future fut_; // protected by mtx_ - bool closed_; // protected by mtx_ - mutex mtx_; + Executor ex_; + future fut_; // protected by mtx_ + bool closed_; // protected by mtx_ + mutex mtx_; - struct continuation { - work task; - template - struct result { - typedef void type; - }; - continuation(BOOST_THREAD_RV_REF(work) tsk) - : task(boost::move(tsk)) {} - void operator()(future f) - { - try { - task(); - } catch (...) { - std::terminate(); + struct continuation { + work task; + template + struct result { + typedef void type; + }; + continuation(BOOST_THREAD_RV_REF(work) tsk) + : task(boost::move(tsk)) {} + void operator()(future f) + { + try { + task(); + } catch (...) { + std::terminate(); + } } + }; + + bool closed(lock_guard&) const + { + return closed_; } - }; + public: + /** + * \par Returns + * The underlying executor wrapped on a generic executor reference. + */ + Executor& underlying_executor() BOOST_NOEXCEPT { return ex_; } - bool closed(lock_guard&) const - { - return closed_; - } - public: - /** - * \par Returns - * The underlying executor wrapped on a generic executor reference. - */ - generic_executor& underlying_executor() BOOST_NOEXCEPT { return ex_; } + /// shared_state is not copyable. + BOOST_THREAD_NO_COPYABLE(shared_state) - /// shared_state is not copyable. - BOOST_THREAD_NO_COPYABLE(shared_state) + /** + * \b Effects: creates a serial executor that runs closures in fifo order using one the associated executor. + * + * \b Throws: Whatever exception is thrown while initializing the needed resources. + * + * \b Notes: + * * The lifetime of the associated executor must outlive the serial executor. + * * The current implementation doesn't support submission from synchronous continuation, that is, + * - the executor must execute the continuation asynchronously or + * - the continuation can not submit to this serial executor. + */ + shared_state(Executor& ex) + : ex_(ex), fut_(make_ready_future()), closed_(false) + { + } + /** + * \b Effects: Destroys the thread pool. + * + * \b Synchronization: The completion of all the closures happen before the completion of the \c serial_executor_cont destructor. + */ + ~shared_state() + { + // signal to the worker thread that there will be no more submissions. + close(); + } - /** - * \b Effects: creates a serial executor that runs closures in fifo order using one the associated executor. - * - * \b Throws: Whatever exception is thrown while initializing the needed resources. - * - * \b Notes: - * * The lifetime of the associated executor must outlive the serial executor. - * * The current implementation doesn't support submission from synchronous continuation, that is, - * - the executor must execute the continuation asynchronously or - * - the continuation can not submit to this serial executor. - */ - template - shared_state(Executor& ex) - : ex_(ex), fut_(make_ready_future()), closed_(false) - { - } - /** - * \b Effects: Destroys the thread pool. - * - * \b Synchronization: The completion of all the closures happen before the completion of the \c serial_executor_cont destructor. - */ - ~shared_state() - { - // signal to the worker thread that there will be no more submissions. - close(); - } + /** + * \b Effects: close the \c serial_executor_cont for submissions. + * The loop will work until there is no more closures to run. + */ + void close() + { + lock_guard lk(mtx_); + closed_ = true;; + } - /** - * \b Effects: close the \c serial_executor_cont for submissions. - * The loop will work until there is no more closures to run. - */ - void close() - { - lock_guard lk(mtx_); - closed_ = true;; - } + /** + * \b Returns: whether the pool is closed for submissions. + */ + bool closed() + { + lock_guard lk(mtx_); + return closed(lk); + } - /** - * \b Returns: whether the pool is closed for submissions. - */ - bool closed() - { - lock_guard lk(mtx_); - return closed(lk); - } + /** + * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible. + * + * \b Effects: The specified \c closure will be scheduled for execution after the last submitted closure finish. + * If the invoked closure throws an exception the \c serial_executor_cont will call \c std::terminate, as is the case with threads. + * + * \b Throws: \c sync_queue_is_closed if the executor is closed. + * Whatever exception that can be throw while storing the closure. + * + */ - /** - * \b Requires: \c Closure is a model of \c Callable(void()) and a model of \c CopyConstructible/MoveConstructible. - * - * \b Effects: The specified \c closure will be scheduled for execution after the last submitted closure finish. - * If the invoked closure throws an exception the \c serial_executor_cont will call \c std::terminate, as is the case with threads. - * - * \b Throws: \c sync_queue_is_closed if the executor is closed. - * Whatever exception that can be throw while storing the closure. - * - */ + #if defined(BOOST_NO_CXX11_RVALUE_REFERENCES) + template + void submit(Closure & closure) + { + lock_guard lk(mtx_); + if (closed(lk)) BOOST_THROW_EXCEPTION( sync_queue_is_closed() ); + fut_ = fut_.then(ex_, continuation(work(closure))); + } + #endif + void submit(void (*closure)()) + { + lock_guard lk(mtx_); + if (closed(lk)) BOOST_THROW_EXCEPTION( sync_queue_is_closed() ); + fut_ = fut_.then(ex_, continuation(work(closure))); + } -#if defined(BOOST_NO_CXX11_RVALUE_REFERENCES) - template - void submit(Closure & closure) - { - lock_guard lk(mtx_); - if (closed(lk)) BOOST_THROW_EXCEPTION( sync_queue_is_closed() ); - fut_ = fut_.then(ex_, continuation(work(closure))); - } -#endif - void submit(void (*closure)()) - { - lock_guard lk(mtx_); - if (closed(lk)) BOOST_THROW_EXCEPTION( sync_queue_is_closed() ); - fut_ = fut_.then(ex_, continuation(work(closure))); - } - - template - void submit(BOOST_THREAD_RV_REF(Closure) closure) - { - lock_guard lk(mtx_); - if (closed(lk)) BOOST_THROW_EXCEPTION( sync_queue_is_closed() ); - fut_ = fut_.then(ex_, continuation(work(boost::forward(closure)))); - } + template + void submit(BOOST_THREAD_RV_REF(Closure) closure) + { + lock_guard lk(mtx_); + if (closed(lk)) BOOST_THROW_EXCEPTION( sync_queue_is_closed() ); + fut_ = fut_.then(ex_, continuation(work(boost::forward(closure)))); + } }; public: /** @@ -158,7 +156,6 @@ namespace executors * * \b Throws: Whatever exception is thrown while initializing the needed resources. */ - template serial_executor_cont(Executor& ex) : pimpl(make_shared(ex)) { @@ -176,7 +173,7 @@ namespace executors * \par Returns * The underlying executor wrapped on a generic executor reference. */ - generic_executor& underlying_executor() BOOST_NOEXCEPT + Executor& underlying_executor() BOOST_NOEXCEPT { return pimpl->underlying_executor(); } diff --git a/test/Jamfile.v2 b/test/Jamfile.v2 index e7ab5a57..065f3a8c 100644 --- a/test/Jamfile.v2 +++ b/test/Jamfile.v2 @@ -809,7 +809,9 @@ rule thread-compile ( sources : reqs * : name ) [ thread-run2 ../example/executor.cpp : ex_executor ] [ thread-run2 ../example/generic_executor_ref.cpp : ex_generic_executor_ref ] [ thread-run2 ../example/generic_executor.cpp : ex_generic_executor ] + [ thread-run2 ../example/generic_serial_executor.cpp : ex_generic_serial_executor ] [ thread-run2 ../example/serial_executor.cpp : ex_serial_executor ] + [ thread-run2 ../example/generic_serial_executor_cont.cpp : ex_generic_serial_executor_cont ] [ thread-run2 ../example/serial_executor_cont.cpp : ex_serial_executor_cont ] [ thread-run2 ../example/future_when_all.cpp : ex_future_when_all ] [ thread-run2 ../example/parallel_accumulate.cpp : ex_parallel_accumulate ]