|
|
Boost-Commit : |
Subject: [Boost-commit] svn:boost r58645 - in sandbox/task/boost/task: . detail spin
From: oliver.kowalke_at_[hidden]
Date: 2010-01-02 14:18:26
Author: olli
Date: 2010-01-02 14:18:25 EST (Sat, 02 Jan 2010)
New Revision: 58645
URL: http://svn.boost.org/trac/boost/changeset/58645
Log:
- support of promise<> and spin::promise<> in task<>
Text files modified:
sandbox/task/boost/task/as_sub_task.hpp | 7
sandbox/task/boost/task/detail/pool_base.hpp | 74 +++++++--
sandbox/task/boost/task/handle.hpp | 203 +++++++++++++++++++---------
sandbox/task/boost/task/new_thread.hpp | 63 ++++----
sandbox/task/boost/task/own_thread.hpp | 33 +++
sandbox/task/boost/task/spin/future.hpp | 20 +-
sandbox/task/boost/task/task.hpp | 286 +++++++++++++++++++++++++++++++--------
7 files changed, 490 insertions(+), 196 deletions(-)
Modified: sandbox/task/boost/task/as_sub_task.hpp
==============================================================================
--- sandbox/task/boost/task/as_sub_task.hpp (original)
+++ sandbox/task/boost/task/as_sub_task.hpp 2010-01-02 14:18:25 EST (Sat, 02 Jan 2010)
@@ -10,7 +10,7 @@
#include <boost/bind.hpp>
#include <boost/config.hpp>
#include <boost/move/move.hpp>
-#include <boost/weak_ptr.hpp>
+#include <boost/thread/future.hpp>
#include <boost/task/callable.hpp>
#include <boost/task/context.hpp>
@@ -33,8 +33,9 @@
detail::worker * w( detail::worker::tss_get() );
if ( w)
{
- spin::shared_future< R > f(
- t.get_future() );
+ spin::promise< R > prom;
+ spin::shared_future< R > f( prom.get_future() );
+ t.set_promise( boost::move( prom) );
context ctx;
handle< R > h( f, ctx);
w->put( callable( t, ctx) );
Modified: sandbox/task/boost/task/detail/pool_base.hpp
==============================================================================
--- sandbox/task/boost/task/detail/pool_base.hpp (original)
+++ sandbox/task/boost/task/detail/pool_base.hpp 2010-01-02 14:18:25 EST (Sat, 02 Jan 2010)
@@ -17,22 +17,23 @@
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/thread/detail/move.hpp>
-
-#include <boost/task/detail/bind_processor.hpp>
-#include <boost/task/detail/worker.hpp>
-#include <boost/task/detail/worker_group.hpp>
+#include <boost/thread/future.hpp>
#include <boost/task/callable.hpp>
#include <boost/task/context.hpp>
+#include <boost/task/detail/bind_processor.hpp>
+#include <boost/task/detail/worker_group.hpp>
+#include <boost/task/detail/worker.hpp>
#include <boost/task/exceptions.hpp>
-#include <boost/thread/future.hpp>
#include <boost/task/handle.hpp>
#include <boost/task/poolsize.hpp>
#include <boost/task/scanns.hpp>
#include <boost/task/spin/future.hpp>
#include <boost/task/stacksize.hpp>
#include <boost/task/task.hpp>
+#include <boost/task/utility.hpp>
#include <boost/task/watermark.hpp>
+#include <boost/thread/future.hpp>
#include <boost/config/abi_prefix.hpp>
@@ -289,11 +290,28 @@
if ( deactivated_() )
throw task_rejected("pool is closed");
- spin::shared_future< R > f( t.get_future() );
- context ctx;
- handle< R > h( f, ctx);
- queue_.put( callable( boost::move( t), ctx) );
- return h;
+ if ( this_task::runs_in_pool() )
+ {
+ spin::promise< R > prom;
+ spin::shared_future< R > f( prom.get_future() );
+ t.set_promise( boost::move( prom) );
+ context ctx;
+ handle< R > h( f, ctx);
+ queue_.put( callable( boost::move( t), ctx) );
+ return h;
+ }
+ else
+ {
+ promise< R > prom;
+ shared_future< R > f( prom.get_future() );
+ // TODO: if boost.thread uses boost.move
+ // use boost::move()
+ t.set_promise( prom);
+ context ctx;
+ handle< R > h( f, ctx);
+ queue_.put( callable( boost::move( t), ctx) );
+ return h;
+ }
}
template< typename R, typename Attr >
@@ -302,14 +320,34 @@
if ( deactivated_() )
throw task_rejected("pool is closed");
- spin::shared_future< R > f( t.get_future() );
- context ctx;
- handle< R > h( f, ctx);
- queue_.put(
- value_type(
- callable( boost::move( t), ctx),
- attr) );
- return h;
+ if ( this_task::runs_in_pool() )
+ {
+ spin::promise< R > prom;
+ spin::shared_future< R > f( prom.get_future() );
+ t.set_promise( boost::move( prom) );
+ context ctx;
+ handle< R > h( f, ctx);
+ queue_.put(
+ value_type(
+ callable( boost::move( t), ctx),
+ attr) );
+ return h;
+ }
+ else
+ {
+ promise< R > prom;
+ shared_future< R > f( prom.get_future() );
+ // TODO: if boost.thread uses boost.move
+ // use boost::move()
+ t.set_promise( prom);
+ context ctx;
+ handle< R > h( f, ctx);
+ queue_.put(
+ value_type(
+ callable( boost::move( t), ctx),
+ attr) );
+ return h;
+ }
}
inline friend void intrusive_ptr_add_ref( pool_base * p)
Modified: sandbox/task/boost/task/handle.hpp
==============================================================================
--- sandbox/task/boost/task/handle.hpp (original)
+++ sandbox/task/boost/task/handle.hpp 2010-01-02 14:18:25 EST (Sat, 02 Jan 2010)
@@ -14,8 +14,8 @@
#include <boost/utility/enable_if.hpp>
#include <boost/task/context.hpp>
-#include <boost/task/exceptions.hpp>
#include <boost/task/spin/future.hpp>
+#include <boost/task/exceptions.hpp>
#include <boost/config/abi_prefix.hpp>
@@ -48,100 +48,169 @@
template< typename T1, typename T2, typename T3, typename T4, typename T5 >
friend unsigned int waitfor_any( handle< T1 > &, handle< T2 > &, handle< T3 > &, handle< T4 > &, handle< T5 > &);
- spin::shared_future< R > fut_;
- context ctx_;
+ struct impl
+ {
+ virtual ~impl() {}
+
+ virtual void interrupt() = 0;
+
+ virtual void interrupt_and_wait() = 0;
+
+ virtual bool interrupt_and_wait_until( system_time const& abs_time) = 0;
+
+ virtual bool interruption_requested() = 0;
+
+ virtual R get() = 0;
+
+ virtual bool is_ready() const = 0;
+
+ virtual bool has_value() const = 0;
+
+ virtual bool has_exception() const = 0;
+
+ virtual void wait() const = 0;
+
+ virtual bool wait_until( system_time const& abs_time) const = 0;
+ };
+
+ template< typename F >
+ class wrapper : public impl
+ {
+ private:
+ F fut_;
+ context ctx_;
+
+ public:
+ wrapper() :
+ fut_(),
+ ctx_()
+ {}
+
+ wrapper(
+ F const& fut,
+ context const& ctx) :
+ fut_( fut),
+ ctx_( ctx)
+ {}
+
+ void interrupt()
+ { ctx_.interrupt(); }
+
+ void interrupt_and_wait()
+ {
+ interrupt();
+ wait();
+ }
+
+ bool interrupt_and_wait_until( system_time const& abs_time)
+ {
+ interrupt();
+ return wait_until( abs_time);
+ }
+
+ bool interruption_requested()
+ { return ctx_.interruption_requested(); }
+
+ R get()
+ {
+ try
+ { return fut_.get(); }
+ catch ( future_uninitialized const&)
+ { throw task_uninitialized(); }
+ catch ( broken_promise const&)
+ { throw broken_task(); }
+ catch ( promise_already_satisfied const&)
+ { throw task_already_executed(); }
+ }
+
+ bool is_ready() const
+ { return fut_.is_ready(); }
+
+ bool has_value() const
+ { return fut_.has_value(); }
+
+ bool has_exception() const
+ { return fut_.has_exception(); }
+
+ void wait() const
+ {
+ try
+ { fut_.wait(); }
+ catch ( future_uninitialized const&)
+ { throw task_uninitialized(); }
+ catch ( broken_promise const&)
+ { throw broken_task(); }
+ catch ( thread_interrupted const&)
+ { throw task_interrupted(); }
+ }
+
+ bool wait_until( system_time const& abs_time) const
+ {
+ try
+ { return fut_.timed_wait_until( abs_time); }
+ catch ( future_uninitialized const&)
+ { throw task_uninitialized(); }
+ catch ( broken_promise const&)
+ { throw broken_task(); }
+ catch ( thread_interrupted const&)
+ { throw task_interrupted(); }
+ }
+ };
+
+ shared_ptr< impl > impl_;
public:
handle() :
- fut_(), ctx_()
+ impl_( new wrapper< spin::shared_future< R > >() )
{}
+ template< typename F >
handle(
- spin::shared_future< R > const& fut,
+ F const& fut,
context const& ctx) :
- fut_( fut),
- ctx_( ctx)
+ impl_( new wrapper< F >( fut, ctx) )
{}
void interrupt()
- { ctx_.interrupt(); }
+ { impl_->interrupt(); }
void interrupt_and_wait()
- {
- interrupt();
- wait();
- }
+ { impl_->interrupt_and_wait(); }
bool interrupt_and_wait_until( system_time const& abs_time)
- {
- interrupt();
- return wait_until( abs_time);
- }
+ { return impl_->interrupt_and_wait_until( abs_time); }
template< typename TimeDuration >
bool interrupt_and_wait_for( TimeDuration const& rel_time)
- {
- interrupt();
- return wait_for( rel_time);
- }
+ { return interrupt_and_wait_until( get_system_time() + rel_time); }
bool interruption_requested()
- { return ctx_.interruption_requested(); }
+ { return impl_->interruption_requested(); }
R get()
- {
- try
- { return fut_.get(); }
- catch ( future_uninitialized const&)
- { throw task_uninitialized(); }
- catch ( broken_promise const&)
- { throw broken_task(); }
- catch ( promise_already_satisfied const&)
- { throw task_already_executed(); }
- }
+ { return impl_->get(); }
bool is_ready() const
- { return fut_.is_ready(); }
+ { return impl_->is_ready(); }
bool has_value() const
- { return fut_.has_value(); }
+ { return impl_->has_value(); }
bool has_exception() const
- { return fut_.has_exception(); }
+ { return impl_->has_exception(); }
void wait() const
- {
- try
- { fut_.wait(); }
- catch ( future_uninitialized const&)
- { throw task_uninitialized(); }
- catch ( broken_promise const&)
- { throw broken_task(); }
- catch ( thread_interrupted const&)
- { throw task_interrupted(); }
- }
+ { impl_->wait(); }
bool wait_until( system_time const& abs_time) const
- {
- try
- { return fut_.timed_wait_until( abs_time); }
- catch ( future_uninitialized const&)
- { throw task_uninitialized(); }
- catch ( broken_promise const&)
- { throw broken_task(); }
- catch ( thread_interrupted const&)
- { throw task_interrupted(); }
- }
+ { return impl_->wait_until( abs_time); }
template< typename TimeDuration >
bool wait_for( TimeDuration const& rel_time) const
{ return wait_until( get_system_time() + rel_time); }
void swap( handle< R > & other)
- {
- fut_.swap( other.fut_);
- ctx_.swap( other.ctx_);
- }
+ { impl_.swap( other.impl_); }
};
template< typename T >
@@ -170,7 +239,7 @@
void waitfor_all( handle< T1 > & t1, handle< T2 > & t2)
{
try
- { spin::wait_for_all( t1.fut_, t2.fut_); }
+ { wait_for_all( * t1.impl_->fut_, * t2.impl_->fut_); }
catch ( thread_interrupted const&)
{ throw task_interrupted(); }
}
@@ -179,7 +248,7 @@
void waitfor_all( handle< T1 > & t1, handle< T2 > & t2, handle< T3 > & t3)
{
try
- { spin::wait_for_all( t1.fut_, t2.fut_, t3.fut_); }
+ { wait_for_all( * t1.impl_->fut_, * t2.impl_->fut_, * t3.impl_->fut_); }
catch ( thread_interrupted const&)
{ throw task_interrupted(); }
}
@@ -188,7 +257,7 @@
void waitfor_all( handle< T1 > & t1, handle< T2 > & t2, handle< T3 > & t3, handle< T4 > & t4)
{
try
- { spin::wait_for_all( t1.fut_, t2.fut_, t3.fut_, t4.fut_); }
+ { wait_for_all( * t1.impl_->fut_, * t2.impl_->fut_, * t3.impl_->fut_, * t4.impl_->fut_); }
catch ( thread_interrupted const&)
{ throw task_interrupted(); }
}
@@ -197,7 +266,7 @@
void waitfor_all( handle< T1 > & t1, handle< T2 > & t2, handle< T3 > & t3, handle< T4 > & t4, handle< T5 > & t5)
{
try
- { spin::wait_for_all( t1.fut_, t2.fut_, t3.fut_, t4.fut_, t5.fut_); }
+ { wait_for_all( * t1.impl_->fut_, * t2.impl_->fut_, * t3.impl_->fut_, * t4.impl_->fut_, * t5.impl_->fut_); }
catch ( thread_interrupted const&)
{ throw task_interrupted(); }
}
@@ -209,7 +278,7 @@
{
boost::detail::future_waiter waiter;
for ( Iterator i = begin; i != end; ++i)
- waiter.add( * ( i->fut_) );
+ waiter.add( * ( i.impl_->fut_) );
return next( begin, waiter.wait() );
}
catch ( thread_interrupted const&)
@@ -220,7 +289,7 @@
typename enable_if< is_handle_type< H1 >, unsigned >::type waitfor_any( H1 & h1, H2 & h2)
{
try
- { return spin::wait_for_any( h1.fut_, h2.fut_ ); }
+ { return wait_for_any( * h1.impl_->fut_, * h2.impl_->fut_ ); }
catch ( thread_interrupted const&)
{ throw task_interrupted(); }
}
@@ -229,7 +298,7 @@
unsigned int waitfor_any( handle< T1 > & t1, handle< T2 > & t2, handle< T3 > & t3)
{
try
- { return spin::wait_for_any( t1.fut_, t2.fut_, t3.fut_ ); }
+ { return wait_for_any( * t1.impl_->fut_, * t2.impl_->fut_, * t3.impl_->fut_ ); }
catch ( thread_interrupted const&)
{ throw task_interrupted(); }
}
@@ -238,7 +307,7 @@
unsigned int waitfor_any( handle< T1 > & t1, handle< T2 > & t2, handle< T3 > & t3, handle< T4 > & t4)
{
try
- { return spin::wait_for_any( t1.fut_, t2.fut_, t3.fut_, t4.fut_ ); }
+ { return wait_for_any( * t1.impl_->fut_, * t2.impl_->fut_, * t3.impl_->fut_, * t4.impl_->fut_ ); }
catch ( thread_interrupted const&)
{ throw task_interrupted(); }
}
@@ -247,7 +316,7 @@
unsigned int waitfor_any( handle< T1 > & t1, handle< T2 > & t2, handle< T3 > & t3, handle< T4 > & t4, handle< T5 > & t5)
{
try
- { return spin::wait_for_any( t1.fut_, t2.fut_, t3.fut_, t4.fut_, t5.fut_ ); }
+ { return wait_for_any( * t1.impl_->fut_, * t2.impl_->fut_, * t3.impl_->fut_, * t4.impl_->fut_, * t5.impl_->fut_ ); }
catch ( thread_interrupted const&)
{ throw task_interrupted(); }
}
Modified: sandbox/task/boost/task/new_thread.hpp
==============================================================================
--- sandbox/task/boost/task/new_thread.hpp (original)
+++ sandbox/task/boost/task/new_thread.hpp 2010-01-02 14:18:25 EST (Sat, 02 Jan 2010)
@@ -12,12 +12,14 @@
#include <boost/move/move.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
+#include <boost/thread/future.hpp>
#include <boost/task/callable.hpp>
#include <boost/task/context.hpp>
#include <boost/task/handle.hpp>
#include <boost/task/spin/future.hpp>
#include <boost/task/task.hpp>
+#include <boost/task/utility.hpp>
#include <boost/config/abi_prefix.hpp>
@@ -43,40 +45,41 @@
}
-class new_thread
+struct new_thread
{
-private:
- struct wrapper
- {
- callable ca;
-
- wrapper( callable const& ca_) :
- ca( ca_)
- {}
-
- void operator()()
- {
- ca();
- ca.clear();
- }
- };
-
-public:
template< typename R >
handle< R > operator()( BOOST_RV_REF( task< R >) t)
{
- spin::shared_future< R > f(
- t.get_future() );
- context ctx1, ctx2;
- handle< R > h( f, ctx1);
- callable ca( t, ctx2);
- shared_ptr< thread > thrd(
- //new thread( wrapper( ca) ),
- new thread( ca),
- detail::joiner() );
- ctx1.reset( thrd);
-
- return h;
+ if ( this_task::runs_in_pool() )
+ {
+ spin::promise< R > prom;
+ spin::shared_future< R > f( prom.get_future() );
+ t.set_promise( boost::move( prom) );
+ context ctx1, ctx2;
+ handle< R > h( f, ctx1);
+ callable ca( t, ctx2);
+ shared_ptr< thread > thrd(
+ new thread( ca),
+ detail::joiner() );
+ ctx1.reset( thrd);
+ return h;
+ }
+ else
+ {
+ promise< R > prom;
+ shared_future< R > f( prom.get_future() );
+ // TODO: if boost.thread uses boost.move
+ // use boost::move()
+ t.set_promise( prom);
+ context ctx1, ctx2;
+ handle< R > h( f, ctx1);
+ callable ca( t, ctx2);
+ shared_ptr< thread > thrd(
+ new thread( ca),
+ detail::joiner() );
+ ctx1.reset( thrd);
+ return h;
+ }
}
};
Modified: sandbox/task/boost/task/own_thread.hpp
==============================================================================
--- sandbox/task/boost/task/own_thread.hpp (original)
+++ sandbox/task/boost/task/own_thread.hpp 2010-01-02 14:18:25 EST (Sat, 02 Jan 2010)
@@ -9,12 +9,14 @@
#include <boost/config.hpp>
#include <boost/move/move.hpp>
+#include <boost/thread/future.hpp>
#include <boost/task/callable.hpp>
#include <boost/task/context.hpp>
#include <boost/task/handle.hpp>
#include <boost/task/spin/future.hpp>
#include <boost/task/task.hpp>
+#include <boost/task/utility.hpp>
#include <boost/config/abi_prefix.hpp>
@@ -26,13 +28,30 @@
template< typename R >
handle< R > operator()( BOOST_RV_REF( task< R >) t)
{
- spin::shared_future< R > f(
- t.get_future() );
- context ctx;
- handle< R > h( f, ctx);
- callable ca( t, ctx);
- ca();
- return h;
+ if ( this_task::runs_in_pool() )
+ {
+ spin::promise< R > prom;
+ spin::shared_future< R > f( prom.get_future() );
+ t.set_promise( boost::move( prom) );
+ context ctx;
+ handle< R > h( f, ctx);
+ callable ca( t, ctx);
+ ca();
+ return h;
+ }
+ else
+ {
+ promise< R > prom;
+ shared_future< R > f( prom.get_future() );
+ // TODO: if boost.thread uses boost.move
+ // use boost::move()
+ t.set_promise( prom);
+ context ctx;
+ handle< R > h( f, ctx);
+ callable ca( t, ctx);
+ ca();
+ return h;
+ }
}
};
Modified: sandbox/task/boost/task/spin/future.hpp
==============================================================================
--- sandbox/task/boost/task/spin/future.hpp (original)
+++ sandbox/task/boost/task/spin/future.hpp 2010-01-02 14:18:25 EST (Sat, 02 Jan 2010)
@@ -171,10 +171,10 @@
template<typename T>
struct future_object: future_object_base
{
- typedef typename fibers::detail::future_traits<T>::storage_type storage_type;
- typedef typename fibers::detail::future_traits<T>::source_reference_type source_reference_type;
- typedef typename fibers::detail::future_traits<T>::rvalue_source_type rvalue_source_type;
- typedef typename fibers::detail::future_traits<T>::move_dest_type move_dest_type;
+ typedef typename tasks::detail::future_traits<T>::storage_type storage_type;
+ typedef typename tasks::detail::future_traits<T>::source_reference_type source_reference_type;
+ typedef typename tasks::detail::future_traits<T>::rvalue_source_type rvalue_source_type;
+ typedef typename tasks::detail::future_traits<T>::move_dest_type move_dest_type;
storage_type result;
@@ -184,12 +184,12 @@
void mark_finished_with_result_internal(source_reference_type result_)
{
- fibers::detail::future_traits<T>::init(result,result_);
+ tasks::detail::future_traits<T>::init(result,result_);
mark_finished_internal();
}
void mark_finished_with_result_internal(rvalue_source_type result_)
{
- fibers::detail::future_traits<T>::init(result,static_cast<rvalue_source_type>(result_));
+ tasks::detail::future_traits<T>::init(result,static_cast<rvalue_source_type>(result_));
mark_finished_internal();
}
@@ -499,7 +499,7 @@
friend class packaged_task<R>;
friend class detail::future_waiter;
- typedef typename fibers::detail::future_traits<R>::move_dest_type move_dest_type;
+ typedef typename tasks::detail::future_traits<R>::move_dest_type move_dest_type;
unique_future(future_ptr future_):
future(future_)
@@ -790,7 +790,7 @@
return unique_future<R>(future);
}
- void set_value(typename fibers::detail::future_traits<R>::source_reference_type r)
+ void set_value(typename tasks::detail::future_traits<R>::source_reference_type r)
{
lazy_init();
boost::lock_guard<mutex> lock(future->mtx);
@@ -802,7 +802,7 @@
}
// void set_value(R && r);
- void set_value(typename fibers::detail::future_traits<R>::rvalue_source_type r)
+ void set_value(typename tasks::detail::future_traits<R>::rvalue_source_type r)
{
lazy_init();
boost::lock_guard<mutex> lock(future->mtx);
@@ -810,7 +810,7 @@
{
throw promise_already_satisfied();
}
- future->mark_finished_with_result_internal(static_cast<typename fibers::detail::future_traits<R>::rvalue_source_type>(r));
+ future->mark_finished_with_result_internal(static_cast<typename tasks::detail::future_traits<R>::rvalue_source_type>(r));
}
void set_exception(boost::exception_ptr p)
Modified: sandbox/task/boost/task/task.hpp
==============================================================================
--- sandbox/task/boost/task/task.hpp (original)
+++ sandbox/task/boost/task/task.hpp 2010-01-02 14:18:25 EST (Sat, 02 Jan 2010)
@@ -10,9 +10,12 @@
#include <boost/atomic.hpp>
#include <boost/bind.hpp>
#include <boost/config.hpp>
-#include <boost/preprocessor/repetition.hpp>
+#include <boost/exception_ptr.hpp>
+#include <boost/intrusive_ptr.hpp>
#include <boost/move/move.hpp>
+#include <boost/preprocessor/repetition.hpp>
#include <boost/thread/future.hpp>
+#include <boost/thread/detail/move.hpp>
#include <boost/utility/enable_if.hpp>
#include <boost/utility/result_of.hpp>
@@ -26,6 +29,161 @@
namespace detail {
template< typename R >
+struct promise_adaptor
+{
+ template< typename X >
+ friend void intrusive_ptr_add_ref( promise_adaptor< X > *);
+ template< typename X >
+ friend void intrusive_ptr_release( promise_adaptor< X > *);
+
+ typedef intrusive_ptr< promise_adaptor > ptr;
+
+ atomic< unsigned int > use_count;
+
+ virtual ~promise_adaptor() {}
+
+ virtual void set_value( typename tasks::detail::future_traits< R >::source_reference_type) = 0;
+
+ virtual void set_value( typename tasks::detail::future_traits< R >::rvalue_source_type) = 0;
+
+ virtual void set_exception( exception_ptr) = 0;
+};
+
+template<>
+struct promise_adaptor< void >
+{
+ template< typename X >
+ friend void intrusive_ptr_add_ref( promise_adaptor< X > *);
+ template< typename X >
+ friend void intrusive_ptr_release( promise_adaptor< X > *);
+
+ typedef intrusive_ptr< promise_adaptor > ptr;
+
+ atomic< unsigned int > use_count;
+
+ virtual ~promise_adaptor() {}
+
+ virtual void set_value() = 0;
+
+ virtual void set_exception( exception_ptr) = 0;
+};
+
+template< typename R >
+void intrusive_ptr_add_ref( promise_adaptor< R > * p)
+{ p->use_count.fetch_add( 1, memory_order_relaxed); }
+
+template< typename R >
+void intrusive_ptr_release( promise_adaptor< R > * p)
+{
+ if ( p->use_count.fetch_sub( 1, memory_order_release) == 1)
+ {
+ atomic_thread_fence( memory_order_acquire);
+ delete p;
+ }
+}
+
+template< typename R, template< typename > class Promise >
+class promise_wrapper;
+
+template< typename R >
+class promise_wrapper< R, promise > : public promise_adaptor< R >
+{
+private:
+ promise< R > prom_;
+
+public:
+#ifdef BOOST_HAS_RVALUE_REFS
+ promise_wrapper( promise< R > && prom) :
+ prom_( prom)
+ {}
+#else
+ promise_wrapper( boost::detail::thread_move_t< promise< R > > prom) :
+ prom_( prom)
+ {}
+#endif
+
+ void set_value( typename tasks::detail::future_traits< R >::source_reference_type r)
+ { prom_.set_value( r); };
+
+ void set_value( typename tasks::detail::future_traits< R >::rvalue_source_type r)
+ { prom_.set_value( r); };
+
+ void set_exception( exception_ptr p)
+ { prom_.set_exception( p); }
+};
+
+template<>
+class promise_wrapper< void, promise > : public promise_adaptor< void >
+{
+private:
+ promise< void > prom_;
+
+public:
+#ifdef BOOST_HAS_RVALUE_REFS
+ promise_wrapper( promise< void > && prom) :
+ prom_( prom)
+ {}
+#else
+ promise_wrapper( boost::detail::thread_move_t< promise< void > > prom) :
+ prom_( prom)
+ {}
+#endif
+
+ void set_value()
+ { prom_.set_value(); };
+
+ void set_exception( exception_ptr p)
+ { prom_.set_exception( p); }
+};
+
+template< typename R >
+class promise_wrapper< R, spin::promise > : public promise_adaptor< R >
+{
+private:
+ spin::promise< R > prom_;
+
+public:
+ promise_wrapper() :
+ prom_()
+ {}
+
+ promise_wrapper( BOOST_RV_REF( spin::promise< R >) prom) :
+ prom_( prom)
+ {}
+
+ void set_value( typename tasks::detail::future_traits< R >::source_reference_type r)
+ { prom_.set_value( r); };
+
+ void set_value( typename tasks::detail::future_traits< R >::rvalue_source_type r)
+ { prom_.set_value( r); };
+
+ void set_exception( exception_ptr p)
+ { prom_.set_exception( p); }
+};
+
+template<>
+class promise_wrapper< void, spin::promise > : public promise_adaptor< void >
+{
+private:
+ spin::promise< void > prom_;
+
+public:
+ promise_wrapper() :
+ prom_()
+ {}
+
+ promise_wrapper( BOOST_RV_REF( spin::promise< void >) prom) :
+ prom_( prom)
+ {}
+
+ void set_value()
+ { prom_.set_value(); };
+
+ void set_exception( exception_ptr p)
+ { prom_.set_exception( p); }
+};
+
+template< typename R >
class task_base
{
private:
@@ -37,14 +195,15 @@
atomic< unsigned int > use_count_;
protected:
- bool done_;
- spin::promise< R > prom_;
+ bool done_;
+ typename promise_adaptor< R >::ptr prom_;
virtual void do_run() = 0;
public:
task_base() :
- done_( false), prom_()
+ done_( false),
+ prom_( new promise_wrapper< R, spin::promise >() )
{}
virtual ~task_base() {}
@@ -56,12 +215,8 @@
done_ = true;
}
- spin::unique_future< R > get_future()
- { return prom_.get_future(); }
-
- template< typename Cb >
- void set_wait_callback( Cb const& cb)
- { prom_.set_wait_callback( cb); }
+ void set_promise( typename promise_adaptor< R >::ptr prom)
+ { prom_ = prom; }
};
template< typename R >
@@ -89,45 +244,45 @@
void do_run()
{
try
- { this->prom_.set_value( fn_() ); }
+ { this->prom_->set_value( fn_() ); }
catch ( promise_already_satisfied const&)
{ throw task_already_executed(); }
catch ( thread_interrupted const&)
- { this->prom_.set_exception( copy_exception( task_interrupted() ) ); }
+ { this->prom_->set_exception( copy_exception( task_interrupted() ) ); }
catch ( task_interrupted const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( boost::exception const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::ios_base::failure const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::domain_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::invalid_argument const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::length_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::out_of_range const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::logic_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::overflow_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::range_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::underflow_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::runtime_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::bad_alloc const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::bad_cast const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::bad_typeid const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::bad_exception const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch (...)
- { this->prom_.set_exception( current_exception() ); }
+ { this->prom_->set_exception( current_exception() ); }
}
public:
@@ -149,46 +304,46 @@
try
{
fn_();
- this->prom_.set_value();
+ this->prom_->set_value();
}
catch ( promise_already_satisfied const&)
{ throw task_already_executed(); }
catch ( thread_interrupted const&)
- { this->prom_.set_exception( copy_exception( task_interrupted() ) ); }
+ { this->prom_->set_exception( copy_exception( task_interrupted() ) ); }
catch ( task_interrupted const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( boost::exception const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::ios_base::failure const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::domain_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::invalid_argument const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::length_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::out_of_range const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::logic_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::overflow_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::range_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::underflow_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::runtime_error const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::bad_alloc const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::bad_cast const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::bad_typeid const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch ( std::bad_exception const& e)
- { this->prom_.set_exception( copy_exception( e) ); }
+ { this->prom_->set_exception( copy_exception( e) ); }
catch (...)
- { this->prom_.set_exception( current_exception() ); }
+ { this->prom_->set_exception( current_exception() ); }
}
public:
@@ -243,8 +398,8 @@
# define BOOST_ENUM_TASK_ARGS(n) BOOST_PP_ENUM(n, BOOST_TASKS_ARG, ~)
# define BOOST_TASKS_CTOR(z, n, unused) \
-template< \
- typename Fn, \
+template<> \
+ typename Fn, \
BOOST_PP_ENUM_PARAMS(n, typename A) \
> \
explicit task( Fn fn, BOOST_ENUM_TASK_ARGS(n)) \
@@ -259,24 +414,33 @@
void operator()()
{
- if ( ! task_)
- throw task_moved();
+ if ( ! task_) throw task_moved();
task_->run();
}
- spin::unique_future< R > get_future()
+// TODO: if boost.thread uses boost.move
+// re-work set_promise
+#ifdef BOOST_HAS_RVALUE_REFS
+ void set_promise( promise< R > && prom)
+ {
+ if ( ! task_) throw task_moved();
+ task_->set_promise(
+ new detail::promise_wrapper< R, promise >( prom) );
+ }
+#else
+ void set_promise( boost::detail::thread_move_t< promise< R > > prom)
{
- if ( ! task_)
- throw task_moved();
- return task_->get_future();
+ if ( ! task_) throw task_moved();
+ task_->set_promise(
+ new detail::promise_wrapper< R, promise >( prom) );
}
+#endif
- template< typename Cb >
- void set_wait_callback( Cb const& cb)
+ void set_promise( BOOST_RV_REF( spin::promise< R >) prom)
{
- if ( ! task_)
- throw task_moved();
- task_->set_wait_callback( cb);
+ if ( ! task_) throw task_moved();
+ task_->set_promise(
+ new detail::promise_wrapper< R, spin::promise >( prom) );
}
typedef typename shared_ptr< detail::task_base< R > >::unspecified_bool_type unspecified_bool_type;
Boost-Commit list run by bdawes at acm.org, david.abrahams at rcn.com, gregod at cs.rpi.edu, cpdaniel at pacbell.net, john at johnmaddock.co.uk