Subject: [Boost-commit] svn:boost r86616 - in trunk: boost/thread libs/thread/example
From: vicente.botet_at_[hidden]
Date: 2013-11-11 05:43:16
Author: viboes
Date: 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013)
New Revision: 86616
URL: http://svn.boost.org/trac/boost/changeset/86616
Log:
Thread: add _back/_front using queue_op_status interface to sync queues and update the depending classes thread_pool and user_scheduler; Simplfy some test to isolate the errors on msvc-12.
Text files modified:
trunk/boost/thread/sync_bounded_queue.hpp | 225 ++++++++++++++++++++++++++++++++++++++-
trunk/boost/thread/sync_queue.hpp | 226 +++++++++++++++++++++++++++++++++++++++
trunk/boost/thread/thread_pool.hpp | 12 +-
trunk/boost/thread/user_scheduler.hpp | 12 +-
trunk/libs/thread/example/executor.cpp | 1
trunk/libs/thread/example/future_fallback_to.cpp | 2
trunk/libs/thread/example/producer_consumer.cpp | 43 +++++--
trunk/libs/thread/example/producer_consumer_bounded.cpp | 44 +++++--
trunk/libs/thread/example/thread_pool.cpp | 1
9 files changed, 512 insertions(+), 54 deletions(-)
Modified: trunk/boost/thread/sync_bounded_queue.hpp
==============================================================================
--- trunk/boost/thread/sync_bounded_queue.hpp Sun Nov 10 23:11:13 2013 (r86615)
+++ trunk/boost/thread/sync_bounded_queue.hpp 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013) (r86616)
@@ -16,9 +16,10 @@
#include <boost/thread/mutex.hpp>
#include <boost/thread/detail/move.hpp>
#include <boost/throw_exception.hpp>
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
#include <boost/smart_ptr/shared_ptr.hpp>
#include <boost/smart_ptr/make_shared.hpp>
-
+#endif
#include <boost/config/abi_prefix.hpp>
namespace boost
@@ -28,8 +29,10 @@
{ success = 0, empty, full, closed, busy }
BOOST_SCOPED_ENUM_DECLARE_END(queue_op_status)
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
struct no_block_tag{};
BOOST_CONSTEXPR_OR_CONST no_block_tag no_block = {};
+#endif
struct sync_queue_is_closed : std::exception
{
@@ -59,14 +62,23 @@
// Modifiers
inline void close();
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline void push(const value_type& x);
inline void push(BOOST_THREAD_RV_REF(value_type) x);
inline bool try_push(const value_type& x);
inline bool try_push(BOOST_THREAD_RV_REF(value_type) x);
inline bool try_push(no_block_tag, const value_type& x);
inline bool try_push(no_block_tag, BOOST_THREAD_RV_REF(value_type) x);
+#endif
+ inline void push_back(const value_type& x);
+ inline void push_back(BOOST_THREAD_RV_REF(value_type) x);
+ inline queue_op_status try_push_back(const value_type& x);
+ inline queue_op_status try_push_back(BOOST_THREAD_RV_REF(value_type) x);
+ inline queue_op_status nonblocking_push_back(const value_type& x);
+ inline queue_op_status nonblocking_push_back(BOOST_THREAD_RV_REF(value_type) x);
// Observers/Modifiers
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline void pull(value_type&);
inline void pull(ValueType& elem, bool & closed);
// enable_if is_nothrow_copy_movable<value_type>
@@ -75,6 +87,12 @@
inline bool try_pull(value_type&);
inline bool try_pull(no_block_tag,value_type&);
inline shared_ptr<ValueType> try_pull();
+#endif
+ inline void pull_front(value_type&);
+ // enable_if is_nothrow_copy_movable<value_type>
+ inline value_type pull_front();
+ inline queue_op_status try_pull_front(value_type&);
+ inline queue_op_status nonblocking_pull_front(value_type&);
private:
mutable mutex mtx_;
@@ -120,11 +138,17 @@
}
inline void throw_if_closed(unique_lock<mutex>&);
+ inline bool closed(unique_lock<mutex>&) const;
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline bool try_pull(value_type& x, unique_lock<mutex>& lk);
+ inline shared_ptr<value_type> try_pull(unique_lock<mutex>& lk);
inline bool try_push(const value_type& x, unique_lock<mutex>& lk);
inline bool try_push(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
- inline shared_ptr<value_type> try_pull(unique_lock<mutex>& lk);
+#endif
+ inline queue_op_status try_pull_front(value_type& x, unique_lock<mutex>& lk);
+ inline queue_op_status try_push_back(const value_type& x, unique_lock<mutex>& lk);
+ inline queue_op_status try_push_back(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
inline void wait_until_not_empty(unique_lock<mutex>& lk);
inline void wait_until_not_empty(unique_lock<mutex>& lk, bool&);
@@ -151,6 +175,7 @@
}
}
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline void pull(value_type& elem, unique_lock<mutex>& lk)
{
elem = boost::move(data_[out_]);
@@ -171,6 +196,20 @@
notify_not_full_if_needed(lk);
return res;
}
+#endif
+ inline void pull_front(value_type& elem, unique_lock<mutex>& lk)
+ {
+ elem = boost::move(data_[out_]);
+ out_ = inc(out_);
+ notify_not_full_if_needed(lk);
+ }
+ inline value_type pull_front(unique_lock<mutex>& lk)
+ {
+ value_type elem = boost::move(data_[out_]);
+ out_ = inc(out_);
+ notify_not_full_if_needed(lk);
+ return boost::move(elem);
+ }
inline void set_in(size_type in, unique_lock<mutex>& lk)
{
@@ -248,6 +287,11 @@
lock_guard<mutex> lk(mtx_);
return closed_;
}
+ template <typename ValueType>
+ bool sync_bounded_queue<ValueType>::closed(unique_lock<mutex>& ) const
+ {
+ return closed_;
+ }
template <typename ValueType>
bool sync_bounded_queue<ValueType>::empty() const
@@ -276,7 +320,7 @@
return size(lk);
}
-
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_bounded_queue<ValueType>::try_pull(ValueType& elem, unique_lock<mutex>& lk)
{
@@ -298,7 +342,6 @@
}
return ptr_pull(lk);
}
-
template <typename ValueType>
bool sync_bounded_queue<ValueType>::try_pull(ValueType& elem)
{
@@ -313,7 +356,28 @@
throw;
}
}
+#endif
+
+ template <typename ValueType>
+ queue_op_status sync_bounded_queue<ValueType>::try_pull_front(ValueType& elem, unique_lock<mutex>& lk)
+ {
+ if (empty(lk))
+ {
+ if (closed(lk)) return queue_op_status::closed;
+ return queue_op_status::empty;
+ }
+ pull_front(elem, lk);
+ return queue_op_status::success;
+ }
+
+ template <typename ValueType>
+ queue_op_status sync_bounded_queue<ValueType>::try_pull_front(ValueType& elem)
+ {
+ unique_lock<mutex> lk(mtx_);
+ return try_pull_front(elem, lk);
+ }
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_bounded_queue<ValueType>::try_pull(no_block_tag,ValueType& elem)
{
@@ -346,6 +410,18 @@
throw;
}
}
+#endif
+
+ template <typename ValueType>
+ queue_op_status sync_bounded_queue<ValueType>::nonblocking_pull_front(ValueType& elem)
+ {
+ unique_lock<mutex> lk(mtx_, try_to_lock);
+ if (!lk.owns_lock())
+ {
+ return queue_op_status::busy;
+ }
+ return try_pull_front(elem, lk);
+ }
template <typename ValueType>
void sync_bounded_queue<ValueType>::throw_if_closed(unique_lock<mutex>&)
@@ -379,6 +455,7 @@
}
}
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
void sync_bounded_queue<ValueType>::pull(ValueType& elem)
{
@@ -410,7 +487,6 @@
throw;
}
}
-
// enable if ValueType is nothrow movable
template <typename ValueType>
ValueType sync_bounded_queue<ValueType>::pull()
@@ -443,6 +519,42 @@
}
}
+#endif
+
+ template <typename ValueType>
+ void sync_bounded_queue<ValueType>::pull_front(ValueType& elem)
+ {
+ try
+ {
+ unique_lock<mutex> lk(mtx_);
+ wait_until_not_empty(lk);
+ pull_front(elem, lk);
+ }
+ catch (...)
+ {
+ close();
+ throw;
+ }
+ }
+
+ // enable if ValueType is nothrow movable
+ template <typename ValueType>
+ ValueType sync_bounded_queue<ValueType>::pull_front()
+ {
+ try
+ {
+ unique_lock<mutex> lk(mtx_);
+ wait_until_not_empty(lk);
+ return pull_front(lk);
+ }
+ catch (...)
+ {
+ close();
+ throw;
+ }
+ }
+
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_bounded_queue<ValueType>::try_push(const ValueType& elem, unique_lock<mutex>& lk)
{
@@ -455,7 +567,6 @@
push_at(elem, in_p_1, lk);
return true;
}
-
template <typename ValueType>
bool sync_bounded_queue<ValueType>::try_push(const ValueType& elem)
{
@@ -471,6 +582,29 @@
}
}
+#endif
+
+ template <typename ValueType>
+ queue_op_status sync_bounded_queue<ValueType>::try_push_back(const ValueType& elem, unique_lock<mutex>& lk)
+ {
+ if (closed(lk)) return queue_op_status::closed;
+ size_type in_p_1 = inc(in_);
+ if (in_p_1 == out_) // full()
+ {
+ return queue_op_status::full;
+ }
+ push_at(elem, in_p_1, lk);
+ return queue_op_status::success;
+ }
+
+ template <typename ValueType>
+ queue_op_status sync_bounded_queue<ValueType>::try_push_back(const ValueType& elem)
+ {
+ unique_lock<mutex> lk(mtx_);
+ return try_push_back(elem, lk);
+ }
+
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_bounded_queue<ValueType>::try_push(no_block_tag, const ValueType& elem)
{
@@ -486,7 +620,15 @@
throw;
}
}
+#endif
+ template <typename ValueType>
+ queue_op_status sync_bounded_queue<ValueType>::nonblocking_push_back(const ValueType& elem)
+ {
+ unique_lock<mutex> lk(mtx_, try_to_lock);
+ if (!lk.owns_lock()) return queue_op_status::busy;
+ return try_push_back(elem, lk);
+ }
template <typename ValueType>
typename sync_bounded_queue<ValueType>::size_type sync_bounded_queue<ValueType>::wait_until_not_full(unique_lock<mutex>& lk)
@@ -504,6 +646,7 @@
}
}
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
void sync_bounded_queue<ValueType>::push(const ValueType& elem)
{
@@ -518,7 +661,23 @@
throw;
}
}
+#endif
+ template <typename ValueType>
+ void sync_bounded_queue<ValueType>::push_back(const ValueType& elem)
+ {
+ try
+ {
+ unique_lock<mutex> lk(mtx_);
+ push_at(elem, wait_until_not_full(lk), lk);
+ }
+ catch (...)
+ {
+ close();
+ throw;
+ }
+ }
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_bounded_queue<ValueType>::try_push(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
{
@@ -546,8 +705,29 @@
throw;
}
}
+#endif
template <typename ValueType>
+ queue_op_status sync_bounded_queue<ValueType>::try_push_back(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
+ {
+ if (closed(lk)) return queue_op_status::closed;
+ size_type in_p_1 = inc(in_);
+ if (in_p_1 == out_) // full()
+ {
+ return queue_op_status::full;
+ }
+ push_at(boost::move(elem), in_p_1, lk);
+ return queue_op_status::success;
+ }
+ template <typename ValueType>
+ queue_op_status sync_bounded_queue<ValueType>::try_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
+ {
+ unique_lock<mutex> lk(mtx_);
+ return try_push_back(boost::move(elem), lk);
+ }
+
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
+ template <typename ValueType>
bool sync_bounded_queue<ValueType>::try_push(no_block_tag, BOOST_THREAD_RV_REF(ValueType) elem)
{
try
@@ -565,7 +745,19 @@
throw;
}
}
+#endif
+ template <typename ValueType>
+ queue_op_status sync_bounded_queue<ValueType>::nonblocking_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
+ {
+ unique_lock<mutex> lk(mtx_, try_to_lock);
+ if (!lk.owns_lock())
+ {
+ return queue_op_status::busy;
+ }
+ return try_push_back(boost::move(elem), lk);
+ }
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
void sync_bounded_queue<ValueType>::push(BOOST_THREAD_RV_REF(ValueType) elem)
{
@@ -580,25 +772,40 @@
throw;
}
}
+#endif
+ template <typename ValueType>
+ void sync_bounded_queue<ValueType>::push_back(BOOST_THREAD_RV_REF(ValueType) elem)
+ {
+ try
+ {
+ unique_lock<mutex> lk(mtx_);
+ push_at(boost::move(elem), wait_until_not_full(lk), lk);
+ }
+ catch (...)
+ {
+ close();
+ throw;
+ }
+ }
template <typename ValueType>
sync_bounded_queue<ValueType>& operator<<(sync_bounded_queue<ValueType>& sbq, BOOST_THREAD_RV_REF(ValueType) elem)
{
- sbq.push(boost::move(elem));
+ sbq.push_back(boost::move(elem));
return sbq;
}
template <typename ValueType>
sync_bounded_queue<ValueType>& operator<<(sync_bounded_queue<ValueType>& sbq, ValueType const&elem)
{
- sbq.push(elem);
+ sbq.push_back(elem);
return sbq;
}
template <typename ValueType>
sync_bounded_queue<ValueType>& operator>>(sync_bounded_queue<ValueType>& sbq, ValueType &elem)
{
- sbq.pull(elem);
+ sbq.pull_front(elem);
return sbq;
}
Modified: trunk/boost/thread/sync_queue.hpp
==============================================================================
--- trunk/boost/thread/sync_queue.hpp Sun Nov 10 23:11:13 2013 (r86615)
+++ trunk/boost/thread/sync_queue.hpp 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013) (r86616)
@@ -27,7 +27,6 @@
namespace boost
{
-
template <typename ValueType>
class sync_queue
{
@@ -52,23 +51,43 @@
// Modifiers
inline void close();
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline void push(const value_type& x);
inline bool try_push(const value_type& x);
inline bool try_push(no_block_tag, const value_type& x);
-
inline void push(BOOST_THREAD_RV_REF(value_type) x);
inline bool try_push(BOOST_THREAD_RV_REF(value_type) x);
inline bool try_push(no_block_tag, BOOST_THREAD_RV_REF(value_type) x);
+#endif
+ inline void push_back(const value_type& x);
+ inline queue_op_status try_push_back(const value_type& x);
+ inline queue_op_status nonblocking_push_back(const value_type& x);
+ inline void push_back(BOOST_THREAD_RV_REF(value_type) x);
+ inline queue_op_status try_push_back(BOOST_THREAD_RV_REF(value_type) x);
+ inline queue_op_status nonblocking_push_back(BOOST_THREAD_RV_REF(value_type) x);
+
// Observers/Modifiers
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline void pull(value_type&);
inline void pull(ValueType& elem, bool & closed);
// enable_if is_nothrow_copy_movable<value_type>
inline value_type pull();
inline shared_ptr<ValueType> ptr_pull();
+#endif
+ inline void pull_front(value_type&);
+ inline void pull_front(ValueType& elem, bool & closed);
+ // enable_if is_nothrow_copy_movable<value_type>
+ inline value_type pull_front();
+
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline bool try_pull(value_type&);
inline bool try_pull(no_block_tag,value_type&);
inline shared_ptr<ValueType> try_pull();
+#endif
+ inline queue_op_status try_pull_front(value_type&);
+ inline queue_op_status nonblocking_pull_front(value_type&);
+
inline underlying_queue_type underlying_queue() {
lock_guard<mutex> lk(mtx_);
return boost::move(data_);
@@ -96,11 +115,17 @@
}
inline void throw_if_closed(unique_lock<mutex>&);
+ inline bool closed(unique_lock<mutex>& lk) const;
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline bool try_pull(value_type& x, unique_lock<mutex>& lk);
inline bool try_push(const value_type& x, unique_lock<mutex>& lk);
inline bool try_push(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
inline shared_ptr<value_type> try_pull(unique_lock<mutex>& lk);
+#endif
+ inline queue_op_status try_pull_front(value_type& x, unique_lock<mutex>& lk);
+ inline queue_op_status try_push_back(const value_type& x, unique_lock<mutex>& lk);
+ inline queue_op_status try_push_back(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
inline void wait_until_not_empty(unique_lock<mutex>& lk);
inline void wait_until_not_empty(unique_lock<mutex>& lk, bool&);
@@ -115,6 +140,7 @@
}
}
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline void pull(value_type& elem, unique_lock<mutex>& )
{
elem = boost::move(data_.front());
@@ -132,7 +158,20 @@
data_.pop_front();
return res;
}
+#endif
+ inline void pull_front(value_type& elem, unique_lock<mutex>& )
+ {
+ elem = boost::move(data_.front());
+ data_.pop_front();
+ }
+ inline value_type pull_front(unique_lock<mutex>& )
+ {
+ value_type e = boost::move(data_.front());
+ data_.pop_front();
+ return boost::move(e);
+ }
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
inline void push(const value_type& elem, unique_lock<mutex>& lk)
{
data_.push_back(elem);
@@ -144,6 +183,18 @@
data_.push_back(boost::move(elem));
notify_not_empty_if_needed(lk);
}
+#endif
+ inline void push_back(const value_type& elem, unique_lock<mutex>& lk)
+ {
+ data_.push_back(elem);
+ notify_not_empty_if_needed(lk);
+ }
+
+ inline void push_back(BOOST_THREAD_RV_REF(value_type) elem, unique_lock<mutex>& lk)
+ {
+ data_.push_back(boost::move(elem));
+ notify_not_empty_if_needed(lk);
+ }
};
template <typename ValueType>
@@ -196,6 +247,11 @@
lock_guard<mutex> lk(mtx_);
return closed_;
}
+ template <typename ValueType>
+ bool sync_queue<ValueType>::closed(unique_lock<mutex>&) const
+ {
+ return closed_;
+ }
template <typename ValueType>
bool sync_queue<ValueType>::empty() const
@@ -217,6 +273,7 @@
}
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_queue<ValueType>::try_pull(ValueType& elem, unique_lock<mutex>& lk)
{
@@ -238,7 +295,20 @@
}
return ptr_pull(lk);
}
+#endif
+ template <typename ValueType>
+ queue_op_status sync_queue<ValueType>::try_pull_front(ValueType& elem, unique_lock<mutex>& lk)
+ {
+ if (empty(lk))
+ {
+ if (closed(lk)) return queue_op_status::closed;
+ return queue_op_status::empty;
+ }
+ pull_front(elem, lk);
+ return queue_op_status::success;
+ }
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_queue<ValueType>::try_pull(ValueType& elem)
{
@@ -253,7 +323,15 @@
throw;
}
}
+#endif
+ template <typename ValueType>
+ queue_op_status sync_queue<ValueType>::try_pull_front(ValueType& elem)
+ {
+ unique_lock<mutex> lk(mtx_);
+ return try_pull_front(elem, lk);
+ }
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_queue<ValueType>::try_pull(no_block_tag,ValueType& elem)
{
@@ -286,6 +364,17 @@
throw;
}
}
+#endif
+ template <typename ValueType>
+ queue_op_status sync_queue<ValueType>::nonblocking_pull_front(ValueType& elem)
+ {
+ unique_lock<mutex> lk(mtx_, try_to_lock);
+ if (!lk.owns_lock())
+ {
+ return queue_op_status::busy;
+ }
+ return try_pull_front(elem, lk);
+ }
template <typename ValueType>
void sync_queue<ValueType>::throw_if_closed(unique_lock<mutex>&)
@@ -320,6 +409,7 @@
closed=false;
}
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
void sync_queue<ValueType>::pull(ValueType& elem)
{
@@ -383,7 +473,42 @@
throw;
}
}
+#endif
+
+ template <typename ValueType>
+ void sync_queue<ValueType>::pull_front(ValueType& elem)
+ {
+ try
+ {
+ unique_lock<mutex> lk(mtx_);
+ wait_until_not_empty(lk);
+ pull_front(elem, lk);
+ }
+ catch (...)
+ {
+ close();
+ throw;
+ }
+ }
+
+ // enable if ValueType is nothrow movable
+ template <typename ValueType>
+ ValueType sync_queue<ValueType>::pull_front()
+ {
+ try
+ {
+ unique_lock<mutex> lk(mtx_);
+ wait_until_not_empty(lk);
+ return pull_front(lk);
+ }
+ catch (...)
+ {
+ close();
+ throw;
+ }
+ }
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_queue<ValueType>::try_push(const ValueType& elem, unique_lock<mutex>& lk)
{
@@ -406,8 +531,25 @@
throw;
}
}
+#endif
template <typename ValueType>
+ queue_op_status sync_queue<ValueType>::try_push_back(const ValueType& elem, unique_lock<mutex>& lk)
+ {
+ if (closed(lk)) return queue_op_status::closed;
+ push_back(elem, lk);
+ return queue_op_status::success;
+ }
+
+ template <typename ValueType>
+ queue_op_status sync_queue<ValueType>::try_push_back(const ValueType& elem)
+ {
+ unique_lock<mutex> lk(mtx_);
+ return try_push_back(elem, lk);
+ }
+
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
+ template <typename ValueType>
bool sync_queue<ValueType>::try_push(no_block_tag, const ValueType& elem)
{
try
@@ -422,7 +564,16 @@
throw;
}
}
+#endif
+ template <typename ValueType>
+ queue_op_status sync_queue<ValueType>::nonblocking_push_back(const ValueType& elem)
+ {
+ unique_lock<mutex> lk(mtx_, try_to_lock);
+ if (!lk.owns_lock()) return queue_op_status::busy;
+ return try_push(elem, lk);
+ }
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
void sync_queue<ValueType>::push(const ValueType& elem)
{
@@ -438,7 +589,25 @@
throw;
}
}
+#endif
+
+ template <typename ValueType>
+ void sync_queue<ValueType>::push_back(const ValueType& elem)
+ {
+ try
+ {
+ unique_lock<mutex> lk(mtx_);
+ throw_if_closed(lk);
+ push_back(elem, lk);
+ }
+ catch (...)
+ {
+ close();
+ throw;
+ }
+ }
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
template <typename ValueType>
bool sync_queue<ValueType>::try_push(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
{
@@ -461,8 +630,25 @@
throw;
}
}
+#endif
+
+ template <typename ValueType>
+ queue_op_status sync_queue<ValueType>::try_push_back(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
+ {
+ if (closed(lk)) return queue_op_status::closed;
+ push_back(boost::move(elem), lk);
+ return queue_op_status::success;
+ }
template <typename ValueType>
+ queue_op_status sync_queue<ValueType>::try_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
+ {
+ unique_lock<mutex> lk(mtx_);
+ return try_push_back(boost::move(elem), lk);
+ }
+
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
+ template <typename ValueType>
bool sync_queue<ValueType>::try_push(no_block_tag, BOOST_THREAD_RV_REF(ValueType) elem)
{
try
@@ -480,8 +666,21 @@
throw;
}
}
+#endif
template <typename ValueType>
+ queue_op_status sync_queue<ValueType>::nonblocking_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
+ {
+ unique_lock<mutex> lk(mtx_, try_to_lock);
+ if (!lk.owns_lock())
+ {
+ return queue_op_status::busy;
+ }
+ return try_push_back(boost::move(elem), lk);
+ }
+
+#ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
+ template <typename ValueType>
void sync_queue<ValueType>::push(BOOST_THREAD_RV_REF(ValueType) elem)
{
try
@@ -496,25 +695,42 @@
throw;
}
}
+#endif
+
+ template <typename ValueType>
+ void sync_queue<ValueType>::push_back(BOOST_THREAD_RV_REF(ValueType) elem)
+ {
+ try
+ {
+ unique_lock<mutex> lk(mtx_);
+ throw_if_closed(lk);
+ push_back(boost::move(elem), lk);
+ }
+ catch (...)
+ {
+ close();
+ throw;
+ }
+ }
template <typename ValueType>
sync_queue<ValueType>& operator<<(sync_queue<ValueType>& sbq, BOOST_THREAD_RV_REF(ValueType) elem)
{
- sbq.push(boost::move(elem));
+ sbq.push_back(boost::move(elem));
return sbq;
}
template <typename ValueType>
sync_queue<ValueType>& operator<<(sync_queue<ValueType>& sbq, ValueType const&elem)
{
- sbq.push(elem);
+ sbq.push_back(elem);
return sbq;
}
template <typename ValueType>
sync_queue<ValueType>& operator>>(sync_queue<ValueType>& sbq, ValueType &elem)
{
- sbq.pull(elem);
+ sbq.pull_front(elem);
return sbq;
}
Modified: trunk/boost/thread/thread_pool.hpp
==============================================================================
--- trunk/boost/thread/thread_pool.hpp Sun Nov 10 23:11:13 2013 (r86615)
+++ trunk/boost/thread/thread_pool.hpp 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013) (r86616)
@@ -48,7 +48,7 @@
work task;
try
{
- if (work_queue.try_pull(task))
+ if (work_queue.try_pull_front(task) == queue_op_status::success)
{
task();
return true;
@@ -167,23 +167,23 @@
void submit(Closure & closure)
{
work w ((closure));
- work_queue.push(boost::move(w));
+ work_queue.push_back(boost::move(w));
//work_queue.push(work(closure)); // todo check why this doesn't work
}
#endif
void submit(void (*closure)())
{
work w ((closure));
- work_queue.push(boost::move(w));
- //work_queue.push(work(closure)); // todo check why this doesn't work
+ work_queue.push_back(boost::move(w));
+ //work_queue.push_back(work(closure)); // todo check why this doesn't work
}
template <typename Closure>
void submit(BOOST_THREAD_RV_REF(Closure) closure)
{
work w =boost::move(closure);
- work_queue.push(boost::move(w));
- //work_queue.push(work(boost::move(closure))); // todo check why this doesn't work
+ work_queue.push_back(boost::move(w));
+ //work_queue.push_back(work(boost::move(closure))); // todo check why this doesn't work
}
/**
Modified: trunk/boost/thread/user_scheduler.hpp
==============================================================================
--- trunk/boost/thread/user_scheduler.hpp Sun Nov 10 23:11:13 2013 (r86615)
+++ trunk/boost/thread/user_scheduler.hpp 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013) (r86616)
@@ -39,7 +39,7 @@
work task;
try
{
- if (work_queue.try_pull(task))
+ if (work_queue.try_pull_front(task) == queue_op_status::success)
{
task();
return true;
@@ -144,23 +144,23 @@
void submit(Closure & closure)
{
work w ((closure));
- work_queue.push(boost::move(w));
+ work_queue.push_back(boost::move(w));
//work_queue.push(work(closure)); // todo check why this doesn't work
}
#endif
void submit(void (*closure)())
{
work w ((closure));
- work_queue.push(boost::move(w));
- //work_queue.push(work(closure)); // todo check why this doesn't work
+ work_queue.push_back(boost::move(w));
+ //work_queue.push_back(work(closure)); // todo check why this doesn't work
}
template <typename Closure>
void submit(BOOST_THREAD_RV_REF(Closure) closure)
{
work w =boost::move(closure);
- work_queue.push(boost::move(w));
- //work_queue.push(work(boost::move(closure))); // todo check why this doesn't work
+ work_queue.push_back(boost::move(w));
+ //work_queue.push_back(work(boost::move(closure))); // todo check why this doesn't work
}
/**
Modified: trunk/libs/thread/example/executor.cpp
==============================================================================
--- trunk/libs/thread/example/executor.cpp Sun Nov 10 23:11:13 2013 (r86615)
+++ trunk/libs/thread/example/executor.cpp 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013) (r86616)
@@ -6,6 +6,7 @@
#define BOOST_THREAD_VERSION 4
#define BOOST_THREAD_PROVIDES_EXECUTORS
#define BOOST_THREAD_USES_LOG_THREAD_ID
+#define BOOST_THREAD_QUEUE_DEPRECATE_OLD
#include <boost/thread/thread_pool.hpp>
#include <boost/thread/user_scheduler.hpp>
Modified: trunk/libs/thread/example/future_fallback_to.cpp
==============================================================================
--- trunk/libs/thread/example/future_fallback_to.cpp Sun Nov 10 23:11:13 2013 (r86615)
+++ trunk/libs/thread/example/future_fallback_to.cpp 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013) (r86616)
@@ -42,6 +42,7 @@
BOOST_THREAD_LOG << "" << BOOST_THREAD_END_LOG;
boost::future<int> f2 = f1.fallback_to(-1);
BOOST_THREAD_LOG << "" << BOOST_THREAD_END_LOG;
+ f2.wait();
BOOST_ASSERT(f2.get()==1);
BOOST_THREAD_LOG << "" << BOOST_THREAD_END_LOG;
}
@@ -68,6 +69,7 @@
BOOST_THREAD_LOG << "" << BOOST_THREAD_END_LOG;
boost::future<int> f2 = f1.fallback_to(-1);
BOOST_THREAD_LOG << "" << BOOST_THREAD_END_LOG;
+ f2.wait();
BOOST_ASSERT(f2.get()==-1);
BOOST_THREAD_LOG << "" << BOOST_THREAD_END_LOG;
}
Modified: trunk/libs/thread/example/producer_consumer.cpp
==============================================================================
--- trunk/libs/thread/example/producer_consumer.cpp Sun Nov 10 23:11:13 2013 (r86615)
+++ trunk/libs/thread/example/producer_consumer.cpp 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013) (r86616)
@@ -7,21 +7,28 @@
// adapted from the example given by Howard Hinnant in
#define BOOST_THREAD_VERSION 4
+#define BOOST_THREAD_QUEUE_DEPRECATE_OLD
#include <iostream>
#include <boost/thread/scoped_thread.hpp>
+#ifdef XXXX
#include <boost/thread/externally_locked_stream.hpp>
+ typedef boost::externally_locked_stream<std::ostream> the_ostream;
+#else
+ typedef std::ostream the_ostream;
+ typedef std::istream the_istream;
+#endif
#include <boost/thread/sync_queue.hpp>
-void producer(boost::externally_locked_stream<std::ostream> &mos, boost::sync_queue<int> & sbq)
+void producer(the_ostream &mos, boost::sync_queue<int> & sbq)
{
using namespace boost;
try {
for(int i=0; ;++i)
{
- //sbq.push(i);
- sbq << i;
- mos << "push(" << i << ") "<< sbq.size()<<"\n";
+ sbq.push_back(i);
+ //sbq << i;
+ mos << "push_back(" << i << ") "<< sbq.size()<<"\n";
this_thread::sleep_for(chrono::milliseconds(200));
}
}
@@ -35,16 +42,19 @@
}
}
-void consumer(boost::externally_locked_stream<std::ostream> &mos, boost::sync_queue<int> & sbq)
+void consumer(
+ the_ostream &mos,
+ boost::sync_queue<int> & sbq)
{
using namespace boost;
try {
for(int i=0; ;++i)
{
int r;
- //sbq.pull(r);
- sbq >> r;
+ sbq.pull_front(r);
+ //sbq >> r;
mos << i << " pull(" << r << ") "<< sbq.size()<<"\n";
+
this_thread::sleep_for(chrono::milliseconds(250));
}
}
@@ -57,17 +67,18 @@
mos << "exception !!!\n";
}
}
-void consumer2(boost::externally_locked_stream<std::ostream> &mos, boost::sync_queue<int> & sbq)
+void consumer2(the_ostream &mos, boost::sync_queue<int> & sbq)
{
using namespace boost;
try {
- bool closed=false;
for(int i=0; ;++i)
{
int r;
- sbq.pull(r, closed);
- if (closed) break;
- mos << i << " pull(" << r << ")\n";
+ queue_op_status st = sbq.try_pull_front(r);
+ if (queue_op_status::closed == st) break;
+ if (queue_op_status::success == st) {
+ mos << i << " pull(" << r << ")\n";
+ }
this_thread::sleep_for(chrono::milliseconds(250));
}
}
@@ -76,7 +87,7 @@
mos << "exception !!!\n";
}
}
-//void consumer3(boost::externally_locked_stream<std::ostream> &mos, boost::sync_queue<int> & sbq)
+//void consumer3(the_ostream &mos, boost::sync_queue<int> & sbq)
//{
// using namespace boost;
// bool closed=false;
@@ -100,11 +111,17 @@
{
using namespace boost;
+#ifdef XXXX
recursive_mutex terminal_mutex;
externally_locked_stream<std::ostream> mcerr(std::cerr, terminal_mutex);
externally_locked_stream<std::ostream> mcout(std::cout, terminal_mutex);
externally_locked_stream<std::istream> mcin(std::cin, terminal_mutex);
+#else
+ the_ostream &mcerr = std::cout;
+ the_ostream &mcout = std::cerr;
+ //the_istream &mcin = std::cin;
+#endif
sync_queue<int> sbq;
Modified: trunk/libs/thread/example/producer_consumer_bounded.cpp
==============================================================================
--- trunk/libs/thread/example/producer_consumer_bounded.cpp Sun Nov 10 23:11:13 2013 (r86615)
+++ trunk/libs/thread/example/producer_consumer_bounded.cpp 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013) (r86616)
@@ -7,21 +7,29 @@
// adapted from the example given by Howard Hinnant in
#define BOOST_THREAD_VERSION 4
+#define BOOST_THREAD_QUEUE_DEPRECATE_OLD
#include <iostream>
#include <boost/thread/scoped_thread.hpp>
+#ifdef XXXX
#include <boost/thread/externally_locked_stream.hpp>
+ typedef boost::externally_locked_stream<std::ostream> the_ostream;
+#else
+ typedef std::ostream the_ostream;
+ typedef std::istream the_istream;
+#endif
+
#include <boost/thread/sync_bounded_queue.hpp>
-void producer(boost::externally_locked_stream<std::ostream> &mos, boost::sync_bounded_queue<int> & sbq)
+void producer(the_ostream &mos, boost::sync_bounded_queue<int> & sbq)
{
using namespace boost;
try {
for(int i=0; ;++i)
{
- //sbq.push(i);
- sbq << i;
- mos << "push(" << i << ") "<< sbq.size()<<"\n";
+ sbq.push_back(i);
+ //sbq << i;
+ mos << "push_back(" << i << ") "<< sbq.size()<<"\n";
this_thread::sleep_for(chrono::milliseconds(200));
}
}
@@ -35,16 +43,16 @@
}
}
-void consumer(boost::externally_locked_stream<std::ostream> &mos, boost::sync_bounded_queue<int> & sbq)
+void consumer(the_ostream &mos, boost::sync_bounded_queue<int> & sbq)
{
using namespace boost;
try {
for(int i=0; ;++i)
{
int r;
- //sbq.pull(r);
- sbq >> r;
- mos << i << " pull(" << r << ") "<< sbq.size()<<"\n";
+ sbq.pull_front(r);
+ //sbq >> r;
+ mos << i << " pull_front(" << r << ") "<< sbq.size()<<"\n";
this_thread::sleep_for(chrono::milliseconds(250));
}
}
@@ -57,17 +65,18 @@
mos << "exception !!!\n";
}
}
-void consumer2(boost::externally_locked_stream<std::ostream> &mos, boost::sync_bounded_queue<int> & sbq)
+void consumer2(the_ostream &mos, boost::sync_bounded_queue<int> & sbq)
{
using namespace boost;
try {
- bool closed=false;
for(int i=0; ;++i)
{
int r;
- sbq.pull(r, closed);
- if (closed) break;
- mos << i << " pull(" << r << ")\n";
+ queue_op_status st = sbq.try_pull_front(r);
+ if (queue_op_status::closed == st) break;
+ if (queue_op_status::success == st) {
+ mos << i << " pull(" << r << ")\n";
+ }
this_thread::sleep_for(chrono::milliseconds(250));
}
}
@@ -76,7 +85,7 @@
mos << "exception !!!\n";
}
}
-//void consumer3(boost::externally_locked_stream<std::ostream> &mos, boost::sync_bounded_queue<int> & sbq)
+//void consumer3(the_ostream &mos, boost::sync_bounded_queue<int> & sbq)
//{
// using namespace boost;
// bool closed=false;
@@ -99,12 +108,17 @@
int main()
{
using namespace boost;
-
+#ifdef XXXX
recursive_mutex terminal_mutex;
externally_locked_stream<std::ostream> mcerr(std::cerr, terminal_mutex);
externally_locked_stream<std::ostream> mcout(std::cout, terminal_mutex);
externally_locked_stream<std::istream> mcin(std::cin, terminal_mutex);
+#else
+ the_ostream &mcerr = std::cout;
+ the_ostream &mcout = std::cerr;
+ //the_istream &mcin = std::cin;
+#endif
sync_bounded_queue<int> sbq(10);
Modified: trunk/libs/thread/example/thread_pool.cpp
==============================================================================
--- trunk/libs/thread/example/thread_pool.cpp Sun Nov 10 23:11:13 2013 (r86615)
+++ trunk/libs/thread/example/thread_pool.cpp 2013-11-11 05:43:15 EST (Mon, 11 Nov 2013) (r86616)
@@ -6,6 +6,7 @@
#define BOOST_THREAD_VERSION 4
#define BOOST_THREAD_USES_LOG
#define BOOST_THREAD_USES_LOG_THREAD_ID
+#define BOOST_THREAD_QUEUE_DEPRECATE_OLD
#include <boost/thread/detail/log.hpp>
#include <boost/thread/thread_pool.hpp>