Subject: [Boost-commit] svn:boost r56542 - in sandbox/task: boost boost/task boost/task/detail libs/task/build libs/task/examples libs/task/src
From: oliver.kowalke_at_[hidden]
Date: 2009-10-03 05:04:50
Author: olli
Date: 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
New Revision: 56542
URL: http://svn.boost.org/trac/boost/changeset/56542
Log:
- spin_lock, spin_mutex, spin_condition added
- unbounded_buffer uses spinwait-objects
- examples buffer_multi and buffer_multi2 enhanced (using buffer synchonization)
Added:
sandbox/task/boost/task/spin_condition.hpp (contents, props changed)
sandbox/task/boost/task/spin_lock.hpp (contents, props changed)
sandbox/task/boost/task/spin_mutex.hpp (contents, props changed)
sandbox/task/libs/task/src/spin_condition.cpp (contents, props changed)
sandbox/task/libs/task/src/spin_mutex.cpp (contents, props changed)
Text files modified:
sandbox/task/boost/task.hpp | 3
sandbox/task/boost/task/detail/atomic_aix.hpp | 4
sandbox/task/boost/task/detail/atomic_gcc.hpp | 13 ++
sandbox/task/boost/task/detail/atomic_gcc_ppc.hpp | 26 +++++
sandbox/task/boost/task/detail/atomic_gcc_x86.hpp | 17 +++
sandbox/task/boost/task/detail/atomic_hpux.hpp | 12 ++
sandbox/task/boost/task/detail/atomic_interlocked.hpp | 12 ++
sandbox/task/boost/task/detail/atomic_interprocess.hpp | 18 +++
sandbox/task/boost/task/detail/atomic_solaris.hpp | 12 ++
sandbox/task/boost/task/detail/atomic_sync.hpp | 12 ++
sandbox/task/boost/task/exceptions.hpp | 16 ++
sandbox/task/boost/task/unbounded_buffer.hpp | 187 ++++++++++++++++++++++++++++++++-------
sandbox/task/libs/task/build/Jamfile.v2 | 4
sandbox/task/libs/task/examples/buffer_multi.cpp | 39 +++-----
sandbox/task/libs/task/examples/buffer_multi2.cpp | 37 +++----
sandbox/task/libs/task/examples/no_deadlock_pool2.cpp | 4
16 files changed, 324 insertions(+), 92 deletions(-)
Modified: sandbox/task/boost/task.hpp
==============================================================================
--- sandbox/task/boost/task.hpp (original)
+++ sandbox/task/boost/task.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -25,6 +25,9 @@
#include <boost/task/poolsize.hpp>
#include <boost/task/scanns.hpp>
#include <boost/task/semaphore.hpp>
+#include <boost/task/spin_condition.hpp>
+#include <boost/task/spin_lock.hpp>
+#include <boost/task/spin_mutex.hpp>
#include <boost/task/stacksize.hpp>
#include <boost/task/static_pool.hpp>
#include <boost/task/task.hpp>
Modified: sandbox/task/boost/task/detail/atomic_aix.hpp
==============================================================================
--- sandbox/task/boost/task/detail/atomic_aix.hpp (original)
+++ sandbox/task/boost/task/detail/atomic_aix.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -26,6 +26,10 @@
{ * object = desired; }
inline
+bool atomic_compare_exchange_strong( uint32_t volatile * object, uint32_t * expected, uint32_t desired)
+{ return ::compare_and_swap( object, expected, desired); }
+
+inline
uint32_t atomic_fetch_add( uint32_t volatile * object, uint32_t operand)
{
BOOST_ASSERT( operand == 1);
Modified: sandbox/task/boost/task/detail/atomic_gcc.hpp
==============================================================================
--- sandbox/task/boost/task/detail/atomic_gcc.hpp (original)
+++ sandbox/task/boost/task/detail/atomic_gcc.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -17,6 +17,7 @@
#include <boost/assert.hpp>
#include <boost/cstdint.hpp>
+#include <boost/interprocess/detail/atomic.hpp>
#include <boost/config/abi_prefix.hpp>
@@ -40,6 +41,18 @@
}
inline
+bool atomic_compare_exchange_strong( uint32_t volatile * object, uint32_t * expected, uint32_t desired)
+{
+ uint32_t prev = interprocess::detail::atomic_cas32( object, desired, * expected);
+ if ( prev != * expected)
+ {
+ * expected = prev;
+ return false;
+ }
+ return true;
+}
+
+inline
uint32_t atomic_fetch_add( uint32_t volatile * object, uint32_t operand)
{
BOOST_ASSERT( operand == 1);
Modified: sandbox/task/boost/task/detail/atomic_gcc_ppc.hpp
==============================================================================
--- sandbox/task/boost/task/detail/atomic_gcc_ppc.hpp (original)
+++ sandbox/task/boost/task/detail/atomic_gcc_ppc.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -33,6 +33,32 @@
}
inline
+bool atomic_compare_exchange_strong( uint32_t volatile * object, uint32_t * expected, uint32_t desired)
+{
+ uint32_t prev = * expected;
+
+ __asm__ __volatile__
+ (
+ "0:\n\t"
+ "lwarx %0,0,%1\n\t"
+ "cmpw %0,%3\n\t"
+ "bne- 1f\n\t"
+ "stwcx. %2,0,%1\n\t"
+ "bne- 0b\n\t"
+ "1:"
+ : "=&r"( * expected)
+ : "b" ( object), "r" ( desired), "r" ( * expected)
+ : "memory", "cc"
+ );
+ if ( prev != * expected)
+ {
+ * expected = prev;
+ return false;
+ }
+ return true;
+}
+
+inline
uint32_t atomic_fetch_add( uint32_t volatile * object, uint32_t operand)
{
int object_ = static_cast< int >( object);
Modified: sandbox/task/boost/task/detail/atomic_gcc_x86.hpp
==============================================================================
--- sandbox/task/boost/task/detail/atomic_gcc_x86.hpp (original)
+++ sandbox/task/boost/task/detail/atomic_gcc_x86.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -27,6 +27,23 @@
}
inline
+bool atomic_compare_exchange_strong( uint32_t volatile * object, uint32_t * expected, uint32_t desired)
+{
+ uint32_t prev = * expected;
+
+ __asm__ __volatile__
+ (
+ "lock\n\t"
+ "cmpxchg %3, %1"
+ : "=a" ( * expected), "=m" ( * object)
+ : "a" ( prev), "r" ( desired)
+ : "memory", "cc"
+ );
+
+ return prev == * expected;
+}
+
+inline
long atomic_fetch_add( uint32_t volatile * object, uint32_t operand)
{
int operand_ = static_cast< int >( operand);
Modified: sandbox/task/boost/task/detail/atomic_hpux.hpp
==============================================================================
--- sandbox/task/boost/task/detail/atomic_hpux.hpp (original)
+++ sandbox/task/boost/task/detail/atomic_hpux.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -25,6 +25,18 @@
{ * object = desired; }
inline
+bool atomic_compare_exchange_strong( uint32_t volatile * object, uint32_t * expected, uint32_t desired)
+{
+ uint32_t prev = ::atomic_cas_32( object, * expected, desired);
+ if ( prev != * expected)
+ {
+ * expected = prev;
+ return false;
+ }
+ return true;
+}
+
+inline
uint32_t atomic_fetch_add( uint32_t volatile * object, uint32_t operand)
{
BOOST_ASSERT( operand == 1);
Modified: sandbox/task/boost/task/detail/atomic_interlocked.hpp
==============================================================================
--- sandbox/task/boost/task/detail/atomic_interlocked.hpp (original)
+++ sandbox/task/boost/task/detail/atomic_interlocked.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -22,6 +22,18 @@
{ BOOST_INTERLOCKED_EXCHANGE( reinterpret_cast< long volatile * >( object), desired); }
inline
+bool atomic_compare_exchange_strong( uint32_t volatile * object, uint32_t * expected, uint32_t desired)
+{
+ uint32_t prev = BOOST_INTERLOCKED_COMPARE_EXCHANGE( object, desired, * expected);
+ if ( prev != * expected)
+ {
+ * expected = prev;
+ return false;
+ }
+ return true;
+}
+
+inline
uint32_t atomic_fetch_add( uint32_t volatile * object, uint32_t operand)
{
BOOST_ASSERT( operand == 1);
Modified: sandbox/task/boost/task/detail/atomic_interprocess.hpp
==============================================================================
--- sandbox/task/boost/task/detail/atomic_interprocess.hpp (original)
+++ sandbox/task/boost/task/detail/atomic_interprocess.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -18,18 +18,30 @@
namespace detail
{
inline
-void atomic_exchange( volatile uint32_t * object, uint32_t desired)
+void atomic_exchange( uint32_t volatile * object, uint32_t desired)
{ interprocess::detail::atomic_write32( object, desired); }
inline
-unsigned int atomic_fetch_add( volatile uint32_t * object, uint32_t operand)
+bool atomic_compare_exchange_strong( uint32_t volatile * object, uint32_t * expected, uint32_t desired)
+{
+ uint32_t prev = interprocess::detail::atomic_cas32( object, desired, * expected);
+ if ( prev != * expected)
+ {
+ * expected = prev;
+ return false;
+ }
+ return true;
+}
+
+inline
+unsigned int atomic_fetch_add( uint32_t volatile * object, uint32_t operand)
{
BOOST_ASSERT( operand == 1);
return interprocess::detail::atomic_inc32( object);
}
inline
-unsigned int atomic_fetch_sub( volatile uint32_t * object, uint32_t operand)
+unsigned int atomic_fetch_sub( uint32_t volatile * object, uint32_t operand)
{
BOOST_ASSERT( operand == 1);
return interprocess::detail::atomic_dec32( object);
Modified: sandbox/task/boost/task/detail/atomic_solaris.hpp
==============================================================================
--- sandbox/task/boost/task/detail/atomic_solaris.hpp (original)
+++ sandbox/task/boost/task/detail/atomic_solaris.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -26,6 +26,18 @@
{ * object = desired; }
inline
+bool atomic_compare_exchange_strong( uint32_t volatile * object, uint32_t * expected, uint32_t desired)
+{
+ uint32_t prev = ::atomic_cas_32( object, * expected, desired);
+ if ( prev != * expected)
+ {
+ * expected = prev;
+ return false;
+ }
+ return true;
+}
+
+inline
uint32_t atomic_fetch_add( uint32_t volatile * object, uint32_t operand)
{
BOOST_ASSERT( operand == 1);
Modified: sandbox/task/boost/task/detail/atomic_sync.hpp
==============================================================================
--- sandbox/task/boost/task/detail/atomic_sync.hpp (original)
+++ sandbox/task/boost/task/detail/atomic_sync.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -30,6 +30,18 @@
{ * object = desired; }
inline
+bool atomic_compare_exchange_strong( uint32_t volatile * object, uint32_t * expected, uint32_t desired)
+{
+ uint32_t prev = __sync_val_compare_and_swap( object, * expected, desired);
+ if ( prev != * expected)
+ {
+ * expected = prev;
+ return false;
+ }
+ return true;
+}
+
+inline
uint32_t atomic_fetch_add( uint32_t volatile * object, uint32_t operand)
{
BOOST_ASSERT( operand == 1);
Modified: sandbox/task/boost/task/exceptions.hpp
==============================================================================
--- sandbox/task/boost/task/exceptions.hpp (original)
+++ sandbox/task/boost/task/exceptions.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -12,8 +12,9 @@
#include <boost/config/abi_prefix.hpp>
-namespace boost { namespace task
-{
+namespace boost {
+namespace task {
+
class invalid_poolsize : public std::invalid_argument
{
public:
@@ -104,7 +105,16 @@
: std::logic_error("pool moved")
{}
};
-} }
+
+class lock_error : public std::logic_error
+{
+public:
+ lock_error()
+ : std::logic_error("lock invalid")
+ {}
+};
+
+}}
#include <boost/config/abi_suffix.hpp>
Added: sandbox/task/boost/task/spin_condition.hpp
==============================================================================
--- (empty file)
+++ sandbox/task/boost/task/spin_condition.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -0,0 +1,123 @@
+
+// Copyright Oliver Kowalke 2009.
+// 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_TASK_SPIN_CONDITION_H
+#define BOOST_TASK_SPIN_CONDITION_H
+
+#include <boost/cstdint.hpp>
+#include <boost/thread/thread_time.hpp>
+#include <boost/utility.hpp>
+
+#include <boost/task/exceptions.hpp>
+#include <boost/task/spin_mutex.hpp>
+
+namespace boost {
+namespace task {
+
+class spin_condition : private noncopyable
+{
+private:
+ enum command
+ {
+ SLEEPING = 0,
+ NOTIFY_ONE,
+ NOTIFY_ALL
+ };
+
+ volatile uint32_t cmd_;
+ volatile uint32_t waiters_;
+ spin_mutex enter_mtx_;
+ spin_mutex check_mtx_;
+
+ spin_condition( spin_condition const&);
+ spin_condition & operator=( spin_condition const&);
+
+ void wait_( spin_mutex &);
+ bool wait_( spin_mutex &, system_time const&);
+ void notify_( command);
+
+public:
+ spin_condition();
+
+ void notify_one();
+
+ void notify_all();
+
+ template< typename Lock >
+ void wait( Lock & lk)
+ {
+ if ( ! lk)
+ throw lock_error();
+ wait_( * lk.mutex() );
+ }
+
+ template<
+ typename Lock,
+ typename Pred
+ >
+ void wait( Lock & lk, Pred pred)
+ {
+ if ( ! lk)
+ throw lock_error();
+
+ while ( ! pred() )
+ wait_( * lk.mutex() );
+ }
+
+ template< typename Lock >
+ bool timed_wait( Lock & lk, system_time const& abs_time)
+ {
+ if ( abs_time.is_infinity() )
+ {
+ wait_( lk);
+ return true;
+ }
+
+ if ( ! lk)
+ throw lock_error();
+ return wait_( * lk.mutex(), abs_time);
+ }
+
+ template<
+ typename Lock,
+ typename Pred
+ >
+ bool timed_wait( Lock & lk, system_time const& abs_time, Pred pred)
+ {
+ if ( abs_time.is_infinity() )
+ {
+ wait_( lk, pred);
+ return true;
+ }
+
+ if ( ! lk)
+ throw lock_error();
+
+ while ( ! pred() )
+ if ( ! wait_( * lk.mutex(), abs_time) )
+ return pred();
+ return true;
+ }
+
+ template<
+ typename Lock,
+ typename TimeDuration
+ >
+ bool timed_wait( Lock & lk, TimeDuration const& rel_time)
+ { return timed_wait( lk, get_system_time() + rel_time); }
+
+ template<
+ typename Lock,
+ typename TimeDuration,
+ typename Pred
+ >
+ bool timed_wait( Lock & lk, TimeDuration const& rel_time, Pred pred)
+ { return timed_wait( lk, get_system_time() + rel_time, pred); }
+};
+
+}}
+
+#endif // BOOST_TASK_SPIN_CONDITION_H
Added: sandbox/task/boost/task/spin_lock.hpp
==============================================================================
--- (empty file)
+++ sandbox/task/boost/task/spin_lock.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -0,0 +1,112 @@
+
+// Copyright Oliver Kowalke 2009.
+// 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_TASK_SPIN_LOCK_H
+#define BOOST_TASK_SPIN_LOCK_H
+
+#include <algorithm>
+
+#include <boost/thread/thread_time.hpp>
+
+#include <boost/task/exceptions.hpp>
+
+namespace boost {
+namespace task {
+
+template< typename Mutex >
+class spin_lock
+{
+private:
+ typedef bool spin_lock::*unspecified_bool_type;
+
+ Mutex * mtx_;
+ bool locked_;
+
+ spin_lock( spin_lock &);
+ spin_lock & operator=( spin_lock &);
+
+public:
+ explicit spin_lock( Mutex & mtx)
+ : mtx_( & mtx), locked_( false)
+ {
+ mtx_->lock();
+ locked_ = true;
+ }
+
+ spin_lock( Mutex & mtx, system_time const& abs_time)
+ : mtx_( & mtx), locked_( mtx_->timed_lock( abs_time) )
+ {}
+
+ template< typename TimeDuration >
+ spin_lock( Mutex & mtx, TimeDuration const& rel_time)
+ : mtx_( & mtx), locked_( mtx_->timed_lock( rel_time) )
+ {}
+
+ ~spin_lock()
+ {
+ try
+ { if ( locked_ && mtx_) mtx_->unlock(); }
+ catch (...) {}
+ }
+
+ void lock()
+ {
+ if ( ! mtx_ || locked_)
+ throw lock_error();
+ mtx_->lock();
+ locked_ = true;
+ }
+
+ bool try_lock()
+ {
+ if ( ! mtx_ || locked_)
+ throw lock_error();
+ locked_ = mtx_->try_lock();
+ return locked_;
+ }
+
+ bool timed_lock( system_time const& abs_time)
+ {
+ if ( ! mtx_ || locked_)
+ throw lock_error();
+ locked_ = mtx_->timed_lock( abs_time);
+ return locked_;
+ }
+
+ template< typename TimeDuration >
+ bool timed_lock( TimeDuration const& rel_time)
+ { return timed_lock( get_system_time() + rel_time); }
+
+ void unlock()
+ {
+ if ( ! mtx_ || ! locked_)
+ throw lock_error();
+ mtx_->unlock();
+ locked_ = false;
+ }
+
+ bool owns() const
+ { return locked_ && mtx_; }
+
+ operator unspecified_bool_type() const
+ { return locked_ ? & locked_ : 0; }
+
+ bool operator!() const
+ { return ! locked_; }
+
+ Mutex * mutex() const
+ { return mtx_; }
+
+ void swap( spin_lock & other)
+ {
+ std::swap( mtx_, other.mtx_);
+ std::swap( locked_, other.locked_);
+ }
+};
+
+}}
+
+#endif // BOOST_TASK_SPIN_LOCK_H
Added: sandbox/task/boost/task/spin_mutex.hpp
==============================================================================
--- (empty file)
+++ sandbox/task/boost/task/spin_mutex.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -0,0 +1,40 @@
+
+// Copyright Oliver Kowalke 2009.
+// 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_TASK_SPIN_MUTEX_H
+#define BOOST_TASK_SPIN_MUTEX_H
+
+#include <boost/cstdint.hpp>
+#include <boost/thread/thread_time.hpp>
+#include <boost/utility.hpp>
+
+namespace boost {
+namespace task {
+
+class spin_mutex : private noncopyable
+{
+private:
+ volatile uint32_t state_;
+
+public:
+ spin_mutex();
+
+ void lock();
+
+ bool try_lock();
+
+ bool timed_lock( system_time const& abs_time);
+
+ template< typename TimeDuration >
+ bool timed_lock( TimeDuration const& rel_time)
+ { return timed_lock( get_system_time() + rel_time); }
+
+ void unlock();
+};
+
+}}
+
+#endif // BOOST_TASK_SPIN_MUTEX_H
Modified: sandbox/task/boost/task/unbounded_buffer.hpp
==============================================================================
--- sandbox/task/boost/task/unbounded_buffer.hpp (original)
+++ sandbox/task/boost/task/unbounded_buffer.hpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -15,7 +15,12 @@
#include <boost/shared_ptr.hpp>
#include <boost/utility.hpp>
+#include <boost/task/detail/atomic.hpp>
+#include <boost/task/exceptions.hpp>
#include <boost/task/semaphore.hpp>
+#include <boost/task/spin_condition.hpp>
+#include <boost/task/spin_lock.hpp>
+#include <boost/task/spin_mutex.hpp>
#include <boost/config/abi_prefix.hpp>
@@ -25,62 +30,152 @@
class unbounded_buffer
{
private:
+ typedef T value_type;
+
class base
{
private:
- class guard : private noncopyable
- {
- private:
- semaphore & sem_;
-
- public:
- guard( semaphore & sem)
- : sem_( sem)
- { sem_.wait(); }
-
- ~guard()
- { sem_.post(); }
- };
-
- semaphore protection_sem_;
- semaphore consumer_sem_;
- std::deque< T > queue_;
+ volatile uint32_t state_;
+ std::deque< value_type > queue_;
+ spin_mutex mtx_;
+ spin_condition not_empty_cond_;
base( base &);
base & operator=( base const&);
+ bool active_() const
+ { return 0 == state_; }
+
+ void deactivate_()
+ { detail::atomic_fetch_add( & state_, 1); }
+
bool empty_() const
{ return queue_.empty(); }
+ void put_( value_type const& va)
+ {
+ if ( ! active_() )
+ throw task_rejected("queue is not active");
+ queue_.push_back( va);
+ not_empty_cond_.notify_one();
+ }
+
+ bool take_(
+ optional< value_type > & va,
+ spin_lock< spin_mutex > & lk)
+ {
+ bool empty = empty_();
+ if ( ! active_() && empty)
+ return false;
+ if ( empty)
+ {
+ try
+ {
+ not_empty_cond_.wait(
+ lk,
+ bind(
+ & base::consumers_activate_,
+ this) );
+ }
+ catch ( thread_interrupted const&)
+ { return false; }
+ }
+ if ( ! active_() && empty_() )
+ return false;
+ va = queue_.front();
+ queue_.pop_front();
+ return va;
+ }
+
+ template< typename Duration >
+ bool take_(
+ optional< value_type > & va,
+ Duration const& rel_time,
+ spin_lock< spin_mutex > & lk)
+ {
+ bool empty = empty_();
+ if ( ! active_() && empty)
+ return false;
+ if ( empty)
+ {
+ try
+ {
+ if ( ! not_empty_cond_.timed_wait(
+ lk,
+ rel_time,
+ bind(
+ & base::consumers_activate_,
+ this) ) )
+ return false;
+ }
+ catch ( thread_interrupted const&)
+ { return false; }
+ }
+ if ( ! active_() && empty_() )
+ return false;
+ va = queue_.front();
+ queue_.pop_front();
+ return va;
+ }
+
+ bool try_take_( optional< value_type > & va)
+ {
+ if ( empty_() )
+ return false;
+ va = queue_.front();
+ queue_.pop_front();
+ return va;
+ }
+
bool consumers_activate_() const
- { return ! empty_(); }
+ { return ! active_() || ! empty_(); }
public:
base()
:
- protection_sem_( 1),
- consumer_sem_( 0),
- queue_()
+ state_( 0),
+ queue_(),
+ mtx_(),
+ not_empty_cond_()
{}
- void put( T const& t)
+ bool active()
+ { return active_(); }
+
+ void deactivate()
+ { deactivate_(); }
+
+ bool empty()
{
- {
- guard lk( protection_sem_);
- queue_.push_back( t);
- }
- consumer_sem_.post();
+ spin_lock< spin_mutex > lk( mtx_);
+ return empty_();
}
- bool take( optional< T > & t)
+ void put( value_type const& va)
{
- consumer_sem_.wait();
- {
- guard lk( protection_sem_);
- t = queue_.front();
- queue_.pop_front();
- }
- return true;
+ spin_lock< spin_mutex > lk( mtx_);
+ put_( va);
+ }
+
+ bool take( optional< value_type > & va)
+ {
+ spin_lock< spin_mutex > lk( mtx_);
+ return take_( va, lk);
+ }
+
+ template< typename TimeDuration >
+ bool take(
+ optional< value_type > & va,
+ TimeDuration const& rel_time)
+ {
+ spin_lock< spin_mutex > lk( mtx_);
+ return take_( va, rel_time, lk);
+ }
+
+ bool try_take( optional< value_type > & va)
+ {
+ spin_lock< spin_mutex > lk( mtx_);
+ return try_take_( va);
}
};
@@ -91,11 +186,29 @@
: impl_( new base)
{}
+ bool active()
+ { return impl_->active(); }
+
+ void deactivate()
+ { impl_->deactivate(); }
+
+ bool empty()
+ { return impl_->empty(); }
+
void put( T const& t)
{ impl_->put( t); }
- void take( optional< T > & t)
- { impl_->take( t); }
+ bool take( optional< T > & t)
+ { return impl_->take( t); }
+
+ template< typename TimeDuration >
+ bool take(
+ optional< T > & t,
+ TimeDuration const& rel_time)
+ { return impl_->take( t, rel_time); }
+
+ bool try_take( optional< T > & t)
+ { return impl_->try_take_( t); }
};
}}
Modified: sandbox/task/libs/task/build/Jamfile.v2
==============================================================================
--- sandbox/task/libs/task/build/Jamfile.v2 (original)
+++ sandbox/task/libs/task/build/Jamfile.v2 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -41,6 +41,8 @@
poolsize.cpp
scanns.cpp
semaphore_windows.cpp
+ spin_condition.cpp
+ spin_mutex.cpp
stacksize.cpp
watermark.cpp
worker.cpp
@@ -58,6 +60,8 @@
poolsize.cpp
scanns.cpp
semaphore_posix.cpp
+ spin_condition.cpp
+ spin_mutex.cpp
stacksize.cpp
watermark.cpp
worker.cpp
Modified: sandbox/task/libs/task/examples/buffer_multi.cpp
==============================================================================
--- sandbox/task/libs/task/examples/buffer_multi.cpp (original)
+++ sandbox/task/libs/task/examples/buffer_multi.cpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -60,11 +60,15 @@
tsk::unbounded_buffer< std::pair< int , int > > & recv,
int n)
{
- send.put( n);
+ for ( int i = 0; i <= n; ++i)
+ send.put( i);
+ send.deactivate();
boost::optional< std::pair< int , int > > r;
- recv.take( r);
- BOOST_ASSERT( r);
- printf("fib(%d) == %d\n", r->first, r->second);
+ while ( recv.take( r) )
+ {
+ BOOST_ASSERT( r);
+ printf("fib(%d) == %d\n", r->first, r->second);
+ }
}
inline
@@ -73,10 +77,13 @@
tsk::unbounded_buffer< std::pair< int , int > > & send)
{
boost::optional< int > n;
- recv.take( n);
- BOOST_ASSERT( n);
- int r = parallel_fib( * n, 5);
- send.put( std::make_pair( * n, r) );
+ while ( recv.take( n) )
+ {
+ BOOST_ASSERT( n);
+ int r = parallel_fib( * n, 5);
+ send.put( std::make_pair( * n, r) );
+ }
+ send.deactivate();
}
@@ -91,30 +98,14 @@
tsk::handle< void > h1(
tsk::async(
- tsk::make_task( submit, buf1, buf2, 5),
- tsk::new_thread() ) );
- tsk::handle< void > h2(
- tsk::async(
- tsk::make_task( submit, buf1, buf2, 10),
- tsk::new_thread() ) );
- tsk::handle< void > h3(
- tsk::async(
tsk::make_task( submit, buf1, buf2, 15),
tsk::new_thread() ) );
tsk::async(
tsk::make_task( calculate, buf1, buf2),
pool);
- tsk::async(
- tsk::make_task( calculate, buf1, buf2),
- pool);
- tsk::async(
- tsk::make_task( calculate, buf1, buf2),
- pool);
h1.get();
- h2.get();
- h3.get();
return EXIT_SUCCESS;
}
Modified: sandbox/task/libs/task/examples/buffer_multi2.cpp
==============================================================================
--- sandbox/task/libs/task/examples/buffer_multi2.cpp (original)
+++ sandbox/task/libs/task/examples/buffer_multi2.cpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -60,11 +60,15 @@
tsk::unbounded_buffer< std::pair< int , int > > & recv,
int n)
{
- send.put( n);
+ for ( int i = 0; i <= n; ++i)
+ send.put( i);
+ send.deactivate();
boost::optional< std::pair< int , int > > r;
- recv.take( r);
- BOOST_ASSERT( r);
- printf("fib(%d) == %d\n", r->first, r->second);
+ while ( recv.take( r) )
+ {
+ BOOST_ASSERT( r);
+ printf("fib(%d) == %d\n", r->first, r->second);
+ }
}
inline
@@ -73,13 +77,15 @@
tsk::unbounded_buffer< std::pair< int , int > > & send)
{
boost::optional< int > n;
- recv.take( n);
- BOOST_ASSERT( n);
- int r = parallel_fib( * n, 5);
- send.put( std::make_pair( * n, r) );
+ while ( recv.take( n) )
+ {
+ BOOST_ASSERT( n);
+ int r = parallel_fib( * n, 5);
+ send.put( std::make_pair( * n, r) );
+ }
+ send.deactivate();
}
-
int main( int argc, char *argv[])
{
try
@@ -90,21 +96,8 @@
tsk::unbounded_buffer< std::pair< int , int > > buf2;
tsk::async(
- tsk::make_task( submit, buf1, buf2, 5),
- pool);
- tsk::async(
- tsk::make_task( submit, buf1, buf2, 10),
- pool);
- tsk::async(
tsk::make_task( submit, buf1, buf2, 15),
pool);
-
- tsk::async(
- tsk::make_task( calculate, buf1, buf2),
- pool);
- tsk::async(
- tsk::make_task( calculate, buf1, buf2),
- pool);
tsk::async(
tsk::make_task( calculate, buf1, buf2),
pool);
Modified: sandbox/task/libs/task/examples/no_deadlock_pool2.cpp
==============================================================================
--- sandbox/task/libs/task/examples/no_deadlock_pool2.cpp (original)
+++ sandbox/task/libs/task/examples/no_deadlock_pool2.cpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -38,9 +38,7 @@
try
{
tsk::poolsize psize( boost::thread::hardware_concurrency() );
- tsk::static_pool<
- tsk::unbounded_channel< tsk::fifo >
- > pool( psize);
+ tsk::static_pool< tsk::unbounded_twolock_fifo > pool( psize);
fprintf( stderr, "pool-size == %d\n", pool.size() );
Added: sandbox/task/libs/task/src/spin_condition.cpp
==============================================================================
--- (empty file)
+++ sandbox/task/libs/task/src/spin_condition.cpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -0,0 +1,195 @@
+
+// Copyright Oliver Kowalke 2009.
+// Distributed under the Boost Software License, Version 1.0.
+// (See accompanying file LICENSE_1_0.txt or copy at
+// http://www.boost.org/LICENSE_1_0.txt)
+
+#include "boost/task/spin_condition.hpp"
+
+#include <boost/thread.hpp>
+
+#include <boost/task/detail/atomic.hpp>
+#include <boost/task/spin_mutex.hpp>
+#include <boost/task/spin_lock.hpp>
+#include <boost/task/utility.hpp>
+
+namespace boost {
+namespace task {
+
+void
+spin_condition::notify_( command cmd)
+{
+ enter_mtx_.lock();
+
+ if ( waiters_ == 0)
+ {
+ enter_mtx_.unlock();
+ return;
+ }
+
+ uint32_t expected = static_cast< uint32_t >( SLEEPING);
+ while ( ! detail::atomic_compare_exchange_strong(
+ static_cast< uint32_t volatile* >( & cmd_), & expected, cmd) )
+ {
+ if ( this_task::runs_in_pool() )
+ this_task::block();
+ else
+ this_thread::yield();
+ }
+}
+
+void
+spin_condition::wait_( spin_mutex & mtx)
+{
+ {
+ spin_lock< spin_mutex > lk( enter_mtx_);
+ if ( ! lk) return;
+ detail::atomic_fetch_add( & waiters_, 1);
+ mtx.unlock();
+ }
+
+ bool unlock_enter_mtx = false;
+ for (;;)
+ {
+ while ( SLEEPING == cmd_)
+ {
+ if ( this_task::runs_in_pool() )
+ this_task::block();
+ else
+ this_thread::yield();
+ }
+
+ spin_lock< spin_mutex > lk( check_mtx_);
+ if ( ! lk)
+ {
+ unlock_enter_mtx = true;
+ break;
+ }
+
+ uint32_t expected = static_cast< uint32_t >( NOTIFY_ONE);
+ detail::atomic_compare_exchange_strong(
+ static_cast< uint32_t volatile* >( & cmd_), & expected, SLEEPING);
+ if ( SLEEPING == expected)
+ continue;
+ else if ( NOTIFY_ONE == expected)
+ {
+ unlock_enter_mtx = true;
+ detail::atomic_fetch_sub( & waiters_, 1);
+ break;
+ }
+ else
+ {
+ unlock_enter_mtx = 1 == detail::atomic_fetch_sub( & waiters_, 1);
+ if ( unlock_enter_mtx)
+ {
+ expected = static_cast< uint32_t >( NOTIFY_ALL);
+ detail::atomic_compare_exchange_strong(
+ static_cast< uint32_t volatile* >( & cmd_), & expected, SLEEPING);
+ }
+ break;
+ }
+ }
+
+ if ( unlock_enter_mtx)
+ enter_mtx_.unlock();
+
+ mtx.lock();
+}
+
+bool
+spin_condition::wait_( spin_mutex & mtx, system_time const& abs_time)
+{
+ if ( get_system_time() >= abs_time) return false;
+
+ {
+ spin_lock< spin_mutex > lk( enter_mtx_, abs_time);
+ if ( ! lk) return false;
+ detail::atomic_fetch_add( & waiters_, 1);
+ mtx.unlock();
+ }
+
+ bool timed_out = false, unlock_enter_mtx = false;
+ for (;;)
+ {
+ while ( SLEEPING == cmd_)
+ {
+ if ( this_task::runs_in_pool() )
+ this_task::block();
+ else
+ this_thread::yield();
+
+ if ( get_system_time() >= abs_time)
+ {
+ timed_out = enter_mtx_.try_lock();
+ if ( ! timed_out)
+ continue;
+ break;
+ }
+ }
+
+ if ( timed_out)
+ {
+ detail::atomic_fetch_sub( & waiters_, 1);
+ unlock_enter_mtx = true;
+ break;
+ }
+ else
+ {
+ spin_lock< spin_mutex > lk( check_mtx_);
+ if ( ! lk)
+ {
+ timed_out = true;
+ unlock_enter_mtx = true;
+ break;
+ }
+
+ uint32_t expected = static_cast< uint32_t >( NOTIFY_ONE);
+ detail::atomic_compare_exchange_strong(
+ static_cast< uint32_t volatile* >( & cmd_), & expected, SLEEPING);
+ if ( SLEEPING == expected)
+ continue;
+ else if ( NOTIFY_ONE == expected)
+ {
+ unlock_enter_mtx = true;
+ detail::atomic_fetch_sub( & waiters_, 1);
+ break;
+ }
+ else
+ {
+ unlock_enter_mtx = 1 == detail::atomic_fetch_sub( & waiters_, 1);
+ if ( unlock_enter_mtx)
+ {
+ expected = static_cast< uint32_t >( NOTIFY_ALL);
+ detail::atomic_compare_exchange_strong(
+ static_cast< uint32_t volatile* >( & cmd_), & expected, SLEEPING);
+ }
+ break;
+ }
+ }
+ }
+
+ if ( unlock_enter_mtx)
+ enter_mtx_.unlock();
+
+ mtx.lock();
+
+ return ! timed_out;
+}
+
+spin_condition::spin_condition()
+:
+cmd_( SLEEPING),
+waiters_( 0),
+enter_mtx_(),
+check_mtx_()
+{}
+
+void
+spin_condition::notify_one()
+{ notify_( NOTIFY_ONE); }
+
+void
+spin_condition::notify_all()
+{ notify_( NOTIFY_ALL); }
+
+}}
Added: sandbox/task/libs/task/src/spin_mutex.cpp
==============================================================================
--- (empty file)
+++ sandbox/task/libs/task/src/spin_mutex.cpp 2009-10-03 05:04:47 EDT (Sat, 03 Oct 2009)
@@ -0,0 +1,81 @@
+
+// Copyright Oliver Kowalke 2009.
+// Distributed under the Boost Software License, Version 1.0.
+// (See accompanying file LICENSE_1_0.txt or copy at
+// http://www.boost.org/LICENSE_1_0.txt)
+
+#include "boost/task/spin_mutex.hpp"
+
+#include <boost/thread.hpp>
+
+#include "boost/task/detail/atomic.hpp"
+#include "boost/task/utility.hpp"
+
+namespace boost {
+namespace task {
+
+spin_mutex::spin_mutex()
+: state_( 0)
+{}
+
+void
+spin_mutex::lock()
+{
+ for (;;)
+ {
+ uint32_t expected = 0;
+ if ( detail::atomic_compare_exchange_strong( & state_, & expected, 1) )
+ break;
+ else
+ {
+ if ( this_task::runs_in_pool() )
+ this_task::block();
+ else
+ this_thread::yield();
+ }
+ }
+}
+
+bool
+spin_mutex::try_lock()
+{
+ uint32_t expected = 0;
+ return detail::atomic_compare_exchange_strong( & state_, & expected, 1);
+}
+
+bool
+spin_mutex::timed_lock( system_time const& abs_time)
+{
+ if ( abs_time.is_infinity() )
+ {
+ lock();
+ return true;
+ }
+
+ if ( get_system_time() >= abs_time)
+ return false;
+
+ for (;;)
+ {
+ if ( try_lock() ) break;
+
+ if ( get_system_time() >= abs_time)
+ return false;
+
+ if ( this_task::runs_in_pool() )
+ this_task::block();
+ else
+ this_thread::yield();
+ }
+
+ return true;
+}
+
+void
+spin_mutex::unlock()
+{
+ uint32_t expected = 1;
+ detail::atomic_compare_exchange_strong( & state_, & expected, 0);
+}
+
+}}