[Boost-commit] svn:boost r74955 - in trunk/boost/asio/detail: . impl

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];