boost/thread
2
0
Fork 0
mirror of https://github.com/boostorg/thread.git synced 2026-01-24 06:22:12 +00:00
Code Issues Projects Releases Wiki Activity

Merge branch 'develop' into feature/add_launch_sync_policy

Browse Source
This commit is contained in:
Vicente J. Botet Escriba
2015-11-15 00:07:51 +01:00
parent 55536c3e23 dbf0160b1e
commit a636c8d605
6 changed files with 126 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
example/future_unwrap.cpp
View File

@@ -41,10 +41,28 @@ int main()
for (int i=0; i< number_of_tests; i++)
try
{
boost::future<boost::future<int> > outer_future = boost::async(boost::launch::async, &p2);
boost::future<int> inner_future = outer_future.unwrap();
int ii = inner_future.get();
BOOST_THREAD_LOG << "ii= "<< ii << "" << BOOST_THREAD_END_LOG;
#ifndef BOOST_NO_CXX11_RVALUE_REFERENCES
{
boost::future<int> inner_future = boost::async(boost::launch::async, &p2);
inner_future.wait();
int ii = inner_future.get();
BOOST_THREAD_LOG << "ii= "<< ii << "" << BOOST_THREAD_END_LOG;
}
#endif
{
boost::future<boost::future<int> > outer_future = boost::async(boost::launch::async, &p2);
boost::future<int> inner_future = outer_future.unwrap();
inner_future.wait();
int ii = inner_future.get();
BOOST_THREAD_LOG << "ii= "<< ii << "" << BOOST_THREAD_END_LOG;
}
{
boost::future<boost::future<int> > outer_future = boost::async(boost::launch::async, &p2);
boost::future<int> inner_future = outer_future.unwrap();
int ii = inner_future.get();
BOOST_THREAD_LOG << "ii= "<< ii << "" << BOOST_THREAD_END_LOG;
}
}
catch (std::exception& ex)
{

24
include/boost/thread/detail/invoker.hpp
View File

@@ -99,7 +99,7 @@ namespace boost
result_type
execute(tuple_indices<Indices...>)
{
return invoke(boost::move(csbl::get<0>(f_)), boost::move(csbl::get<Indices>(f_))...);
return detail::invoke(boost::move(csbl::get<0>(f_)), boost::move(csbl::get<Indices>(f_))...);
}
};
@@ -136,7 +136,7 @@ namespace boost
result_type
execute(tuple_indices<Indices...>)
{
return invoke<R>(boost::move(csbl::get<0>(f_)), boost::move(csbl::get<Indices>(f_))...);
return detail::invoke<R>(boost::move(csbl::get<0>(f_)), boost::move(csbl::get<Indices>(f_))...);
}
};
//BOOST_THREAD_DCL_MOVABLE_BEG(X) invoker<Fp> BOOST_THREAD_DCL_MOVABLE_END
@@ -190,7 +190,7 @@ namespace boost
{} \
\
result_type operator()() { \
return invoke(boost::move(fp_) \
return detail::invoke(boost::move(fp_) \
BOOST_PP_REPEAT(n, BOOST_THREAD_MOVE_DCL, ~) \
); \
} \
@@ -315,7 +315,7 @@ namespace boost
result_type operator()()
{
return invoke(boost::move(fp_)
return detail::invoke(boost::move(fp_)
, boost::move(v0_)
, boost::move(v1_)
, boost::move(v2_)
@@ -381,7 +381,7 @@ namespace boost
result_type operator()()
{
return invoke(boost::move(fp_)
return detail::invoke(boost::move(fp_)
, boost::move(v0_)
, boost::move(v1_)
, boost::move(v2_)
@@ -442,7 +442,7 @@ namespace boost
result_type operator()()
{
return invoke(boost::move(fp_)
return detail::invoke(boost::move(fp_)
, boost::move(v0_)
, boost::move(v1_)
, boost::move(v2_)
@@ -498,7 +498,7 @@ namespace boost
result_type operator()()
{
return invoke(boost::move(fp_)
return detail::invoke(boost::move(fp_)
, boost::move(v0_)
, boost::move(v1_)
, boost::move(v2_)
@@ -549,7 +549,7 @@ namespace boost
result_type operator()()
{
return invoke(boost::move(fp_)
return detail::invoke(boost::move(fp_)
, boost::move(v0_)
, boost::move(v1_)
, boost::move(v2_)
@@ -595,7 +595,7 @@ namespace boost
result_type operator()()
{
return invoke(boost::move(fp_)
return detail::invoke(boost::move(fp_)
, boost::move(v0_)
, boost::move(v1_)
, boost::move(v2_)
@@ -636,7 +636,7 @@ namespace boost
result_type operator()()
{
return invoke(boost::move(fp_)
return detail::invoke(boost::move(fp_)
, boost::move(v0_)
, boost::move(v1_)
, boost::move(v2_)
@@ -672,7 +672,7 @@ namespace boost
result_type operator()()
{
return invoke(boost::move(fp_)
return detail::invoke(boost::move(fp_)
, boost::move(v0_)
, boost::move(v1_)
);
@@ -703,7 +703,7 @@ namespace boost
result_type operator()()
{
return invoke(boost::move(fp_)
return detail::invoke(boost::move(fp_)
, boost::move(v0_)
);
}

2
include/boost/thread/detail/thread.hpp
View File

@@ -72,7 +72,7 @@ namespace boost
void run2(tuple_indices<Indices...>)
{
invoke(std::move(std::get<0>(fp)), std::move(std::get<Indices>(fp))...);
detail::invoke(std::move(std::get<0>(fp)), std::move(std::get<Indices>(fp))...);
}
void run()
{

194
include/boost/thread/future.hpp
View File

@@ -3693,9 +3693,6 @@ namespace detail
typename decay<ArgTypes>::type...
)>::type>
async(launch policy, BOOST_THREAD_FWD_REF(F) f, BOOST_THREAD_FWD_REF(ArgTypes)... args) {
typedef typename boost::result_of<typename decay<F>::type(
typename decay<ArgTypes>::type...
)>::type R;
typedef detail::invoker<typename decay<F>::type, typename decay<ArgTypes>::type...> BF;
typedef typename BF::result_type Rp;
@@ -4252,13 +4249,11 @@ namespace detail
{
F parent;
Fp continuation;
shared_ptr<shared_state_base> centinel;
public:
continuation_shared_state(BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c)
: parent(boost::move(f)),
continuation(boost::move(c)),
centinel(parent.future_)
continuation(boost::move(c))
{
}
@@ -4273,6 +4268,31 @@ namespace detail
} catch(...) {
this->mark_exceptional_finish();
}
// make sure parent is really cleared to prevent memory "leaks"
this->parent = F();
}
void call(boost::unique_lock<boost::mutex>& lck) {
try {
relocker relock(lck);
// neither continuation nor parent are protected by the lock - call() must only
// be called once, and no one else must modify it.
Rp res = this->continuation(boost::move(this->parent));
// make sure parent is really cleared to prevent memory "leaks"
this->parent = F();
relock.lock();
this->mark_finished_with_result_internal(boost::move(res), lck);
} catch (...) {
this->mark_exceptional_finish_internal(current_exception(), lck);
// make sure parent is really cleared to prevent memory "leaks"
relocker relock(lck);
this->parent = F();
}
}
static void run(shared_ptr<boost::detail::shared_state_base> that_)
@@ -4289,13 +4309,11 @@ namespace detail
{
F parent;
Fp continuation;
shared_ptr<shared_state_base> centinel;
public:
continuation_shared_state(BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c)
: parent(boost::move(f)),
continuation(boost::move(c)),
centinel(parent.future_)
continuation(boost::move(c))
{
}
@@ -4312,6 +4330,29 @@ namespace detail
} catch(...) {
this->mark_exceptional_finish();
}
// make sure parent is really cleared to prevent memory "leaks"
this->parent = F();
}
void call(boost::unique_lock<boost::mutex>& lck) {
try {
{
relocker relock(lck);
// neither continuation nor parent are protected by the lock - call() must only
// be called once, and no one else must modify it.
this->continuation(boost::move(this->parent));
// make sure parent is really cleared to prevent memory "leaks"
this->parent = F();
}
this->mark_finished_with_result_internal(lck);
} catch (...) {
this->mark_exceptional_finish_internal(current_exception(), lck);
// make sure parent is really cleared to prevent memory "leaks"
relocker relock(lck);
this->parent = F();
}
}
static void run(shared_ptr<boost::detail::shared_state_base> that_)
@@ -4536,56 +4577,7 @@ namespace detail {
boost::unique_lock<boost::mutex> lk(this->mutex);
if (this->is_deferred_) {
this->is_deferred_=false;
this->execute(lk);
}
}
virtual void execute(boost::unique_lock<boost::mutex>& lck) {
try {
Fp local_fuct=boost::move(this->continuation);
F ftmp = boost::move(this->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<typename F, typename Fp>
struct future_deferred_continuation_shared_state<F,void,Fp>: continuation_shared_state<F,void,Fp>
{
typedef continuation_shared_state<F,void,Fp> base_type;
public:
future_deferred_continuation_shared_state(BOOST_THREAD_RV_REF(F) f, BOOST_THREAD_FWD_REF(Fp) c)
: base_type(boost::move(f), boost::forward<Fp>(c))
{
this->set_deferred();
}
~future_deferred_continuation_shared_state() {
}
virtual void launch_continuation() {
boost::unique_lock<boost::mutex> lk(this->mutex);
if (this->is_deferred_) {
this->is_deferred_=false;
this->execute(lk);
}
}
virtual void execute(boost::unique_lock<boost::mutex>& lck) {
try {
Fp local_fuct=boost::move(this->continuation);
F ftmp = boost::move(this->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);
this->call(lk);
}
}
};
@@ -4609,53 +4601,7 @@ namespace detail {
boost::unique_lock<boost::mutex> lk(this->mutex);
if (this->is_deferred_) {
this->is_deferred_=false;
this->execute(lk);
}
}
virtual void execute(boost::unique_lock<boost::mutex>& lck) {
try {
Fp local_fuct=boost::move(this->continuation);
F ftmp = this->parent;
relocker relock(lck);
Rp res = local_fuct(ftmp);
relock.lock();
this->mark_finished_with_result_internal(boost::move(res), lck);
} catch (...) {
this->mark_exceptional_finish_internal(current_exception(), lck);
}
}
};
template<typename F, typename Fp>
struct shared_future_deferred_continuation_shared_state<F,void,Fp>: continuation_shared_state<F,void,Fp>
{
typedef continuation_shared_state<F,void,Fp> base_type;
public:
shared_future_deferred_continuation_shared_state(F f, BOOST_THREAD_FWD_REF(Fp) c)
: base_type(boost::move(f), boost::forward<Fp>(c))
{
this->set_deferred();
}
virtual void launch_continuation() {
boost::unique_lock<boost::mutex> lk(this->mutex);
if (this->is_deferred_) {
this->is_deferred_=false;
this->execute(lk);
}
}
virtual void execute(boost::unique_lock<boost::mutex>& lck) {
try {
Fp local_fuct=boost::move(this->continuation);
F ftmp = this->parent;
relocker relock(lck);
local_fuct(ftmp);
relock.lock();
this->mark_finished_with_result_internal(lck);
} catch (...) {
this->mark_exceptional_finish_internal(current_exception(), lck);
this->call(lk);
}
}
};
@@ -4800,7 +4746,10 @@ namespace detail {
typedef typename boost::result_of<F(BOOST_THREAD_FUTURE<R>)>::type future_type;
BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized());
boost::unique_lock<boost::mutex> lock(this->future_->mutex);
// keep state alive as we move ourself but hold the lock
shared_ptr<detail::shared_state_base> sentinel(this->future_);
boost::unique_lock<boost::mutex> lock(sentinel->mutex);
if (underlying_cast<int>(policy) & int(launch::async)) {
return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_async_continuation_shared_state<BOOST_THREAD_FUTURE<R>, future_type>(
lock, boost::move(*this), boost::forward<F>(func)
@@ -4863,7 +4812,10 @@ namespace detail {
typedef typename boost::result_of<F(BOOST_THREAD_FUTURE<R>)>::type future_type;
BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized());
boost::unique_lock<boost::mutex> lock(this->future_->mutex);
// keep state alive as we move ourself but hold the lock
shared_ptr<detail::shared_state_base> sentinel(this->future_);
boost::unique_lock<boost::mutex> lock(sentinel->mutex);
return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_executor_continuation_shared_state<Ex, BOOST_THREAD_FUTURE<R>, future_type>(ex,
lock, boost::move(*this), boost::forward<F>(func)
)));
@@ -4884,7 +4836,10 @@ namespace detail {
typedef typename boost::result_of<F(BOOST_THREAD_FUTURE<R>)>::type future_type;
BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized());
boost::unique_lock<boost::mutex> lock(this->future_->mutex);
// keep state alive as we move ourself but hold the lock
shared_ptr<detail::shared_state_base> sentinel(this->future_);
boost::unique_lock<boost::mutex> lock(sentinel->mutex);
launch policy = this->launch_policy(lock);
if (underlying_cast<int>(policy) & int(launch::deferred)) {
this->future_->wait_internal(lock);
@@ -4912,7 +4867,10 @@ namespace detail {
typedef typename boost::result_of<F(BOOST_THREAD_FUTURE<R>)>::type future_type;
BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized());
boost::unique_lock<boost::mutex> lock(this->future_->mutex);
// keep state alive as we move ourself but hold the lock
shared_ptr<detail::shared_state_base> sentinel(this->future_);
boost::unique_lock<boost::mutex> lock(sentinel->mutex);
if (underlying_cast<int>(policy) & int(launch::async)) {
return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_async_continuation_shared_state<BOOST_THREAD_FUTURE<R>, future_type>(
lock, boost::move(*this), boost::forward<F>(func)
@@ -4983,7 +4941,10 @@ namespace detail {
typedef typename boost::result_of<F(BOOST_THREAD_FUTURE<R>)>::type future_type;
BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized());
boost::unique_lock<boost::mutex> lock(this->future_->mutex);
// keep state alive as we move ourself but hold the lock
shared_ptr<detail::shared_state_base> sentinel(this->future_);
boost::unique_lock<boost::mutex> lock(sentinel->mutex);
return BOOST_THREAD_MAKE_RV_REF((boost::detail::make_future_executor_continuation_shared_state<Ex, BOOST_THREAD_FUTURE<R>, future_type>(ex,
lock, boost::move(*this), boost::forward<F>(func)
)));
@@ -5006,7 +4967,10 @@ namespace detail {
typedef typename boost::result_of<F(BOOST_THREAD_FUTURE<R>)>::type future_type;
BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized());
boost::unique_lock<boost::mutex> lock(this->future_->mutex);
// keep state alive as we move ourself but hold the lock
shared_ptr<detail::shared_state_base> sentinel(this->future_);
boost::unique_lock<boost::mutex> lock(sentinel->mutex);
launch policy = this->launch_policy(lock);
if (underlying_cast<int>(policy) & int(launch::deferred)) {
@@ -5297,7 +5261,11 @@ namespace detail
BOOST_THREAD_FUTURE<BOOST_THREAD_FUTURE<R2> >::unwrap()
{
BOOST_THREAD_ASSERT_PRECONDITION(this->future_!=0, future_uninitialized());
boost::unique_lock<boost::mutex> lock(this->future_->mutex);
// keep state alive as we move ourself but hold the lock
shared_ptr<detail::shared_state_base> sentinel(this->future_);
boost::unique_lock<boost::mutex> lock(sentinel->mutex);
return boost::detail::make_future_unwrap_shared_state<BOOST_THREAD_FUTURE<BOOST_THREAD_FUTURE<R2> >, R2>(lock, boost::move(*this));
}
#endif

20
include/boost/thread/pthread/condition_variable.hpp
View File

@@ -68,17 +68,14 @@ namespace boost
#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
thread_cv_detail::lock_on_exit<unique_lock<mutex> > guard;
detail::interruption_checker check_for_interruption(&internal_mutex,&cond);
pthread_mutex_t* the_mutex = &internal_mutex;
guard.activate(m);
do {
res = pthread_cond_wait(&cond,&internal_mutex);
} while (res == EINTR);
#else
//boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
pthread_mutex_t* the_mutex = m.mutex()->native_handle();
#endif
do {
res = pthread_cond_wait(&cond,the_mutex);
} while (res == EINTR);
#endif
}
#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
this_thread::interruption_point();
@@ -99,18 +96,17 @@ namespace boost
boost::throw_exception(condition_error(EPERM, "boost::condition_variable::do_wait_until() failed precondition mutex not owned"));
}
#endif
thread_cv_detail::lock_on_exit<unique_lock<mutex> > guard;
int cond_res;
{
#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
thread_cv_detail::lock_on_exit<unique_lock<mutex> > guard;
detail::interruption_checker check_for_interruption(&internal_mutex,&cond);
pthread_mutex_t* the_mutex = &internal_mutex;
guard.activate(m);
cond_res=pthread_cond_timedwait(&cond,&internal_mutex,&timeout);
#else
//boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
pthread_mutex_t* the_mutex = m.mutex()->native_handle();
cond_res=pthread_cond_timedwait(&cond,the_mutex,&timeout);
#endif
cond_res=pthread_cond_timedwait(&cond,the_mutex,&timeout);
}
#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
this_thread::interruption_point();
@@ -178,7 +174,7 @@ namespace boost
#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
detail::interruption_checker check_for_interruption(&internal_mutex,&cond);
#else
boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
#endif
guard.activate(m);
res=pthread_cond_wait(&cond,&internal_mutex);
@@ -405,7 +401,7 @@ namespace boost
#if defined BOOST_THREAD_PROVIDES_INTERRUPTIONS
detail::interruption_checker check_for_interruption(&internal_mutex,&cond);
#else
boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
boost::pthread::pthread_mutex_scoped_lock check_for_interruption(&internal_mutex);
#endif
guard.activate(m);
res=pthread_cond_timedwait(&cond,&internal_mutex,&timeout);
@@ -423,8 +419,6 @@ namespace boost
}
return true;
}
};
}

9
src/pthread/thread.cpp
View File

@@ -80,8 +80,8 @@ namespace boost
{
static void tls_destructor(void* data)
{
boost::detail::thread_data_base* thread_info=static_cast<boost::detail::thread_data_base*>(data);
//boost::detail::thread_data_ptr thread_info = static_cast<boost::detail::thread_data_base*>(data)->shared_from_this();
//boost::detail::thread_data_base* thread_info=static_cast<boost::detail::thread_data_base*>(data);
boost::detail::thread_data_ptr thread_info = static_cast<boost::detail::thread_data_base*>(data)->shared_from_this();
if(thread_info)
{
@@ -110,10 +110,7 @@ namespace boost
thread_info->tss_data.erase(current);
}
}
if (thread_info) // fixme: should we test this?
{
thread_info->self.reset();
}
thread_info->self.reset();
}
}
}