Subject: [Boost-commit] svn:boost r74955 - in trunk/boost/asio/detail: . impl
From: chris_at_[hidden]
Date: 2011-10-15 19:59:12
Author: chris_kohlhoff
Date: 2011-10-15 19:59:12 EDT (Sat, 15 Oct 2011)
New Revision: 74955
URL: http://svn.boost.org/trac/boost/changeset/74955
Log:
Make number of strand implementations configurable by defining
BOOST_ASIO_STRAND_IMPLEMENTATIONS to the number.
Programs can now define BOOST_ASIO_ENABLE_SEQUENTIAL_STRAND_ALLOCATION to
switch the allocation of strand implementations to use a round-robin approach
rather than hashing.
Fix potential strand starvation issue that can occur when strand.post() is used.
Text files modified:
trunk/boost/asio/detail/impl/strand_service.hpp | 5 +
trunk/boost/asio/detail/impl/strand_service.ipp | 82 +++++++++++++++++++++++++--------------
trunk/boost/asio/detail/strand_service.hpp | 23 ++++++++--
3 files changed, 74 insertions(+), 36 deletions(-)
Modified: trunk/boost/asio/detail/impl/strand_service.hpp
==============================================================================
--- trunk/boost/asio/detail/impl/strand_service.hpp (original)
+++ trunk/boost/asio/detail/impl/strand_service.hpp 2011-10-15 19:59:12 EDT (Sat, 15 Oct 2011)
@@ -29,7 +29,7 @@
inline strand_service::strand_impl::strand_impl()
: operation(&strand_service::do_complete),
- count_(0)
+ locked_(false)
{
}
@@ -41,7 +41,8 @@
~on_dispatch_exit()
{
impl_->mutex_.lock();
- bool more_handlers = (--impl_->count_ > 0);
+ impl_->ready_queue_.push(impl_->waiting_queue_);
+ bool more_handlers = impl_->locked_ = !impl_->ready_queue_.empty();
impl_->mutex_.unlock();
if (more_handlers)
Modified: trunk/boost/asio/detail/impl/strand_service.ipp
==============================================================================
--- trunk/boost/asio/detail/impl/strand_service.ipp (original)
+++ trunk/boost/asio/detail/impl/strand_service.ipp 2011-10-15 19:59:12 EDT (Sat, 15 Oct 2011)
@@ -33,7 +33,8 @@
~on_do_complete_exit()
{
impl_->mutex_.lock();
- bool more_handlers = (--impl_->count_ > 0);
+ impl_->ready_queue_.push(impl_->waiting_queue_);
+ bool more_handlers = impl_->locked_ = !impl_->ready_queue_.empty();
impl_->mutex_.unlock();
if (more_handlers)
@@ -56,20 +57,29 @@
boost::asio::detail::mutex::scoped_lock lock(mutex_);
for (std::size_t i = 0; i < num_implementations; ++i)
+ {
if (strand_impl* impl = implementations_[i].get())
- ops.push(impl->queue_);
+ {
+ ops.push(impl->waiting_queue_);
+ ops.push(impl->ready_queue_);
+ }
+ }
}
void strand_service::construct(strand_service::implementation_type& impl)
{
+ boost::asio::detail::mutex::scoped_lock lock(mutex_);
+
std::size_t salt = salt_++;
+#if defined(BOOST_ASIO_ENABLE_SEQUENTIAL_STRAND_ALLOCATION)
+ std::size_t index = salt;
+#else // defined(BOOST_ASIO_ENABLE_SEQUENTIAL_STRAND_ALLOCATION)
std::size_t index = reinterpret_cast<std::size_t>(&impl);
index += (reinterpret_cast<std::size_t>(&impl) >> 3);
index ^= salt + 0x9e3779b9 + (index << 6) + (index >> 2);
+#endif // defined(BOOST_ASIO_ENABLE_SEQUENTIAL_STRAND_ALLOCATION)
index = index % num_implementations;
- boost::asio::detail::mutex::scoped_lock lock(mutex_);
-
if (!implementations_[index].get())
implementations_[index].reset(new strand_impl);
impl = implementations_[index].get();
@@ -77,41 +87,55 @@
bool strand_service::do_dispatch(implementation_type& impl, operation* op)
{
- // If we are running inside the io_service, and no other handler is queued
- // or running, then the handler can run immediately.
+ // If we are running inside the io_service, and no other handler already
+ // holds the strand lock, then the handler can run immediately.
bool can_dispatch = io_service_.can_dispatch();
impl->mutex_.lock();
- bool first = (++impl->count_ == 1);
- if (can_dispatch && first)
+ if (can_dispatch && !impl->locked_)
{
// Immediate invocation is allowed.
+ impl->locked_ = true;
impl->mutex_.unlock();
return true;
}
- // Immediate invocation is not allowed, so enqueue for later.
- impl->queue_.push(op);
- impl->mutex_.unlock();
-
- // The first handler to be enqueued is responsible for scheduling the
- // strand.
- if (first)
+ if (impl->locked_)
+ {
+ // Some other handler already holds the strand lock. Enqueue for later.
+ impl->waiting_queue_.push(op);
+ impl->mutex_.unlock();
+ }
+ else
+ {
+ // The handler is acquiring the strand lock and so is responsible for
+ // scheduling the strand.
+ impl->locked_ = true;
+ impl->mutex_.unlock();
+ impl->ready_queue_.push(op);
io_service_.post_immediate_completion(impl);
+ }
return false;
}
void strand_service::do_post(implementation_type& impl, operation* op)
{
- // Add the handler to the queue.
impl->mutex_.lock();
- bool first = (++impl->count_ == 1);
- impl->queue_.push(op);
- impl->mutex_.unlock();
-
- // The first handler to be enqueue is responsible for scheduling the strand.
- if (first)
+ if (impl->locked_)
+ {
+ // Some other handler already holds the strand lock. Enqueue for later.
+ impl->waiting_queue_.push(op);
+ impl->mutex_.unlock();
+ }
+ else
+ {
+ // The handler is acquiring the strand lock and so is responsible for
+ // scheduling the strand.
+ impl->locked_ = true;
+ impl->mutex_.unlock();
+ impl->ready_queue_.push(op);
io_service_.post_immediate_completion(impl);
+ }
}
void strand_service::do_complete(io_service_impl* owner, operation* base,
@@ -121,12 +145,6 @@
{
strand_impl* impl = static_cast<strand_impl*>(base);
- // Get the next handler to be executed.
- impl->mutex_.lock();
- operation* o = impl->queue_.front();
- impl->queue_.pop();
- impl->mutex_.unlock();
-
// Indicate that this strand is executing on the current thread.
call_stack<strand_impl>::context ctx(impl);
@@ -134,7 +152,13 @@
on_do_complete_exit on_exit = { owner, impl };
(void)on_exit;
- o->complete(*owner, ec, 0);
+ // Run all ready handlers. No lock is required since the ready queue is
+ // accessed only within the strand.
+ while (operation* o = impl->ready_queue_.front())
+ {
+ impl->ready_queue_.pop();
+ o->complete(*owner, ec, 0);
+ }
}
}
Modified: trunk/boost/asio/detail/strand_service.hpp
==============================================================================
--- trunk/boost/asio/detail/strand_service.hpp (original)
+++ trunk/boost/asio/detail/strand_service.hpp 2011-10-15 19:59:12 EDT (Sat, 15 Oct 2011)
@@ -57,11 +57,20 @@
// Mutex to protect access to internal data.
boost::asio::detail::mutex mutex_;
- // The count of handlers in the strand, including the upcall (if any).
- std::size_t count_;
-
- // The handlers waiting on the strand.
- op_queue<operation> queue_;
+ // Indicates whether the strand is currently "locked" by a handler. This
+ // means that there is a handler upcall in progress, or that the strand
+ // itself has been scheduled in order to invoke some pending handlers.
+ bool locked_;
+
+ // The handlers that are waiting on the strand but should not be run until
+ // after the next time the strand is scheduled. This queue must only be
+ // modified while the mutex is locked.
+ op_queue<operation> waiting_queue_;
+
+ // The handlers that are ready to be run. Logically speaking, these are the
+ // handlers that hold the strand's lock. The ready queue is only modified
+ // from within the strand and so may be accessed without locking the mutex.
+ op_queue<operation> ready_queue_;
};
typedef strand_impl* implementation_type;
@@ -105,7 +114,11 @@
boost::asio::detail::mutex mutex_;
// Number of implementations shared between all strand objects.
+#if defined(BOOST_ASIO_STRAND_IMPLEMENTATIONS)
+ enum { num_implementations = BOOST_ASIO_STRAND_IMPLEMENTATIONS };
+#else // defined(BOOST_ASIO_STRAND_IMPLEMENTATIONS)
enum { num_implementations = 193 };
+#endif // defined(BOOST_ASIO_STRAND_IMPLEMENTATIONS)
// Pool of implementations.
scoped_ptr<strand_impl> implementations_[num_implementations];