// (C) Copyright 2008-10 Anthony Williams // (C) Copyright 2011-2013 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) #ifndef BOOST_THREAD_FUTURE_HPP #define BOOST_THREAD_FUTURE_HPP #include // boost::thread::future requires exception handling // due to boost::exception::exception_ptr dependency #ifndef BOOST_NO_EXCEPTIONS //#include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #ifdef BOOST_THREAD_USES_CHRONO #include #endif #if defined BOOST_THREAD_PROVIDES_FUTURE_CTOR_ALLOCATORS #include #endif #include #include #if defined BOOST_THREAD_PROVIDES_FUTURE #define BOOST_THREAD_FUTURE future #else #define BOOST_THREAD_FUTURE unique_future #endif namespace boost { //enum class launch BOOST_SCOPED_ENUM_DECLARE_BEGIN(launch) { none = 0, async = 1, deferred = 2, any = async | deferred } BOOST_SCOPED_ENUM_DECLARE_END(launch) //enum class future_status BOOST_SCOPED_ENUM_DECLARE_BEGIN(future_status) { ready, timeout, deferred } BOOST_SCOPED_ENUM_DECLARE_END(future_status) class BOOST_SYMBOL_VISIBLE future_error : public std::logic_error { system::error_code ec_; public: future_error(system::error_code ec) : logic_error(ec.message()), ec_(ec) { } const system::error_code& code() const BOOST_NOEXCEPT { return ec_; } }; class BOOST_SYMBOL_VISIBLE future_uninitialized: public future_error { public: future_uninitialized() : future_error(system::make_error_code(future_errc::no_state)) {} }; class BOOST_SYMBOL_VISIBLE broken_promise: public future_error { public: broken_promise(): future_error(system::make_error_code(future_errc::broken_promise)) {} }; class BOOST_SYMBOL_VISIBLE future_already_retrieved: public future_error { public: future_already_retrieved(): future_error(system::make_error_code(future_errc::future_already_retrieved)) {} }; class BOOST_SYMBOL_VISIBLE promise_already_satisfied: public future_error { public: promise_already_satisfied(): future_error(system::make_error_code(future_errc::promise_already_satisfied)) {} }; class BOOST_SYMBOL_VISIBLE task_already_started: public future_error { public: task_already_started(): future_error(system::make_error_code(future_errc::promise_already_satisfied)) {} }; class BOOST_SYMBOL_VISIBLE task_moved: public future_error { public: task_moved(): future_error(system::make_error_code(future_errc::no_state)) {} }; class promise_moved: public future_error { public: promise_moved(): future_error(system::make_error_code(future_errc::no_state)) {} }; namespace future_state { enum state { uninitialized, waiting, ready, moved, deferred }; } namespace detail { struct relocker { boost::unique_lock& lock_; bool unlocked_; relocker(boost::unique_lock& lk): lock_(lk) { lock_.unlock(); unlocked_=true; } ~relocker() { if (unlocked_) { lock_.lock(); } } void lock() { if (unlocked_) { lock_.lock(); unlocked_=false; } } private: relocker& operator=(relocker const&); }; struct shared_state_base : enable_shared_from_this { typedef std::list waiter_list; // This type should be only included conditionally if interruptions are allowed, but is included to maintain the same layout. typedef shared_ptr continuation_ptr_type; boost::exception_ptr exception; bool done; bool is_deferred_; launch policy_; bool is_constructed; mutable boost::mutex mutex; boost::condition_variable waiters; waiter_list external_waiters; boost::function callback; // This declaration should be only included conditionally if interruptions are allowed, but is included to maintain the same layout. bool thread_was_interrupted; // This declaration should be only included conditionally, but is included to maintain the same layout. continuation_ptr_type continuation_ptr; // This declaration should be only included conditionally, but is included to maintain the same layout. virtual void launch_continuation(boost::unique_lock&) { } shared_state_base(): done(false), is_deferred_(false), policy_(launch::none), is_constructed(false), thread_was_interrupted(false), continuation_ptr() {} virtual ~shared_state_base() {} void set_deferred() { is_deferred_ = true; policy_ = launch::deferred; } void set_async() { is_deferred_ = false; policy_ = launch::async; } waiter_list::iterator register_external_waiter(boost::condition_variable_any& cv) { boost::unique_lock lock(mutex); do_callback(lock); return external_waiters.insert(external_waiters.end(),&cv); } void remove_external_waiter(waiter_list::iterator it) { boost::lock_guard lock(mutex); external_waiters.erase(it); } #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION void do_continuation(boost::unique_lock& lock) { if (continuation_ptr) { continuation_ptr->launch_continuation(lock); if (! lock.owns_lock()) lock.lock(); continuation_ptr.reset(); } } #else void do_continuation(boost::unique_lock&) { } #endif #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION void set_continuation_ptr(continuation_ptr_type continuation, boost::unique_lock& lock) { continuation_ptr= continuation; if (done) { do_continuation(lock); } } #endif void mark_finished_internal(boost::unique_lock& lock) { done=true; waiters.notify_all(); for(waiter_list::const_iterator it=external_waiters.begin(), end=external_waiters.end();it!=end;++it) { (*it)->notify_all(); } do_continuation(lock); } void make_ready() { boost::unique_lock lock(mutex); mark_finished_internal(lock); } void do_callback(boost::unique_lock& lock) { if(callback && !done) { boost::function local_callback=callback; relocker relock(lock); local_callback(); } } void wait_internal(boost::unique_lock &lk, bool rethrow=true) { do_callback(lk); //if (!done) // fixme why this doesn't work? { if (is_deferred_) { is_deferred_=false; execute(lk); //lk.unlock(); } else { while(!done) { waiters.wait(lk); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS if(rethrow && thread_was_interrupted) { throw boost::thread_interrupted(); } #endif if(rethrow && exception) { boost::rethrow_exception(exception); } } } } virtual void wait(bool rethrow=true) { boost::unique_lock lock(mutex); wait_internal(lock, rethrow); } #if defined BOOST_THREAD_USES_DATETIME bool timed_wait_until(boost::system_time const& target_time) { boost::unique_lock lock(mutex); if (is_deferred_) return false; do_callback(lock); while(!done) { bool const success=waiters.timed_wait(lock,target_time); if(!success && !done) { return false; } } return true; } #endif #ifdef BOOST_THREAD_USES_CHRONO template future_status wait_until(const chrono::time_point& abs_time) { boost::unique_lock lock(mutex); if (is_deferred_) return future_status::deferred; do_callback(lock); while(!done) { cv_status const st=waiters.wait_until(lock,abs_time); if(st==cv_status::timeout && !done) { return future_status::timeout; } } return future_status::ready; } #endif void mark_exceptional_finish_internal(boost::exception_ptr const& e, boost::unique_lock& lock) { exception=e; mark_finished_internal(lock); } void mark_exceptional_finish() { boost::unique_lock lock(mutex); mark_exceptional_finish_internal(boost::current_exception(), lock); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS void mark_interrupted_finish() { boost::unique_lock lock(mutex); thread_was_interrupted=true; mark_finished_internal(lock); } void set_interrupted_at_thread_exit() { unique_lock lk(mutex); thread_was_interrupted=true; if (has_value(lk)) { throw_exception(promise_already_satisfied()); } detail::make_ready_at_thread_exit(shared_from_this()); } #endif void set_exception_at_thread_exit(exception_ptr e) { unique_lock lk(mutex); if (has_value(lk)) { throw_exception(promise_already_satisfied()); } exception=e; this->is_constructed = true; detail::make_ready_at_thread_exit(shared_from_this()); } bool has_value() const { boost::lock_guard lock(mutex); return done && !(exception #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS || thread_was_interrupted #endif ); } bool has_value(unique_lock& ) const { return done && !(exception #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS || thread_was_interrupted #endif ); } bool has_exception() const { boost::lock_guard lock(mutex); return done && (exception #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS || thread_was_interrupted #endif ); } bool has_exception(unique_lock&) const { return done && (exception #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS || thread_was_interrupted #endif ); } bool is_deferred(boost::lock_guard&) const { return is_deferred_; } launch launch_policy(boost::unique_lock&) const { return policy_; } future_state::state get_state() const { boost::lock_guard guard(mutex); if(!done) { return future_state::waiting; } else { return future_state::ready; } } exception_ptr get_exception_ptr() { boost::unique_lock lock(mutex); return get_exception_ptr(lock); } exception_ptr get_exception_ptr(boost::unique_lock& lock) { wait_internal(lock, false); #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS if(thread_was_interrupted) { return copy_exception(boost::thread_interrupted()); } #endif return exception; } template void set_wait_callback(F f,U* u) { boost::lock_guard lock(mutex); callback=boost::bind(f,boost::ref(*u)); } virtual void execute(boost::unique_lock&) {} private: shared_state_base(shared_state_base const&); shared_state_base& operator=(shared_state_base const&); }; template struct future_traits { typedef boost::scoped_ptr storage_type; struct dummy; #ifndef BOOST_NO_CXX11_RVALUE_REFERENCES typedef T const& source_reference_type; //typedef typename boost::mpl::if_,dummy&,BOOST_THREAD_RV_REF(T)>::type rvalue_source_type; typedef BOOST_THREAD_RV_REF(T) rvalue_source_type; //typedef typename boost::mpl::if_,T,BOOST_THREAD_RV_REF(T)>::type move_dest_type; typedef T move_dest_type; #elif defined BOOST_THREAD_USES_MOVE typedef typename boost::mpl::if_c::value,T,T&>::type source_reference_type; //typedef typename boost::mpl::if_c::value,T,BOOST_THREAD_RV_REF(T)>::type rvalue_source_type; //typedef typename boost::mpl::if_c::value,BOOST_THREAD_RV_REF(T),T>::type move_dest_type; typedef BOOST_THREAD_RV_REF(T) rvalue_source_type; typedef T move_dest_type; #else typedef T& source_reference_type; typedef typename boost::mpl::if_,BOOST_THREAD_RV_REF(T),T const&>::type rvalue_source_type; typedef typename boost::mpl::if_,BOOST_THREAD_RV_REF(T),T>::type move_dest_type; #endif typedef const T& shared_future_get_result_type; static void init(storage_type& storage,source_reference_type t) { storage.reset(new T(t)); } static void init(storage_type& storage,rvalue_source_type t) { #if ! defined BOOST_NO_CXX11_RVALUE_REFERENCES storage.reset(new T(boost::forward(t))); #else storage.reset(new T(static_cast(t))); #endif } static void cleanup(storage_type& storage) { storage.reset(); } }; template struct future_traits { typedef T* storage_type; typedef T& source_reference_type; //struct rvalue_source_type //{}; typedef T& move_dest_type; typedef T& shared_future_get_result_type; static void init(storage_type& storage,T& t) { storage=&t; } static void cleanup(storage_type& storage) { storage=0; } }; template<> struct future_traits { typedef bool storage_type; typedef void move_dest_type; typedef void shared_future_get_result_type; static void init(storage_type& storage) { storage=true; } static void cleanup(storage_type& storage) { storage=false; } }; // Used to create stand-alone futures template struct shared_state: detail::shared_state_base { typedef typename future_traits::storage_type storage_type; typedef typename future_traits::source_reference_type source_reference_type; typedef typename future_traits::rvalue_source_type rvalue_source_type; typedef typename future_traits::move_dest_type move_dest_type; typedef typename future_traits::shared_future_get_result_type shared_future_get_result_type; storage_type result; shared_state(): result(0) {} ~shared_state() {} void mark_finished_with_result_internal(source_reference_type result_, boost::unique_lock& lock) { future_traits::init(result,result_); this->mark_finished_internal(lock); } void mark_finished_with_result_internal(rvalue_source_type result_, boost::unique_lock& lock) { #if ! defined BOOST_NO_CXX11_RVALUE_REFERENCES future_traits::init(result,boost::forward(result_)); #else future_traits::init(result,static_cast(result_)); #endif this->mark_finished_internal(lock); } void mark_finished_with_result(source_reference_type result_) { boost::unique_lock lock(mutex); this->mark_finished_with_result_internal(result_, lock); } void mark_finished_with_result(rvalue_source_type result_) { boost::unique_lock lock(mutex); #if ! defined BOOST_NO_CXX11_RVALUE_REFERENCES mark_finished_with_result_internal(boost::forward(result_), lock); #else mark_finished_with_result_internal(static_cast(result_), lock); #endif } virtual move_dest_type get() { wait(); return boost::move(*result); } virtual shared_future_get_result_type get_sh() { wait(); return *result; } //void set_value_at_thread_exit(const T & result_) void set_value_at_thread_exit(source_reference_type result_) { unique_lock lk(this->mutex); if (this->has_value(lk)) { throw_exception(promise_already_satisfied()); } //future_traits::init(result,result_); result.reset(new T(result_)); this->is_constructed = true; detail::make_ready_at_thread_exit(shared_from_this()); } //void set_value_at_thread_exit(BOOST_THREAD_RV_REF(T) result_) void set_value_at_thread_exit(rvalue_source_type result_) { unique_lock lk(this->mutex); if (this->has_value(lk)) throw_exception(promise_already_satisfied()); result.reset(new T(boost::move(result_))); //future_traits::init(result,static_cast(result_)); this->is_constructed = true; detail::make_ready_at_thread_exit(shared_from_this()); } private: shared_state(shared_state const&); shared_state& operator=(shared_state const&); }; template struct shared_state: detail::shared_state_base { typedef typename future_traits::storage_type storage_type; typedef typename future_traits::source_reference_type source_reference_type; typedef typename future_traits::move_dest_type move_dest_type; typedef typename future_traits::shared_future_get_result_type shared_future_get_result_type; T* result; shared_state(): result(0) {} ~shared_state() { } void mark_finished_with_result_internal(source_reference_type result_, boost::unique_lock& lock) { //future_traits::init(result,result_); result= &result_; mark_finished_internal(lock); } void mark_finished_with_result(source_reference_type result_) { boost::unique_lock lock(mutex); mark_finished_with_result_internal(result_, lock); } virtual T& get() { wait(); return *result; } virtual T& get_sh() { wait(); return *result; } void set_value_at_thread_exit(T& result_) { unique_lock lk(this->mutex); if (this->has_value(lk)) throw_exception(promise_already_satisfied()); //future_traits::init(result,result_); result= &result_; this->is_constructed = true; detail::make_ready_at_thread_exit(shared_from_this()); } private: shared_state(shared_state const&); shared_state& operator=(shared_state const&); }; template<> struct shared_state: detail::shared_state_base { typedef void shared_future_get_result_type; shared_state() {} void mark_finished_with_result_internal(boost::unique_lock& lock) { mark_finished_internal(lock); } void mark_finished_with_result() { boost::unique_lock lock(mutex); mark_finished_with_result_internal(lock); } virtual void get() { this->wait(); } virtual void get_sh() { wait(); } void set_value_at_thread_exit() { unique_lock lk(this->mutex); if (this->has_value(lk)) { throw_exception(promise_already_satisfied()); } this->is_constructed = true; detail::make_ready_at_thread_exit(shared_from_this()); } private: shared_state(shared_state const&); shared_state& operator=(shared_state const&); }; ///////////////////////// /// future_async_shared_state_base ///////////////////////// template struct future_async_shared_state_base: shared_state { typedef shared_state base_type; protected: boost::thread thr_; void join() { if (thr_.joinable()) thr_.join(); } public: future_async_shared_state_base() { this->set_async(); } explicit future_async_shared_state_base(BOOST_THREAD_RV_REF(boost::thread) th) : thr_(boost::move(th)) { this->set_async(); } ~future_async_shared_state_base() { join(); } virtual void wait(bool rethrow) { join(); this->base_type::wait(rethrow); } }; ///////////////////////// /// future_async_shared_state ///////////////////////// template struct future_async_shared_state: future_async_shared_state_base { typedef future_async_shared_state_base base_type; public: explicit future_async_shared_state(BOOST_THREAD_FWD_REF(Fp) f) : base_type(thread(&future_async_shared_state::run, this, boost::forward(f))) { } static void run(future_async_shared_state* that, BOOST_THREAD_FWD_REF(Fp) f) { try { that->mark_finished_with_result(f()); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { that->mark_interrupted_finish(); } #endif catch(...) { that->mark_exceptional_finish(); } } }; template struct future_async_shared_state: public future_async_shared_state_base { typedef future_async_shared_state_base base_type; public: explicit future_async_shared_state(BOOST_THREAD_FWD_REF(Fp) f) : base_type(thread(&future_async_shared_state::run, this, boost::forward(f))) { } static void run(future_async_shared_state* that, BOOST_THREAD_FWD_REF(Fp) f) { try { f(); that->mark_finished_with_result(); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { that->mark_interrupted_finish(); } #endif catch(...) { that->mark_exceptional_finish(); } } }; template struct future_async_shared_state: future_async_shared_state_base { typedef future_async_shared_state_base base_type; public: explicit future_async_shared_state(BOOST_THREAD_FWD_REF(Fp) f) : base_type(thread(&future_async_shared_state::run, this, boost::forward(f))) { } static void run(future_async_shared_state* that, BOOST_THREAD_FWD_REF(Fp) f) { try { that->mark_finished_with_result(f()); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { that->mark_interrupted_finish(); } #endif catch(...) { that->mark_exceptional_finish(); } } }; ////////////////////////// /// future_deferred_shared_state ////////////////////////// template struct future_deferred_shared_state: shared_state { typedef shared_state base_type; Fp func_; public: explicit future_deferred_shared_state(BOOST_THREAD_FWD_REF(Fp) f) : func_(boost::forward(f)) { this->set_deferred(); } virtual void execute(boost::unique_lock& lck) { try { Fp local_fuct=boost::move(func_); relocker relock(lck); Rp res = local_fuct(); relock.lock(); this->mark_finished_with_result_internal(boost::move(res), lck); } catch (...) { this->mark_exceptional_finish_internal(current_exception(), lck); } } }; template struct future_deferred_shared_state: shared_state { typedef shared_state base_type; Fp func_; public: explicit future_deferred_shared_state(BOOST_THREAD_FWD_REF(Fp) f) : func_(boost::forward(f)) { this->set_deferred(); } virtual void execute(boost::unique_lock& lck) { try { this->mark_finished_with_result_internal(func_(), lck); } catch (...) { this->mark_exceptional_finish_internal(current_exception(), lck); } } }; template struct future_deferred_shared_state: shared_state { typedef shared_state base_type; Fp func_; public: explicit future_deferred_shared_state(BOOST_THREAD_FWD_REF(Fp) f) : func_(boost::forward(f)) { this->set_deferred(); } virtual void execute(boost::unique_lock& lck) { try { Fp local_fuct=boost::move(func_); relocker relock(lck); local_fuct(); relock.lock(); this->mark_finished_with_result_internal(lck); } catch (...) { this->mark_exceptional_finish_internal(current_exception(), lck); } } }; // template // struct shared_state_alloc: public shared_state // { // typedef shared_state base; // Allocator alloc_; // // public: // explicit shared_state_alloc(const Allocator& a) // : alloc_(a) {} // // }; class future_waiter { struct registered_waiter; typedef std::vector::size_type count_type; struct registered_waiter { boost::shared_ptr future_; detail::shared_state_base::waiter_list::iterator wait_iterator; count_type index; registered_waiter(boost::shared_ptr const& a_future, detail::shared_state_base::waiter_list::iterator wait_iterator_, count_type index_): future_(a_future),wait_iterator(wait_iterator_),index(index_) {} }; struct all_futures_lock { #ifdef _MANAGED typedef std::ptrdiff_t count_type_portable; #else typedef count_type count_type_portable; #endif count_type_portable count; boost::scoped_array > locks; all_futures_lock(std::vector& futures): count(futures.size()),locks(new boost::unique_lock[count]) { for(count_type_portable i=0;i(futures[i].future_->mutex).move(); #else locks[i]=boost::unique_lock(futures[i].future_->mutex); #endif } } void lock() { boost::lock(locks.get(),locks.get()+count); } void unlock() { for(count_type_portable i=0;i futures; count_type future_count; public: future_waiter(): future_count(0) {} template void add(F& f) { if(f.future_) { futures.push_back(registered_waiter(f.future_,f.future_->register_external_waiter(cv),future_count)); } ++future_count; } count_type wait() { all_futures_lock lk(futures); for(;;) { for(count_type i=0;idone) { return futures[i].index; } } cv.wait(lk); } } ~future_waiter() { for(count_type i=0;iremove_external_waiter(futures[i].wait_iterator); } } }; } template class BOOST_THREAD_FUTURE; template class shared_future; template struct is_future_type { BOOST_STATIC_CONSTANT(bool, value=false); typedef void type; }; template struct is_future_type > { BOOST_STATIC_CONSTANT(bool, value=true); typedef T type; }; template struct is_future_type > { BOOST_STATIC_CONSTANT(bool, value=true); typedef T type; }; template typename boost::disable_if,void>::type wait_for_all(Iterator begin,Iterator end) { for(Iterator current=begin;current!=end;++current) { current->wait(); } } template typename boost::enable_if,void>::type wait_for_all(F1& f1,F2& f2) { f1.wait(); f2.wait(); } template void wait_for_all(F1& f1,F2& f2,F3& f3) { f1.wait(); f2.wait(); f3.wait(); } template void wait_for_all(F1& f1,F2& f2,F3& f3,F4& f4) { f1.wait(); f2.wait(); f3.wait(); f4.wait(); } template void wait_for_all(F1& f1,F2& f2,F3& f3,F4& f4,F5& f5) { f1.wait(); f2.wait(); f3.wait(); f4.wait(); f5.wait(); } template typename boost::disable_if,Iterator>::type wait_for_any(Iterator begin,Iterator end) { if(begin==end) return end; detail::future_waiter waiter; for(Iterator current=begin;current!=end;++current) { waiter.add(*current); } return boost::next(begin,waiter.wait()); } template typename boost::enable_if,unsigned>::type wait_for_any(F1& f1,F2& f2) { detail::future_waiter waiter; waiter.add(f1); waiter.add(f2); return waiter.wait(); } template unsigned wait_for_any(F1& f1,F2& f2,F3& f3) { detail::future_waiter waiter; waiter.add(f1); waiter.add(f2); waiter.add(f3); return waiter.wait(); } template unsigned wait_for_any(F1& f1,F2& f2,F3& f3,F4& f4) { detail::future_waiter waiter; waiter.add(f1); waiter.add(f2); waiter.add(f3); waiter.add(f4); return waiter.wait(); } template unsigned wait_for_any(F1& f1,F2& f2,F3& f3,F4& f4,F5& f5) { detail::future_waiter waiter; waiter.add(f1); waiter.add(f2); waiter.add(f3); waiter.add(f4); waiter.add(f5); return waiter.wait(); } template class promise; template class packaged_task; namespace detail { /// Common implementation for all the futures independently of the return type class base_future { //BOOST_THREAD_MOVABLE(base_future) }; /// Common implementation for future and shared_future. template class basic_future : public base_future { protected: public: typedef boost::shared_ptr > future_ptr; future_ptr future_; basic_future(future_ptr a_future): future_(a_future) { } // Copy construction from a shared_future explicit basic_future(const shared_future&) BOOST_NOEXCEPT; public: typedef future_state::state state; BOOST_THREAD_MOVABLE(basic_future) basic_future(): future_() {} ~basic_future() {} basic_future(BOOST_THREAD_RV_REF(basic_future) other) BOOST_NOEXCEPT: future_(BOOST_THREAD_RV(other).future_) { BOOST_THREAD_RV(other).future_.reset(); } basic_future& operator=(BOOST_THREAD_RV_REF(basic_future) other) BOOST_NOEXCEPT { future_=BOOST_THREAD_RV(other).future_; BOOST_THREAD_RV(other).future_.reset(); return *this; } void swap(basic_future& that) BOOST_NOEXCEPT { future_.swap(that.future_); } // functions to check state, and wait for ready state get_state() const { if(!future_) { return future_state::uninitialized; } return future_->get_state(); } bool is_ready() const { return get_state()==future_state::ready; } bool has_exception() const { return future_ && future_->has_exception(); } bool has_value() const { return future_ && future_->has_value(); } launch launch_policy(boost::unique_lock& lk) const { if ( future_ ) return future_->launch_policy(lk); else return launch(launch::none); } exception_ptr get_exception_ptr() { return future_ ? future_->get_exception_ptr() : exception_ptr(); } bool valid() const BOOST_NOEXCEPT { return future_ != 0; } void wait() const { if(!future_) { boost::throw_exception(future_uninitialized()); } future_->wait(false); } #if defined BOOST_THREAD_USES_DATETIME template bool timed_wait(Duration const& rel_time) const { return timed_wait_until(boost::get_system_time()+rel_time); } bool timed_wait_until(boost::system_time const& abs_time) const { if(!future_) { boost::throw_exception(future_uninitialized()); } return future_->timed_wait_until(abs_time); } #endif #ifdef BOOST_THREAD_USES_CHRONO template future_status wait_for(const chrono::duration& rel_time) const { return wait_until(chrono::steady_clock::now() + rel_time); } template future_status wait_until(const chrono::time_point& abs_time) const { if(!future_) { boost::throw_exception(future_uninitialized()); } return future_->wait_until(abs_time); } #endif }; } // detail BOOST_THREAD_DCL_MOVABLE_BEG(R) detail::basic_future BOOST_THREAD_DCL_MOVABLE_END namespace detail { #if (!defined _MSC_VER || _MSC_VER >= 1400) // _MSC_VER == 1400 on MSVC 2005 template BOOST_THREAD_FUTURE make_future_async_shared_state(BOOST_THREAD_FWD_REF(Fp) f); template BOOST_THREAD_FUTURE make_future_deferred_shared_state(BOOST_THREAD_FWD_REF(Fp) f); #endif // #if (!defined _MSC_VER || _MSC_VER >= 1400) #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION template struct future_deferred_continuation_shared_state; template struct future_async_continuation_shared_state; template BOOST_THREAD_FUTURE make_future_async_continuation_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c); template BOOST_THREAD_FUTURE make_future_deferred_continuation_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c); #endif #if defined BOOST_THREAD_PROVIDES_FUTURE_UNWRAP template struct future_unwrap_shared_state; template inline BOOST_THREAD_FUTURE make_future_unwrap_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f); #endif } template class BOOST_THREAD_FUTURE : public detail::basic_future { private: typedef detail::basic_future base_type; typedef typename base_type::future_ptr future_ptr; friend class shared_future; friend class promise; #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION template friend struct detail::future_async_continuation_shared_state; template friend struct detail::future_deferred_continuation_shared_state; template friend BOOST_THREAD_FUTURE detail::make_future_async_continuation_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c); template friend BOOST_THREAD_FUTURE detail::make_future_deferred_continuation_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c); #endif #if defined BOOST_THREAD_PROVIDES_FUTURE_UNWRAP template friend struct detail::future_unwrap_shared_state; template friend BOOST_THREAD_FUTURE detail::make_future_unwrap_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f); #endif #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK template friend class packaged_task; // todo check if this works in windows #else friend class packaged_task; #endif friend class detail::future_waiter; template friend BOOST_THREAD_FUTURE detail::make_future_async_shared_state(BOOST_THREAD_FWD_REF(Fp) f); template friend BOOST_THREAD_FUTURE detail::make_future_deferred_shared_state(BOOST_THREAD_FWD_REF(Fp) f); typedef typename detail::future_traits::move_dest_type move_dest_type; BOOST_THREAD_FUTURE(future_ptr a_future): base_type(a_future) { } public: BOOST_THREAD_MOVABLE_ONLY(BOOST_THREAD_FUTURE) typedef future_state::state state; typedef R value_type; // EXTENSION BOOST_CONSTEXPR BOOST_THREAD_FUTURE() {} ~BOOST_THREAD_FUTURE() {} BOOST_THREAD_FUTURE(BOOST_THREAD_RV_REF(BOOST_THREAD_FUTURE) other) BOOST_NOEXCEPT: base_type(boost::move(static_cast(BOOST_THREAD_RV(other)))) { } inline BOOST_THREAD_FUTURE(BOOST_THREAD_RV_REF(BOOST_THREAD_FUTURE >) other); // EXTENSION BOOST_THREAD_FUTURE& operator=(BOOST_THREAD_RV_REF(BOOST_THREAD_FUTURE) other) BOOST_NOEXCEPT { this->base_type::operator=(boost::move(static_cast(BOOST_THREAD_RV(other)))); return *this; } shared_future share() { return shared_future(::boost::move(*this)); } void swap(BOOST_THREAD_FUTURE& other) { static_cast(this)->swap(other); } // todo this function must be private and friendship provided to the internal users. void set_async() { this->future_->set_async(); } // todo this function must be private and friendship provided to the internal users. void set_deferred() { this->future_->set_deferred(); } // retrieving the value move_dest_type get() { if(!this->future_) { boost::throw_exception(future_uninitialized()); } future_ptr fut_=this->future_; #ifdef BOOST_THREAD_PROVIDES_FUTURE_INVALID_AFTER_GET this->future_.reset(); #endif return fut_->get(); } template typename boost::disable_if< is_void, move_dest_type>::type get_or(BOOST_THREAD_RV_REF(R2) v) { if(!this->future_) { boost::throw_exception(future_uninitialized()); } this->future_->wait(false); future_ptr fut_=this->future_; #ifdef BOOST_THREAD_PROVIDES_FUTURE_INVALID_AFTER_GET this->future_.reset(); #endif if (fut_->has_value()) { return fut_->get(); } else { return boost::move(v); } } template typename boost::disable_if< is_void, move_dest_type>::type get_or(R2 const& v) // EXTENSION { if(!this->future_) { boost::throw_exception(future_uninitialized()); } this->future_->wait(false); future_ptr fut_=this->future_; #ifdef BOOST_THREAD_PROVIDES_FUTURE_INVALID_AFTER_GET this->future_.reset(); #endif if (fut_->has_value()) { return fut_->get(); } else { return v; } } #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION // template // auto then(F&& func) -> BOOST_THREAD_FUTURE; //#if defined(BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR) // template // inline BOOST_THREAD_FUTURE then(RF(*func)(BOOST_THREAD_FUTURE&)); // template // inline BOOST_THREAD_FUTURE then(launch policy, RF(*func)(BOOST_THREAD_FUTURE&)); //#endif template inline BOOST_THREAD_FUTURE::type> then(BOOST_THREAD_FWD_REF(F) func); // EXTENSION template inline BOOST_THREAD_FUTURE::type> then(launch policy, BOOST_THREAD_FWD_REF(F) func); // EXTENSION template inline typename boost::disable_if< is_void, BOOST_THREAD_FUTURE >::type fallback_to(BOOST_THREAD_RV_REF(R2) v); // EXTENSION template inline typename boost::disable_if< is_void, BOOST_THREAD_FUTURE >::type fallback_to(R2 const& v); // EXTENSION #endif //#if defined BOOST_THREAD_PROVIDES_FUTURE_UNWRAP // inline // typename boost::enable_if< // is_future_type, // value_type // //BOOST_THREAD_FUTURE::type> // >::type // unwrap(); //#endif }; BOOST_THREAD_DCL_MOVABLE_BEG(T) BOOST_THREAD_FUTURE BOOST_THREAD_DCL_MOVABLE_END template class BOOST_THREAD_FUTURE > : public detail::basic_future > { typedef BOOST_THREAD_FUTURE R; private: typedef detail::basic_future base_type; typedef typename base_type::future_ptr future_ptr; friend class shared_future; friend class promise; #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION template friend struct detail::future_async_continuation_shared_state; template friend struct detail::future_deferred_continuation_shared_state; template friend BOOST_THREAD_FUTURE detail::make_future_async_continuation_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c); template friend BOOST_THREAD_FUTURE detail::make_future_deferred_continuation_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c); #endif #if defined BOOST_THREAD_PROVIDES_FUTURE_UNWRAP template friend struct detail::future_unwrap_shared_state; template friend BOOST_THREAD_FUTURE detail::make_future_unwrap_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f); #endif #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK template friend class packaged_task; // todo check if this works in windows #else friend class packaged_task; #endif friend class detail::future_waiter; template friend BOOST_THREAD_FUTURE detail::make_future_async_shared_state(BOOST_THREAD_FWD_REF(Fp) f); template friend BOOST_THREAD_FUTURE detail::make_future_deferred_shared_state(BOOST_THREAD_FWD_REF(Fp) f); typedef typename detail::future_traits::move_dest_type move_dest_type; BOOST_THREAD_FUTURE(future_ptr a_future): base_type(a_future) { } public: BOOST_THREAD_MOVABLE_ONLY(BOOST_THREAD_FUTURE) typedef future_state::state state; typedef R value_type; // EXTENSION BOOST_CONSTEXPR BOOST_THREAD_FUTURE() {} ~BOOST_THREAD_FUTURE() {} BOOST_THREAD_FUTURE(BOOST_THREAD_RV_REF(BOOST_THREAD_FUTURE) other) BOOST_NOEXCEPT: base_type(boost::move(static_cast(BOOST_THREAD_RV(other)))) { } BOOST_THREAD_FUTURE& operator=(BOOST_THREAD_RV_REF(BOOST_THREAD_FUTURE) other) BOOST_NOEXCEPT { this->base_type::operator=(boost::move(static_cast(BOOST_THREAD_RV(other)))); return *this; } shared_future share() { return shared_future(::boost::move(*this)); } void swap(BOOST_THREAD_FUTURE& other) { static_cast(this)->swap(other); } // todo this function must be private and friendship provided to the internal users. void set_async() { this->future_->set_async(); } // todo this function must be private and friendship provided to the internal users. void set_deferred() { this->future_->set_deferred(); } // retrieving the value move_dest_type get() { if(!this->future_) { boost::throw_exception(future_uninitialized()); } future_ptr fut_=this->future_; #ifdef BOOST_THREAD_PROVIDES_FUTURE_INVALID_AFTER_GET this->future_.reset(); #endif return fut_->get(); } move_dest_type get_or(BOOST_THREAD_RV_REF(R) v) // EXTENSION { if(!this->future_) { boost::throw_exception(future_uninitialized()); } this->future_->wait(false); future_ptr fut_=this->future_; #ifdef BOOST_THREAD_PROVIDES_FUTURE_INVALID_AFTER_GET this->future_.reset(); #endif if (fut_->has_value()) return fut_->get(); else return boost::move(v); } move_dest_type get_or(R const& v) // EXTENSION { if(!this->future_) { boost::throw_exception(future_uninitialized()); } this->future_->wait(false); future_ptr fut_=this->future_; #ifdef BOOST_THREAD_PROVIDES_FUTURE_INVALID_AFTER_GET this->future_.reset(); #endif if (fut_->has_value()) return fut_->get(); else return v; } #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION // template // auto then(F&& func) -> BOOST_THREAD_FUTURE; //#if defined(BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR) // template // inline BOOST_THREAD_FUTURE then(RF(*func)(BOOST_THREAD_FUTURE&)); // template // inline BOOST_THREAD_FUTURE then(launch policy, RF(*func)(BOOST_THREAD_FUTURE&)); //#endif template inline BOOST_THREAD_FUTURE::type> then(BOOST_THREAD_FWD_REF(F) func); // EXTENSION template inline BOOST_THREAD_FUTURE::type> then(launch policy, BOOST_THREAD_FWD_REF(F) func); // EXTENSION #endif #if defined BOOST_THREAD_PROVIDES_FUTURE_UNWRAP inline BOOST_THREAD_FUTURE unwrap(); // EXTENSION #endif }; template class shared_future : public detail::basic_future { typedef detail::basic_future base_type; typedef typename base_type::future_ptr future_ptr; friend class detail::future_waiter; friend class promise; #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION template friend struct detail::future_async_continuation_shared_state; template friend struct detail::future_deferred_continuation_shared_state; template friend BOOST_THREAD_FUTURE detail::make_future_async_continuation_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c); template friend BOOST_THREAD_FUTURE detail::make_future_deferred_continuation_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c); #endif #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK template friend class packaged_task;// todo check if this works in windows #else friend class packaged_task; #endif shared_future(future_ptr a_future): base_type(a_future) {} public: BOOST_THREAD_MOVABLE(shared_future) typedef R value_type; // EXTENSION shared_future(shared_future const& other): base_type(other) {} typedef future_state::state state; BOOST_CONSTEXPR shared_future() {} ~shared_future() {} shared_future& operator=(shared_future const& other) { shared_future(other).swap(*this); return *this; } shared_future(BOOST_THREAD_RV_REF(shared_future) other) BOOST_NOEXCEPT : base_type(boost::move(static_cast(BOOST_THREAD_RV(other)))) { BOOST_THREAD_RV(other).future_.reset(); } shared_future(BOOST_THREAD_RV_REF( BOOST_THREAD_FUTURE ) other) BOOST_NOEXCEPT : base_type(boost::move(static_cast(BOOST_THREAD_RV(other)))) { } shared_future& operator=(BOOST_THREAD_RV_REF(shared_future) other) BOOST_NOEXCEPT { base_type::operator=(boost::move(static_cast(BOOST_THREAD_RV(other)))); return *this; } shared_future& operator=(BOOST_THREAD_RV_REF( BOOST_THREAD_FUTURE ) other) BOOST_NOEXCEPT { base_type::operator=(boost::move(static_cast(BOOST_THREAD_RV(other)))); return *this; } void swap(shared_future& other) BOOST_NOEXCEPT { static_cast(this)->swap(other); } // retrieving the value typename detail::shared_state::shared_future_get_result_type get() { if(!this->future_) { boost::throw_exception(future_uninitialized()); } return this->future_->get_sh(); } template typename boost::disable_if< is_void, typename detail::shared_state::shared_future_get_result_type>::type get_or(BOOST_THREAD_RV_REF(R2) v) // EXTENSION { if(!this->future_) { boost::throw_exception(future_uninitialized()); } future_ptr fut_=this->future_; fut_->wait(); if (fut_->has_value()) return fut_->get_sh(); else return boost::move(v); } #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION // template // auto then(F&& func) -> BOOST_THREAD_FUTURE; // template // auto then(launch, F&& func) -> BOOST_THREAD_FUTURE; //#if defined(BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR) // template // inline BOOST_THREAD_FUTURE then(RF(*func)(shared_future&)); // template // inline BOOST_THREAD_FUTURE then(launch policy, RF(*func)(shared_future&)); //#endif template inline BOOST_THREAD_FUTURE::type> then(BOOST_THREAD_FWD_REF(F) func); // EXTENSION template inline BOOST_THREAD_FUTURE::type> then(launch policy, BOOST_THREAD_FWD_REF(F) func); // EXTENSION #endif //#if defined BOOST_THREAD_PROVIDES_FUTURE_UNWRAP // inline // typename boost::enable_if_c< // is_future_type::value, // BOOST_THREAD_FUTURE::type> // >::type // unwrap(); //#endif }; BOOST_THREAD_DCL_MOVABLE_BEG(T) shared_future BOOST_THREAD_DCL_MOVABLE_END namespace detail { /// Copy construction from a shared_future template inline basic_future::basic_future(const shared_future& other) BOOST_NOEXCEPT : future_(other.future_) { } } template class promise { typedef boost::shared_ptr > future_ptr; future_ptr future_; bool future_obtained; void lazy_init() { #if defined BOOST_THREAD_PROVIDES_PROMISE_LAZY #include if(!atomic_load(&future_)) { future_ptr blank; atomic_compare_exchange(&future_,&blank,future_ptr(new detail::shared_state)); } #include #endif } public: BOOST_THREAD_MOVABLE_ONLY(promise) #if defined BOOST_THREAD_PROVIDES_FUTURE_CTOR_ALLOCATORS template promise(boost::allocator_arg_t, Allocator a) { typedef typename Allocator::template rebind >::other A2; A2 a2(a); typedef thread_detail::allocator_destructor D; future_ = future_ptr(::new(a2.allocate(1)) detail::shared_state(), D(a2, 1) ); future_obtained = false; } #endif promise(): #if defined BOOST_THREAD_PROVIDES_PROMISE_LAZY future_(), #else future_(new detail::shared_state()), #endif future_obtained(false) {} ~promise() { if(future_) { boost::unique_lock lock(future_->mutex); if(!future_->done && !future_->is_constructed) { future_->mark_exceptional_finish_internal(boost::copy_exception(broken_promise()), lock); } } } // Assignment promise(BOOST_THREAD_RV_REF(promise) rhs) BOOST_NOEXCEPT : future_(BOOST_THREAD_RV(rhs).future_),future_obtained(BOOST_THREAD_RV(rhs).future_obtained) { BOOST_THREAD_RV(rhs).future_.reset(); BOOST_THREAD_RV(rhs).future_obtained=false; } promise & operator=(BOOST_THREAD_RV_REF(promise) rhs) BOOST_NOEXCEPT { future_=BOOST_THREAD_RV(rhs).future_; future_obtained=BOOST_THREAD_RV(rhs).future_obtained; BOOST_THREAD_RV(rhs).future_.reset(); BOOST_THREAD_RV(rhs).future_obtained=false; return *this; } void swap(promise& other) { future_.swap(other.future_); std::swap(future_obtained,other.future_obtained); } // Result retrieval BOOST_THREAD_FUTURE get_future() { lazy_init(); if (future_.get()==0) { boost::throw_exception(promise_moved()); } if (future_obtained) { boost::throw_exception(future_already_retrieved()); } future_obtained=true; return BOOST_THREAD_FUTURE(future_); } void set_value(typename detail::future_traits::source_reference_type r) { lazy_init(); boost::unique_lock lock(future_->mutex); if(future_->done) { boost::throw_exception(promise_already_satisfied()); } future_->mark_finished_with_result_internal(r, lock); } // void set_value(R && r); void set_value(typename detail::future_traits::rvalue_source_type r) { lazy_init(); boost::unique_lock lock(future_->mutex); if(future_->done) { boost::throw_exception(promise_already_satisfied()); } #if ! defined BOOST_NO_CXX11_RVALUE_REFERENCES future_->mark_finished_with_result_internal(boost::forward(r), lock); #else future_->mark_finished_with_result_internal(static_cast::rvalue_source_type>(r), lock); #endif } void set_exception(boost::exception_ptr p) { lazy_init(); boost::unique_lock lock(future_->mutex); if(future_->done) { boost::throw_exception(promise_already_satisfied()); } future_->mark_exceptional_finish_internal(p, lock); } template void set_exception(E ex) { set_exception(copy_exception(ex)); } // setting the result with deferred notification void set_value_at_thread_exit(const R& r) { if (future_.get()==0) { boost::throw_exception(promise_moved()); } future_->set_value_at_thread_exit(r); } void set_value_at_thread_exit(BOOST_THREAD_RV_REF(R) r) { if (future_.get()==0) { boost::throw_exception(promise_moved()); } future_->set_value_at_thread_exit(boost::move(r)); } void set_exception_at_thread_exit(exception_ptr e) { if (future_.get()==0) { boost::throw_exception(promise_moved()); } future_->set_exception_at_thread_exit(e); } template void set_exception_at_thread_exit(E ex) { set_exception_at_thread_exit(copy_exception(ex)); } template void set_wait_callback(F f) { lazy_init(); future_->set_wait_callback(f,this); } }; template class promise { typedef boost::shared_ptr > future_ptr; future_ptr future_; bool future_obtained; void lazy_init() { #if defined BOOST_THREAD_PROVIDES_PROMISE_LAZY #include if(!atomic_load(&future_)) { future_ptr blank; atomic_compare_exchange(&future_,&blank,future_ptr(new detail::shared_state)); } #include #endif } public: BOOST_THREAD_MOVABLE_ONLY(promise) #if defined BOOST_THREAD_PROVIDES_FUTURE_CTOR_ALLOCATORS template promise(boost::allocator_arg_t, Allocator a) { typedef typename Allocator::template rebind >::other A2; A2 a2(a); typedef thread_detail::allocator_destructor D; future_ = future_ptr(::new(a2.allocate(1)) detail::shared_state(), D(a2, 1) ); future_obtained = false; } #endif promise(): #if defined BOOST_THREAD_PROVIDES_PROMISE_LAZY future_(), #else future_(new detail::shared_state()), #endif future_obtained(false) {} ~promise() { if(future_) { boost::unique_lock lock(future_->mutex); if(!future_->done && !future_->is_constructed) { future_->mark_exceptional_finish_internal(boost::copy_exception(broken_promise()), lock); } } } // Assignment promise(BOOST_THREAD_RV_REF(promise) rhs) BOOST_NOEXCEPT : future_(BOOST_THREAD_RV(rhs).future_),future_obtained(BOOST_THREAD_RV(rhs).future_obtained) { BOOST_THREAD_RV(rhs).future_.reset(); BOOST_THREAD_RV(rhs).future_obtained=false; } promise & operator=(BOOST_THREAD_RV_REF(promise) rhs) BOOST_NOEXCEPT { future_=BOOST_THREAD_RV(rhs).future_; future_obtained=BOOST_THREAD_RV(rhs).future_obtained; BOOST_THREAD_RV(rhs).future_.reset(); BOOST_THREAD_RV(rhs).future_obtained=false; return *this; } void swap(promise& other) { future_.swap(other.future_); std::swap(future_obtained,other.future_obtained); } // Result retrieval BOOST_THREAD_FUTURE get_future() { lazy_init(); if (future_.get()==0) { boost::throw_exception(promise_moved()); } if (future_obtained) { boost::throw_exception(future_already_retrieved()); } future_obtained=true; return BOOST_THREAD_FUTURE(future_); } void set_value(R& r) { lazy_init(); boost::unique_lock lock(future_->mutex); if(future_->done) { boost::throw_exception(promise_already_satisfied()); } future_->mark_finished_with_result_internal(r, lock); } void set_exception(boost::exception_ptr p) { lazy_init(); boost::unique_lock lock(future_->mutex); if(future_->done) { boost::throw_exception(promise_already_satisfied()); } future_->mark_exceptional_finish_internal(p, lock); } template void set_exception(E ex) { set_exception(copy_exception(ex)); } // setting the result with deferred notification void set_value_at_thread_exit(R& r) { if (future_.get()==0) { boost::throw_exception(promise_moved()); } future_->set_value_at_thread_exit(r); } void set_exception_at_thread_exit(exception_ptr e) { if (future_.get()==0) { boost::throw_exception(promise_moved()); } future_->set_exception_at_thread_exit(e); } template void set_exception_at_thread_exit(E ex) { set_exception_at_thread_exit(copy_exception(ex)); } template void set_wait_callback(F f) { lazy_init(); future_->set_wait_callback(f,this); } }; template <> class promise { typedef boost::shared_ptr > future_ptr; future_ptr future_; bool future_obtained; void lazy_init() { #if defined BOOST_THREAD_PROVIDES_PROMISE_LAZY if(!atomic_load(&future_)) { future_ptr blank; atomic_compare_exchange(&future_,&blank,future_ptr(new detail::shared_state)); } #endif } public: BOOST_THREAD_MOVABLE_ONLY(promise) #if defined BOOST_THREAD_PROVIDES_FUTURE_CTOR_ALLOCATORS template promise(boost::allocator_arg_t, Allocator a) { typedef typename Allocator::template rebind >::other A2; A2 a2(a); typedef thread_detail::allocator_destructor D; future_ = future_ptr(::new(a2.allocate(1)) detail::shared_state(), D(a2, 1) ); future_obtained = false; } #endif promise(): #if defined BOOST_THREAD_PROVIDES_PROMISE_LAZY future_(), #else future_(new detail::shared_state), #endif future_obtained(false) {} ~promise() { if(future_) { boost::unique_lock lock(future_->mutex); if(!future_->done && !future_->is_constructed) { future_->mark_exceptional_finish_internal(boost::copy_exception(broken_promise()), lock); } } } // Assignment promise(BOOST_THREAD_RV_REF(promise) rhs) BOOST_NOEXCEPT : future_(BOOST_THREAD_RV(rhs).future_),future_obtained(BOOST_THREAD_RV(rhs).future_obtained) { // we need to release the future as shared_ptr doesn't implements move semantics BOOST_THREAD_RV(rhs).future_.reset(); BOOST_THREAD_RV(rhs).future_obtained=false; } promise & operator=(BOOST_THREAD_RV_REF(promise) rhs) BOOST_NOEXCEPT { future_=BOOST_THREAD_RV(rhs).future_; future_obtained=BOOST_THREAD_RV(rhs).future_obtained; BOOST_THREAD_RV(rhs).future_.reset(); BOOST_THREAD_RV(rhs).future_obtained=false; return *this; } void swap(promise& other) { future_.swap(other.future_); std::swap(future_obtained,other.future_obtained); } // Result retrieval BOOST_THREAD_FUTURE get_future() { lazy_init(); if (future_.get()==0) { boost::throw_exception(promise_moved()); } if(future_obtained) { boost::throw_exception(future_already_retrieved()); } future_obtained=true; return BOOST_THREAD_FUTURE(future_); } void set_value() { lazy_init(); boost::unique_lock lock(future_->mutex); if(future_->done) { boost::throw_exception(promise_already_satisfied()); } future_->mark_finished_with_result_internal(lock); } void set_exception(boost::exception_ptr p) { lazy_init(); boost::unique_lock lock(future_->mutex); if(future_->done) { boost::throw_exception(promise_already_satisfied()); } future_->mark_exceptional_finish_internal(p,lock); } template void set_exception(E ex) { set_exception(copy_exception(ex)); } // setting the result with deferred notification void set_value_at_thread_exit() { if (future_.get()==0) { boost::throw_exception(promise_moved()); } future_->set_value_at_thread_exit(); } void set_exception_at_thread_exit(exception_ptr e) { if (future_.get()==0) { boost::throw_exception(promise_moved()); } future_->set_exception_at_thread_exit(e); } template void set_exception_at_thread_exit(E ex) { set_exception_at_thread_exit(copy_exception(ex)); } template void set_wait_callback(F f) { lazy_init(); future_->set_wait_callback(f,this); } }; #if defined BOOST_THREAD_PROVIDES_FUTURE_CTOR_ALLOCATORS namespace container { template struct uses_allocator , Alloc> : true_type { }; } #endif BOOST_THREAD_DCL_MOVABLE_BEG(T) promise BOOST_THREAD_DCL_MOVABLE_END namespace detail { #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK template struct task_base_shared_state; #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template struct task_base_shared_state: #else template struct task_base_shared_state: #endif #else template struct task_base_shared_state: #endif detail::shared_state { bool started; task_base_shared_state(): started(false) {} void reset() { started=false; } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) virtual void do_run(BOOST_THREAD_RV_REF(ArgTypes) ... args)=0; void run(BOOST_THREAD_RV_REF(ArgTypes) ... args) #else virtual void do_run()=0; void run() #endif { { boost::lock_guard lk(this->mutex); if(started) { boost::throw_exception(task_already_started()); } started=true; } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) do_run(boost::forward(args)...); #else do_run(); #endif } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) virtual void do_apply(BOOST_THREAD_RV_REF(ArgTypes) ... args)=0; void apply(BOOST_THREAD_RV_REF(ArgTypes) ... args) #else virtual void do_apply()=0; void apply() #endif { { boost::lock_guard lk(this->mutex); if(started) { boost::throw_exception(task_already_started()); } started=true; } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) do_apply(boost::forward(args)...); #else do_apply(); #endif } void owner_destroyed() { boost::unique_lock lk(this->mutex); if(!started) { started=true; this->mark_exceptional_finish_internal(boost::copy_exception(boost::broken_promise()), lk); } } }; #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK template struct task_shared_state; #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template struct task_shared_state: task_base_shared_state #else template struct task_shared_state: task_base_shared_state #endif #else template struct task_shared_state: task_base_shared_state #endif { private: task_shared_state(task_shared_state&); public: F f; task_shared_state(F const& f_): f(f_) {} task_shared_state(BOOST_THREAD_RV_REF(F) f_): f(boost::move(f_)) {} #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_apply(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { this->set_value_at_thread_exit(f(boost::forward(args)...)); } #else void do_apply() { try { this->set_value_at_thread_exit(f()); } #endif #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->set_interrupted_at_thread_exit(); } #endif catch(...) { this->set_exception_at_thread_exit(current_exception()); } } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_run(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { this->mark_finished_with_result(f(boost::forward(args)...)); } #else void do_run() { try { #if ! defined BOOST_NO_CXX11_RVALUE_REFERENCES R res((f())); this->mark_finished_with_result(boost::move(res)); #else this->mark_finished_with_result(f()); #endif } #endif #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->mark_interrupted_finish(); } #endif catch(...) { this->mark_exceptional_finish(); } } }; #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template struct task_shared_state: task_base_shared_state #else template struct task_shared_state: task_base_shared_state #endif #else template struct task_shared_state: task_base_shared_state #endif { private: task_shared_state(task_shared_state&); public: F f; task_shared_state(F const& f_): f(f_) {} task_shared_state(BOOST_THREAD_RV_REF(F) f_): f(boost::move(f_)) {} #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_apply(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { this->set_value_at_thread_exit(f(boost::forward(args)...)); } #else void do_apply() { try { this->set_value_at_thread_exit(f()); } #endif #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->set_interrupted_at_thread_exit(); } #endif catch(...) { this->set_exception_at_thread_exit(current_exception()); } } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_run(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { this->mark_finished_with_result(f(boost::forward(args)...)); } #else void do_run() { try { R& res((f())); this->mark_finished_with_result(res); } #endif #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->mark_interrupted_finish(); } #endif catch(...) { this->mark_exceptional_finish(); } } }; #if defined(BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR) #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template struct task_shared_state: task_base_shared_state #else template struct task_shared_state: task_base_shared_state #endif #else template struct task_shared_state : task_base_shared_state #endif { private: task_shared_state(task_shared_state&); public: R (*f)(); task_shared_state(R (*f_)()): f(f_) {} #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_apply(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { this->set_value_at_thread_exit(f(boost::forward(args)...)); } #else void do_apply() { try { R r((f())); this->set_value_at_thread_exit(boost::move(r)); } #endif #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->set_interrupted_at_thread_exit(); } #endif catch(...) { this->set_exception_at_thread_exit(current_exception()); } } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_run(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { this->mark_finished_with_result(f(boost::forward(args)...)); } #else void do_run() { try { R res((f())); this->mark_finished_with_result(boost::move(res)); } #endif #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->mark_interrupted_finish(); } #endif catch(...) { this->mark_exceptional_finish(); } } }; #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template struct task_shared_state: task_base_shared_state #else template struct task_shared_state: task_base_shared_state #endif #else template struct task_shared_state : task_base_shared_state #endif { private: task_shared_state(task_shared_state&); public: R& (*f)(); task_shared_state(R& (*f_)()): f(f_) {} #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_apply(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { this->set_value_at_thread_exit(f(boost::forward(args)...)); } #else void do_apply() { try { this->set_value_at_thread_exit(f()); } #endif #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->set_interrupted_at_thread_exit(); } #endif catch(...) { this->set_exception_at_thread_exit(current_exception()); } } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_run(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { this->mark_finished_with_result(f(boost::forward(args)...)); } #else void do_run() { try { this->mark_finished_with_result(f()); } #endif #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->mark_interrupted_finish(); } #endif catch(...) { this->mark_exceptional_finish(); } } }; #endif #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template struct task_shared_state: task_base_shared_state #else template struct task_shared_state: task_base_shared_state #endif #else template struct task_shared_state: task_base_shared_state #endif { private: task_shared_state(task_shared_state&); public: F f; task_shared_state(F const& f_): f(f_) {} task_shared_state(BOOST_THREAD_RV_REF(F) f_): f(boost::move(f_)) {} #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_apply(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { f(boost::forward(args)...); #else void do_apply() { try { f(); #endif this->set_value_at_thread_exit(); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->set_interrupted_at_thread_exit(); } #endif catch(...) { this->set_exception_at_thread_exit(current_exception()); } } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_run(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { f(boost::forward(args)...); #else void do_run() { try { f(); #endif this->mark_finished_with_result(); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->mark_interrupted_finish(); } #endif catch(...) { this->mark_exceptional_finish(); } } }; #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template struct task_shared_state: task_base_shared_state #else template<> struct task_shared_state: task_base_shared_state #endif #else template<> struct task_shared_state: task_base_shared_state #endif { private: task_shared_state(task_shared_state&); public: void (*f)(); task_shared_state(void (*f_)()): f(f_) {} #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_apply(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { f(boost::forward(args)...); #else void do_apply() { try { f(); #endif this->set_value_at_thread_exit(); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->set_interrupted_at_thread_exit(); } #endif catch(...) { this->set_exception_at_thread_exit(current_exception()); } } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void do_run(BOOST_THREAD_RV_REF(ArgTypes) ... args) { try { f(boost::forward(args)...); #else void do_run() { try { f(); #endif this->mark_finished_with_result(); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { this->mark_interrupted_finish(); } #endif catch(...) { this->mark_exceptional_finish(); } } }; } #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template class packaged_task { typedef boost::shared_ptr > task_ptr; boost::shared_ptr > task; #else template class packaged_task { typedef boost::shared_ptr > task_ptr; boost::shared_ptr > task; #endif #else template class packaged_task { typedef boost::shared_ptr > task_ptr; boost::shared_ptr > task; #endif bool future_obtained; struct dummy; public: typedef R result_type; BOOST_THREAD_MOVABLE_ONLY(packaged_task) packaged_task(): future_obtained(false) {} // construction and destruction #if defined(BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR) #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) explicit packaged_task(R(*f)(), BOOST_THREAD_FWD_REF(ArgTypes)... args) { typedef R(*FR)(BOOST_THREAD_FWD_REF(ArgTypes)...); typedef detail::task_shared_state task_shared_state_type; task= task_ptr(new task_shared_state_type(f, boost::forward(args)...)); future_obtained=false; } #else explicit packaged_task(R(*f)()) { typedef R(*FR)(); typedef detail::task_shared_state task_shared_state_type; task= task_ptr(new task_shared_state_type(f)); future_obtained=false; } #endif #else explicit packaged_task(R(*f)()) { typedef R(*FR)(); typedef detail::task_shared_state task_shared_state_type; task= task_ptr(new task_shared_state_type(f)); future_obtained=false; } #endif #endif #ifndef BOOST_NO_CXX11_RVALUE_REFERENCES template explicit packaged_task(BOOST_THREAD_FWD_REF(F) f , typename boost::disable_if::type, packaged_task>, dummy* >::type=0 ) { typedef typename remove_cv::type>::type FR; #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) typedef detail::task_shared_state task_shared_state_type; #else typedef detail::task_shared_state task_shared_state_type; #endif #else typedef detail::task_shared_state task_shared_state_type; #endif task = task_ptr(new task_shared_state_type(boost::forward(f))); future_obtained = false; } #else template explicit packaged_task(F const& f , typename boost::disable_if::type, packaged_task>, dummy* >::type=0 ) { #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) typedef detail::task_shared_state task_shared_state_type; #else typedef detail::task_shared_state task_shared_state_type; #endif #else typedef detail::task_shared_state task_shared_state_type; #endif task = task_ptr(new task_shared_state_type(f)); future_obtained=false; } template explicit packaged_task(BOOST_THREAD_RV_REF(F) f) { #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) typedef detail::task_shared_state task_shared_state_type; task = task_ptr(new task_shared_state_type(boost::forward(f))); #else typedef detail::task_shared_state task_shared_state_type; task = task_ptr(new task_shared_state_type(boost::move(f))); // TODO forward #endif #else typedef detail::task_shared_state task_shared_state_type; task = task_ptr(new task_shared_state_type(boost::forward(f))); #endif future_obtained=false; } #endif #if defined BOOST_THREAD_PROVIDES_FUTURE_CTOR_ALLOCATORS #if defined(BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR) template packaged_task(boost::allocator_arg_t, Allocator a, R(*f)()) { typedef R(*FR)(); #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) typedef detail::task_shared_state task_shared_state_type; #else typedef detail::task_shared_state task_shared_state_type; #endif #else typedef detail::task_shared_state task_shared_state_type; #endif typedef typename Allocator::template rebind::other A2; A2 a2(a); typedef thread_detail::allocator_destructor D; task = task_ptr(::new(a2.allocate(1)) task_shared_state_type(f), D(a2, 1) ); future_obtained = false; } #endif // BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR #if ! defined BOOST_NO_CXX11_RVALUE_REFERENCES template packaged_task(boost::allocator_arg_t, Allocator a, BOOST_THREAD_FWD_REF(F) f) { typedef typename remove_cv::type>::type FR; #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) typedef detail::task_shared_state task_shared_state_type; #else typedef detail::task_shared_state task_shared_state_type; #endif #else typedef detail::task_shared_state task_shared_state_type; #endif typedef typename Allocator::template rebind::other A2; A2 a2(a); typedef thread_detail::allocator_destructor D; task = task_ptr(::new(a2.allocate(1)) task_shared_state_type(boost::forward(f)), D(a2, 1) ); future_obtained = false; } #else // ! defined BOOST_NO_CXX11_RVALUE_REFERENCES template packaged_task(boost::allocator_arg_t, Allocator a, const F& f) { #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) typedef detail::task_shared_state task_shared_state_type; #else typedef detail::task_shared_state task_shared_state_type; #endif #else typedef detail::task_shared_state task_shared_state_type; #endif typedef typename Allocator::template rebind::other A2; A2 a2(a); typedef thread_detail::allocator_destructor D; task = task_ptr(::new(a2.allocate(1)) task_shared_state_type(f), D(a2, 1) ); future_obtained = false; } template packaged_task(boost::allocator_arg_t, Allocator a, BOOST_THREAD_RV_REF(F) f) { #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) typedef detail::task_shared_state task_shared_state_type; #else typedef detail::task_shared_state task_shared_state_type; #endif #else typedef detail::task_shared_state task_shared_state_type; #endif typedef typename Allocator::template rebind::other A2; A2 a2(a); typedef thread_detail::allocator_destructor D; #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) task = task_ptr(::new(a2.allocate(1)) task_shared_state_type(boost::forward(f)), D(a2, 1) ); #else task = task_ptr(::new(a2.allocate(1)) task_shared_state_type(boost::move(f)), D(a2, 1) ); // TODO forward #endif future_obtained = false; } #endif //BOOST_NO_CXX11_RVALUE_REFERENCES #endif // BOOST_THREAD_PROVIDES_FUTURE_CTOR_ALLOCATORS ~packaged_task() { if(task) { task->owner_destroyed(); } } // assignment packaged_task(BOOST_THREAD_RV_REF(packaged_task) other) BOOST_NOEXCEPT : future_obtained(BOOST_THREAD_RV(other).future_obtained) { task.swap(BOOST_THREAD_RV(other).task); BOOST_THREAD_RV(other).future_obtained=false; } packaged_task& operator=(BOOST_THREAD_RV_REF(packaged_task) other) BOOST_NOEXCEPT { // todo use forward #if ! defined BOOST_NO_CXX11_RVALUE_REFERENCES packaged_task temp(boost::move(other)); #else packaged_task temp(static_cast(other)); #endif swap(temp); return *this; } void reset() { if (!valid()) throw future_error(system::make_error_code(future_errc::no_state)); task->reset(); future_obtained=false; } void swap(packaged_task& other) BOOST_NOEXCEPT { task.swap(other.task); std::swap(future_obtained,other.future_obtained); } bool valid() const BOOST_NOEXCEPT { return task.get()!=0; } // result retrieval BOOST_THREAD_FUTURE get_future() { if(!task) { boost::throw_exception(task_moved()); } else if(!future_obtained) { future_obtained=true; return BOOST_THREAD_FUTURE(task); } else { boost::throw_exception(future_already_retrieved()); } //return BOOST_THREAD_FUTURE(); } // execution #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) void operator()(BOOST_THREAD_RV_REF(ArgTypes)... args) { if(!task) { boost::throw_exception(task_moved()); } task->run(boost::forward(args)...); } void make_ready_at_thread_exit(ArgTypes... args) { if(!task) { boost::throw_exception(task_moved()); } if (task->has_value()) { boost::throw_exception(promise_already_satisfied()); } task->apply(boost::forward(args)...); } #else void operator()() { if(!task) { boost::throw_exception(task_moved()); } task->run(); } void make_ready_at_thread_exit() { if(!task) { boost::throw_exception(task_moved()); } if (task->has_value()) boost::throw_exception(promise_already_satisfied()); task->apply(); } #endif template void set_wait_callback(F f) { task->set_wait_callback(f,this); } }; #if defined BOOST_THREAD_PROVIDES_FUTURE_CTOR_ALLOCATORS namespace container { template struct uses_allocator, Alloc> : public true_type {}; } #endif BOOST_THREAD_DCL_MOVABLE_BEG(T) packaged_task BOOST_THREAD_DCL_MOVABLE_END namespace detail { //////////////////////////////// // make_future_deferred_shared_state //////////////////////////////// template BOOST_THREAD_FUTURE make_future_deferred_shared_state(BOOST_THREAD_FWD_REF(Fp) f) { shared_ptr > h(new future_deferred_shared_state(boost::forward(f))); return BOOST_THREAD_FUTURE(h); } //////////////////////////////// // make_future_async_shared_state //////////////////////////////// template BOOST_THREAD_FUTURE make_future_async_shared_state(BOOST_THREAD_FWD_REF(Fp) f) { shared_ptr > h(new future_async_shared_state(boost::forward(f))); return BOOST_THREAD_FUTURE(h); } } //////////////////////////////// // template // future async(launch policy, F&&, ArgTypes&&...); //////////////////////////////// #if defined BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template BOOST_THREAD_FUTURE async(launch policy, R(*f)(BOOST_THREAD_FWD_REF(ArgTypes)...), BOOST_THREAD_FWD_REF(ArgTypes)... args) { typedef R(*F)(BOOST_THREAD_FWD_REF(ArgTypes)...); typedef detail::async_func::type, typename decay::type...> BF; typedef typename BF::result_type Rp; #else template BOOST_THREAD_FUTURE async(launch policy, R(*f)()) { typedef packaged_task packaged_task_type; #endif #else template BOOST_THREAD_FUTURE async(launch policy, R(*f)()) { typedef packaged_task packaged_task_type; #endif if (int(policy) & int(launch::async)) { #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) return BOOST_THREAD_MAKE_RV_REF(boost::detail::make_future_async_shared_state( BF( thread_detail::decay_copy(boost::forward(f)) , thread_detail::decay_copy(boost::forward(args))... ) )); #else packaged_task_type pt( f ); BOOST_THREAD_FUTURE ret = BOOST_THREAD_MAKE_RV_REF(pt.get_future()); ret.set_async(); boost::thread( boost::move(pt) ).detach(); return ::boost::move(ret); #endif } else if (int(policy) & int(launch::deferred)) { #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) return BOOST_THREAD_MAKE_RV_REF(boost::detail::make_future_deferred_shared_state( BF( thread_detail::decay_copy(boost::forward(f)) , thread_detail::decay_copy(boost::forward(args))... ) )); #else std::terminate(); BOOST_THREAD_FUTURE ret; return ::boost::move(ret); #endif } else { std::terminate(); BOOST_THREAD_FUTURE ret; return ::boost::move(ret); } } #endif #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK #if defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template BOOST_THREAD_FUTURE::type( typename decay::type... )>::type> async(launch policy, BOOST_THREAD_FWD_REF(F) f, BOOST_THREAD_FWD_REF(ArgTypes)... args) { typedef typename boost::result_of::type( typename decay::type... )>::type R; typedef detail::async_func::type, typename decay::type...> BF; typedef typename BF::result_type Rp; #else template BOOST_THREAD_FUTURE::type()>::type> async(launch policy, BOOST_THREAD_FWD_REF(F) f) { typedef typename boost::result_of::type()>::type R; typedef packaged_task packaged_task_type; #endif #else template BOOST_THREAD_FUTURE::type()>::type> async(launch policy, BOOST_THREAD_FWD_REF(F) f) { typedef typename boost::result_of::type()>::type R; typedef packaged_task packaged_task_type; #endif if (int(policy) & int(launch::async)) { #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) return BOOST_THREAD_MAKE_RV_REF(boost::detail::make_future_async_shared_state( BF( thread_detail::decay_copy(boost::forward(f)) , thread_detail::decay_copy(boost::forward(args))... ) )); #else packaged_task_type pt( boost::forward(f) ); BOOST_THREAD_FUTURE ret = pt.get_future(); ret.set_async(); boost::thread( boost::move(pt) ).detach(); return ::boost::move(ret); #endif } else if (int(policy) & int(launch::deferred)) { #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) return BOOST_THREAD_MAKE_RV_REF(boost::detail::make_future_deferred_shared_state( BF( thread_detail::decay_copy(boost::forward(f)) , thread_detail::decay_copy(boost::forward(args))... ) )); #else std::terminate(); BOOST_THREAD_FUTURE ret; return ::boost::move(ret); // return boost::detail::make_future_deferred_shared_state( // BF( // thread_detail::decay_copy(boost::forward(f)) // ) // ); #endif } else { std::terminate(); BOOST_THREAD_FUTURE ret; return ::boost::move(ret); } } //////////////////////////////// // template // future async(F&&, ArgTypes&&...); //////////////////////////////// #if defined BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template BOOST_THREAD_FUTURE async(R(*f)(BOOST_THREAD_FWD_REF(ArgTypes)...), BOOST_THREAD_FWD_REF(ArgTypes)... args) { return BOOST_THREAD_MAKE_RV_REF(async(launch(launch::any), f, boost::forward(args)...)); } #else template BOOST_THREAD_FUTURE async(R(*f)()) { return BOOST_THREAD_MAKE_RV_REF(async(launch(launch::any), f)); } #endif #endif #if defined BOOST_THREAD_PROVIDES_SIGNATURE_PACKAGED_TASK && defined(BOOST_THREAD_PROVIDES_VARIADIC_THREAD) template BOOST_THREAD_FUTURE::type( typename decay::type... )>::type> async(BOOST_THREAD_FWD_REF(F) f, BOOST_THREAD_FWD_REF(ArgTypes)... args) { return BOOST_THREAD_MAKE_RV_REF(async(launch(launch::any), boost::forward(f), boost::forward(args)...)); } #else template BOOST_THREAD_FUTURE::type> async(BOOST_THREAD_RV_REF(F) f) { return BOOST_THREAD_MAKE_RV_REF(async(launch(launch::any), boost::forward(f))); } #endif //////////////////////////////// // make_future deprecated //////////////////////////////// template BOOST_THREAD_FUTURE::type> make_future(BOOST_THREAD_FWD_REF(T) value) { typedef typename decay::type future_value_type; promise p; p.set_value(boost::forward(value)); return BOOST_THREAD_MAKE_RV_REF(p.get_future()); } #if defined BOOST_THREAD_USES_MOVE inline BOOST_THREAD_FUTURE make_future() { promise p; p.set_value(); return BOOST_THREAD_MAKE_RV_REF(p.get_future()); } #endif //////////////////////////////// // make_ready_future //////////////////////////////// template BOOST_THREAD_FUTURE::type> make_ready_future(BOOST_THREAD_FWD_REF(T) value) { typedef typename decay::type future_value_type; promise p; p.set_value(boost::forward(value)); return BOOST_THREAD_MAKE_RV_REF(p.get_future()); } #if defined BOOST_THREAD_USES_MOVE inline BOOST_THREAD_FUTURE make_ready_future() { promise p; p.set_value(); return BOOST_THREAD_MAKE_RV_REF(p.get_future()); } #endif template BOOST_THREAD_FUTURE make_ready_future(exception_ptr ex) { promise p; p.set_exception(ex); return BOOST_THREAD_MAKE_RV_REF(p.get_future()); } template BOOST_THREAD_FUTURE make_ready_future(E ex) { promise p; p.set_exception(boost::copy_exception(ex)); return BOOST_THREAD_MAKE_RV_REF(p.get_future()); } #if 0 template make_future(CLOSURE closure) -> BOOST_THREAD_FUTURE { typedef decltype(closure()) T; promise p; try { p.set_value(closure()); } catch(...) { p.set_exception(std::current_exception()); } return BOOST_THREAD_MAKE_RV_REF(p.get_future()); } #endif //////////////////////////////// // make_shared_future deprecated //////////////////////////////// template shared_future::type> make_shared_future(BOOST_THREAD_FWD_REF(T) value) { typedef typename decay::type future_type; promise p; p.set_value(boost::forward(value)); return BOOST_THREAD_MAKE_RV_REF(p.get_future().share()); } inline shared_future make_shared_future() { promise p; return BOOST_THREAD_MAKE_RV_REF(p.get_future().share()); } // //////////////////////////////// // // make_ready_shared_future // //////////////////////////////// // template // shared_future::type> make_ready_shared_future(BOOST_THREAD_FWD_REF(T) value) // { // typedef typename decay::type future_type; // promise p; // p.set_value(boost::forward(value)); // return p.get_future().share(); // } // // // inline shared_future make_ready_shared_future() // { // promise p; // return BOOST_THREAD_MAKE_RV_REF(p.get_future().share()); // // } // // //////////////////////////////// // // make_exceptional_shared_future // //////////////////////////////// // template // shared_future make_exceptional_shared_future(exception_ptr ex) // { // promise p; // p.set_exception(ex); // return p.get_future().share(); // } //////////////////////////////// // detail::future_async_continuation_shared_state //////////////////////////////// #if defined BOOST_THREAD_PROVIDES_FUTURE_CONTINUATION namespace detail { ///////////////////////// /// future_async_continuation_shared_state ///////////////////////// template struct future_async_continuation_shared_state: future_async_shared_state_base { F parent; Fp continuation; public: future_async_continuation_shared_state( BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c ) : parent(boost::move(f)), continuation(boost::move(c)) { } void launch_continuation(boost::unique_lock& lock) { lock.unlock(); this->thr_ = thread(&future_async_continuation_shared_state::run, this); } static void run(future_async_continuation_shared_state* that) { try { that->mark_finished_with_result(that->continuation(boost::move(that->parent))); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { that->mark_interrupted_finish(); } #endif catch(...) { that->mark_exceptional_finish(); } } ~future_async_continuation_shared_state() { this->join(); } }; template struct future_async_continuation_shared_state: public future_async_shared_state_base { F parent; Fp continuation; public: future_async_continuation_shared_state( BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c ) : parent(boost::move(f)), continuation(boost::move(c)) { } void launch_continuation(boost::unique_lock& lk) { lk.unlock(); this->thr_ = thread(&future_async_continuation_shared_state::run, this); } static void run(future_async_continuation_shared_state* that) { try { that->continuation(boost::move(that->parent)); that->mark_finished_with_result(); } #if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS catch(thread_interrupted& ) { that->mark_interrupted_finish(); } #endif catch(...) { that->mark_exceptional_finish(); } } ~future_async_continuation_shared_state() { this->join(); } }; ////////////////////////// /// future_deferred_continuation_shared_state ////////////////////////// template struct future_deferred_continuation_shared_state: shared_state { F parent; Fp continuation; public: future_deferred_continuation_shared_state( BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c ) : parent(boost::move(f)), continuation(boost::move(c)) { this->set_deferred(); } virtual void launch_continuation(boost::unique_lock& lk) { execute(lk); } virtual void execute(boost::unique_lock& lck) { try { Fp local_fuct=boost::move(continuation); F ftmp = boost::move(parent); relocker relock(lck); Rp res = local_fuct(boost::move(ftmp)); relock.lock(); this->mark_finished_with_result_internal(boost::move(res), lck); } catch (...) { this->mark_exceptional_finish_internal(current_exception(), lck); } } }; template struct future_deferred_continuation_shared_state: shared_state { F parent; Fp continuation; public: future_deferred_continuation_shared_state( BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c ): parent(boost::move(f)), continuation(boost::move(c)) { this->set_deferred(); } virtual void launch_continuation(boost::unique_lock& lk) { execute(lk); } virtual void execute(boost::unique_lock& lck) { try { Fp local_fuct=boost::move(continuation); F ftmp = boost::move(parent); relocker relock(lck); local_fuct(boost::move(ftmp)); relock.lock(); this->mark_finished_with_result_internal(lck); } catch (...) { this->mark_exceptional_finish_internal(current_exception(), lck); } } }; //////////////////////////////// // make_future_deferred_continuation_shared_state //////////////////////////////// template BOOST_THREAD_FUTURE make_future_deferred_continuation_shared_state( boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c ) { shared_ptr > h(new future_deferred_continuation_shared_state(boost::move(f), boost::forward(c))); h->parent.future_->set_continuation_ptr(h, lock); return BOOST_THREAD_FUTURE(h); } //////////////////////////////// // make_future_async_continuation_shared_state //////////////////////////////// template BOOST_THREAD_FUTURE make_future_async_continuation_shared_state( boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c ) { shared_ptr > h(new future_async_continuation_shared_state(boost::move(f), boost::forward(c))); h->parent.future_->set_continuation_ptr(h, lock); return BOOST_THREAD_FUTURE(h); } } //////////////////////////////// // template // auto future::then(F&& func) -> BOOST_THREAD_FUTURE; //////////////////////////////// template template inline BOOST_THREAD_FUTURE)>::type> BOOST_THREAD_FUTURE::then(launch policy, BOOST_THREAD_FWD_REF(F) func) { typedef typename boost::result_of)>::type future_type; BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized()); boost::unique_lock lock(this->future_->mutex); if (int(policy) & int(launch::async)) { return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_async_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ))); } else if (int(policy) & int(launch::deferred)) { return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_deferred_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ))); } else { return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_async_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ))); } } template template inline BOOST_THREAD_FUTURE)>::type> BOOST_THREAD_FUTURE::then(BOOST_THREAD_FWD_REF(F) func) { typedef typename boost::result_of)>::type future_type; BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized()); boost::unique_lock lock(this->future_->mutex); if (int(this->launch_policy(lock)) & int(launch::async)) { return boost::detail::make_future_async_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ); } else if (int(this->launch_policy(lock)) & int(launch::deferred)) { this->future_->wait_internal(lock); return boost::detail::make_future_deferred_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ); } else { return boost::detail::make_future_async_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ); } } //#if 0 && defined(BOOST_THREAD_RVALUE_REFERENCES_DONT_MATCH_FUNTION_PTR) // template // template // BOOST_THREAD_FUTURE // BOOST_THREAD_FUTURE::then(RF(*func)(BOOST_THREAD_FUTURE&)) // { // // typedef RF future_type; // // if (this->future_) // { // boost::unique_lock lock(this->future_->mutex); // detail::future_continuation, future_type, RF(*)(BOOST_THREAD_FUTURE&) > *ptr = // new detail::future_continuation, future_type, RF(*)(BOOST_THREAD_FUTURE&)>(*this, func); // if (ptr==0) // { // return BOOST_THREAD_MAKE_RV_REF(BOOST_THREAD_FUTURE()); // } // this->future_->set_continuation_ptr(ptr, lock); // return ptr->get_future(); // } else { // // fixme what to do when the future has no associated state? // return BOOST_THREAD_MAKE_RV_REF(BOOST_THREAD_FUTURE()); // } // // } // template // template // BOOST_THREAD_FUTURE // BOOST_THREAD_FUTURE::then(launch policy, RF(*func)(BOOST_THREAD_FUTURE&)) // { // // typedef RF future_type; // // if (this->future_) // { // boost::unique_lock lock(this->future_->mutex); // detail::future_continuation, future_type, RF(*)(BOOST_THREAD_FUTURE&) > *ptr = // new detail::future_continuation, future_type, RF(*)(BOOST_THREAD_FUTURE&)>(*this, func, policy); // if (ptr==0) // { // return BOOST_THREAD_MAKE_RV_REF(BOOST_THREAD_FUTURE()); // } // this->future_->set_continuation_ptr(ptr, lock); // return ptr->get_future(); // } else { // // fixme what to do when the future has no associated state? // return BOOST_THREAD_MAKE_RV_REF(BOOST_THREAD_FUTURE()); // } // // } //#endif template template inline BOOST_THREAD_FUTURE)>::type> shared_future::then(launch policy, BOOST_THREAD_FWD_REF(F) func) { typedef typename boost::result_of)>::type future_type; BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized()); boost::unique_lock lock(this->future_->mutex); if (int(policy) & int(launch::async)) { return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_async_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ))); } else if (int(policy) & int(launch::deferred)) { return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_deferred_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ))); } else { return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_async_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ))); } } template template inline BOOST_THREAD_FUTURE)>::type> shared_future::then(BOOST_THREAD_FWD_REF(F) func) { typedef typename boost::result_of)>::type future_type; BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized()); boost::unique_lock lock(this->future_->mutex); if (int(this->launch_policy(lock)) & int(launch::async)) { return boost::detail::make_future_async_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ); } else if (int(this->launch_policy(lock)) & int(launch::deferred)) { this->future_->wait_internal(lock); return boost::detail::make_future_deferred_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ); } else { return boost::detail::make_future_async_continuation_shared_state, future_type, F>( lock, boost::move(*this), boost::forward(func) ); } } namespace detail { template struct mfallbacker_to { T value_; typedef T result_type; mfallbacker_to(BOOST_THREAD_RV_REF(T) v) : value_(boost::move(v)) {} T operator()(BOOST_THREAD_FUTURE fut) { return fut.get_or(boost::move(value_)); } }; template struct cfallbacker_to { T value_; typedef T result_type; cfallbacker_to(T const& v) : value_(v) {} T operator()(BOOST_THREAD_FUTURE fut) { return fut.get_or(value_); } }; } //////////////////////////////// // future future::fallback_to(R&& v); //////////////////////////////// template template inline typename boost::disable_if< is_void, BOOST_THREAD_FUTURE >::type BOOST_THREAD_FUTURE::fallback_to(BOOST_THREAD_RV_REF(R2) v) { return then(detail::mfallbacker_to(boost::move(v))); } template template inline typename boost::disable_if< is_void, BOOST_THREAD_FUTURE >::type BOOST_THREAD_FUTURE::fallback_to(R2 const& v) { return then(detail::cfallbacker_to(v)); } #endif #if defined BOOST_THREAD_PROVIDES_FUTURE_UNWRAP namespace detail { ///////////////////////// /// future_unwrap_shared_state ///////////////////////// template struct future_unwrap_shared_state: shared_state { F parent; public: explicit future_unwrap_shared_state( BOOST_THREAD_RV_REF(F) f ) : parent(boost::move(f)) { } virtual void wait(bool ) // todo see if rethrow must be used { boost::unique_lock lock(mutex); parent.get().wait(); } virtual Rp get() { boost::unique_lock lock(mutex); return parent.get().get(); } }; template BOOST_THREAD_FUTURE make_future_unwrap_shared_state(boost::unique_lock &lock, BOOST_THREAD_RV_REF(F) f) { shared_ptr > h(new future_unwrap_shared_state(boost::move(f))); h->parent.future_->set_continuation_ptr(h, lock); return BOOST_THREAD_FUTURE(h); } } template inline BOOST_THREAD_FUTURE::BOOST_THREAD_FUTURE(BOOST_THREAD_RV_REF(BOOST_THREAD_FUTURE >) other): base_type(other.unwrap()) { } template BOOST_THREAD_FUTURE BOOST_THREAD_FUTURE >::unwrap() { BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized()); boost::unique_lock lock(this->future_->mutex); return boost::detail::make_future_unwrap_shared_state >, R2>(lock, boost::move(*this)); } #endif } #endif // BOOST_NO_EXCEPTION #endif // header