[Boost-commit] svn:boost r75009 - in branches/release: boost/asio boost/asio/detail boost/asio/detail/impl libs/asio/doc

Subject: [Boost-commit] svn:boost r75009 - in branches/release: boost/asio boost/asio/detail boost/asio/detail/impl libs/asio/doc
From: chris_at_[hidden]
Date: 2011-10-17 07:32:29

Author: chris_kohlhoff
Date: 2011-10-17 07:32:28 EDT (Mon, 17 Oct 2011)
New Revision: 75009
URL: http://svn.boost.org/trac/boost/changeset/75009

Log:
Merge from trunk:

* 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.

* Update descriptor state allocation in kqueue_reactor to match approach used in
  epoll_reactor.

* Construct epoll_reactor's interrupter member first to fix exception safety
  issue. Fixes #6006

* Clarify that the read operation ends when the buffer is full. Fixes #5999

Text files modified: 
   branches/release/boost/asio/detail/epoll_reactor.hpp       |     6 +-                                      
   branches/release/boost/asio/detail/impl/epoll_reactor.ipp  |     2                                         
   branches/release/boost/asio/detail/impl/kqueue_reactor.ipp |    32 +++++++++-----                          
   branches/release/boost/asio/detail/impl/strand_service.hpp |     5 +                                       
   branches/release/boost/asio/detail/impl/strand_service.ipp |    82 +++++++++++++++++++++++++-------------- 
   branches/release/boost/asio/detail/kqueue_reactor.hpp      |     6 ++                                      
   branches/release/boost/asio/detail/strand_service.hpp      |    23 ++++++++--                              
   branches/release/boost/asio/read.hpp                       |    12 +++++                                   
   branches/release/libs/asio/doc/history.qbk                 |    45 +++++++++++++++++++++                   
   9 files changed, 161 insertions(+), 52 deletions(-)

Modified: branches/release/boost/asio/detail/epoll_reactor.hpp
==============================================================================
--- branches/release/boost/asio/detail/epoll_reactor.hpp	(original)
+++ branches/release/boost/asio/detail/epoll_reactor.hpp	2011-10-17 07:32:28 EDT (Mon, 17 Oct 2011)
@@ -205,15 +205,15 @@
   // Mutex to protect access to internal data.
   mutex mutex_;
 
+  // The interrupter is used to break a blocking epoll_wait call.
+  select_interrupter interrupter_;
+
   // The epoll file descriptor.
   int epoll_fd_;
 
   // The timer file descriptor.
   int timer_fd_;
 
-  // The interrupter is used to break a blocking epoll_wait call.
-  select_interrupter interrupter_;
-
   // The timer queues.
   timer_queue_set timer_queues_;
 

Modified: branches/release/boost/asio/detail/impl/epoll_reactor.ipp
==============================================================================
--- branches/release/boost/asio/detail/impl/epoll_reactor.ipp	(original)
+++ branches/release/boost/asio/detail/impl/epoll_reactor.ipp	2011-10-17 07:32:28 EDT (Mon, 17 Oct 2011)
@@ -39,9 +39,9 @@
   : boost::asio::detail::service_base<epoll_reactor>(io_service),
     io_service_(use_service<io_service_impl>(io_service)),
     mutex_(),
+    interrupter_(),
     epoll_fd_(do_epoll_create()),
     timer_fd_(do_timerfd_create()),
-    interrupter_(),
     shutdown_(false)
 {
   // Add the interrupter's descriptor to epoll.

Modified: branches/release/boost/asio/detail/impl/kqueue_reactor.ipp
==============================================================================
--- branches/release/boost/asio/detail/impl/kqueue_reactor.ipp	(original)
+++ branches/release/boost/asio/detail/impl/kqueue_reactor.ipp	2011-10-17 07:32:28 EDT (Mon, 17 Oct 2011)
@@ -126,9 +126,10 @@
 int kqueue_reactor::register_descriptor(socket_type descriptor,
     kqueue_reactor::per_descriptor_data& descriptor_data)
 {
-  mutex::scoped_lock lock(registered_descriptors_mutex_);
+  descriptor_data = allocate_descriptor_state();
+
+  mutex::scoped_lock lock(descriptor_data->mutex_);
 
-  descriptor_data = registered_descriptors_.alloc();
   descriptor_data->descriptor_ = descriptor;
   descriptor_data->shutdown_ = false;
 
@@ -143,9 +144,10 @@
     int op_type, socket_type descriptor,
     kqueue_reactor::per_descriptor_data& descriptor_data, reactor_op* op)
 {
-  mutex::scoped_lock lock(registered_descriptors_mutex_);
+  descriptor_data = allocate_descriptor_state();
+
+  mutex::scoped_lock lock(descriptor_data->mutex_);
 
-  descriptor_data = registered_descriptors_.alloc();
   descriptor_data->descriptor_ = descriptor;
   descriptor_data->shutdown_ = false;
   descriptor_data->op_queue_[op_type].push(op);
@@ -304,7 +306,6 @@
     return;
 
   mutex::scoped_lock descriptor_lock(descriptor_data->mutex_);
-  mutex::scoped_lock descriptors_lock(registered_descriptors_mutex_);
 
   if (!descriptor_data->shutdown_)
   {
@@ -339,11 +340,9 @@
 
     descriptor_lock.unlock();
 
-    registered_descriptors_.free(descriptor_data);
+    free_descriptor_state(descriptor_data);
     descriptor_data = 0;
 
-    descriptors_lock.unlock();
-
     io_service_.post_deferred_completions(ops);
   }
 }
@@ -355,7 +354,6 @@
     return;
 
   mutex::scoped_lock descriptor_lock(descriptor_data->mutex_);
-  mutex::scoped_lock descriptors_lock(registered_descriptors_mutex_);
 
   if (!descriptor_data->shutdown_)
   {
@@ -375,10 +373,8 @@
 
     descriptor_lock.unlock();
 
-    registered_descriptors_.free(descriptor_data);
+    free_descriptor_state(descriptor_data);
     descriptor_data = 0;
-
-    descriptors_lock.unlock();
   }
 }
 
@@ -511,6 +507,18 @@
   return fd;
 }
 
+kqueue_reactor::descriptor_state* kqueue_reactor::allocate_descriptor_state()
+{
+  mutex::scoped_lock descriptors_lock(registered_descriptors_mutex_);
+  return registered_descriptors_.alloc();
+}
+
+void kqueue_reactor::free_descriptor_state(kqueue_reactor::descriptor_state* s)
+{
+  mutex::scoped_lock descriptors_lock(registered_descriptors_mutex_);
+  registered_descriptors_.free(s);
+}
+
 void kqueue_reactor::do_add_timer_queue(timer_queue_base& queue)
 {
   mutex::scoped_lock lock(mutex_);

Modified: branches/release/boost/asio/detail/impl/strand_service.hpp
==============================================================================
--- branches/release/boost/asio/detail/impl/strand_service.hpp	(original)
+++ branches/release/boost/asio/detail/impl/strand_service.hpp	2011-10-17 07:32:28 EDT (Mon, 17 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: branches/release/boost/asio/detail/impl/strand_service.ipp
==============================================================================
--- branches/release/boost/asio/detail/impl/strand_service.ipp	(original)
+++ branches/release/boost/asio/detail/impl/strand_service.ipp	2011-10-17 07:32:28 EDT (Mon, 17 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: branches/release/boost/asio/detail/kqueue_reactor.hpp
==============================================================================
--- branches/release/boost/asio/detail/kqueue_reactor.hpp	(original)
+++ branches/release/boost/asio/detail/kqueue_reactor.hpp	2011-10-17 07:32:28 EDT (Mon, 17 Oct 2011)
@@ -169,6 +169,12 @@
   // cannot be created.
   BOOST_ASIO_DECL static int do_kqueue_create();
 
+  // Allocate a new descriptor state object.
+  BOOST_ASIO_DECL descriptor_state* allocate_descriptor_state();
+
+  // Free an existing descriptor state object.
+  BOOST_ASIO_DECL void free_descriptor_state(descriptor_state* s);
+
   // Helper function to add a new timer queue.
   BOOST_ASIO_DECL void do_add_timer_queue(timer_queue_base& queue);
 

Modified: branches/release/boost/asio/detail/strand_service.hpp
==============================================================================
--- branches/release/boost/asio/detail/strand_service.hpp	(original)
+++ branches/release/boost/asio/detail/strand_service.hpp	2011-10-17 07:32:28 EDT (Mon, 17 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];

Modified: branches/release/boost/asio/read.hpp
==============================================================================
--- branches/release/boost/asio/read.hpp	(original)
+++ branches/release/boost/asio/read.hpp	2011-10-17 07:32:28 EDT (Mon, 17 Oct 2011)
@@ -214,6 +214,8 @@
  * This function is used to read a certain number of bytes of data from a
  * stream. The call will block until one of the following conditions is true:
  *
+ * @li The supplied buffer is full (that is, it has reached maximum size).
+ *
  * @li An error occurred.
  *
  * This operation is implemented in terms of zero or more calls to the stream's
@@ -241,6 +243,8 @@
  * This function is used to read a certain number of bytes of data from a
  * stream. The call will block until one of the following conditions is true:
  *
+ * @li The supplied buffer is full (that is, it has reached maximum size).
+ *
  * @li An error occurred.
  *
  * This operation is implemented in terms of zero or more calls to the stream's
@@ -269,6 +273,8 @@
  * This function is used to read a certain number of bytes of data from a
  * stream. The call will block until one of the following conditions is true:
  *
+ * @li The supplied buffer is full (that is, it has reached maximum size).
+ *
  * @li The completion_condition function object returns 0.
  *
  * This operation is implemented in terms of zero or more calls to the stream's
@@ -307,6 +313,8 @@
  * This function is used to read a certain number of bytes of data from a
  * stream. The call will block until one of the following conditions is true:
  *
+ * @li The supplied buffer is full (that is, it has reached maximum size).
+ *
  * @li The completion_condition function object returns 0.
  *
  * This operation is implemented in terms of zero or more calls to the stream's
@@ -496,6 +504,8 @@
  * asynchronous operation will continue until one of the following conditions is
  * true:
  *
+ * @li The supplied buffer is full (that is, it has reached maximum size).
+ *
  * @li An error occurred.
  *
  * This operation is implemented in terms of zero or more calls to the stream's
@@ -546,6 +556,8 @@
  * asynchronous operation will continue until one of the following conditions is
  * true:
  *
+ * @li The supplied buffer is full (that is, it has reached maximum size).
+ *
  * @li The completion_condition function object returns 0.
  *
  * This operation is implemented in terms of zero or more calls to the stream's

Modified: branches/release/libs/asio/doc/history.qbk
==============================================================================
--- branches/release/libs/asio/doc/history.qbk	(original)
+++ branches/release/libs/asio/doc/history.qbk	2011-10-17 07:32:28 EDT (Mon, 17 Oct 2011)
@@ -7,6 +7,51 @@
 
 [section:history Revision History]
 
+[heading Asio 1.6.1 / Boost 1.48]
+
+* Implemented various performance improvements, including:
+  * Using thread-local operation queues in single-threaded use cases (i.e. when
+    `concurrency_hint` is 1) to eliminate a lock/unlock pair. 
+  * Allowing some `epoll_reactor` speculative operations to be performed
+    without holding the lock. 
+  * Improving locality of reference by performing an `epoll_reactor`'s I/O
+    operation immediately before the corresponding handler is called. This also
+    improves scalability across CPUs when multiple threads are running the
+    `io_service`. 
+  * Specialising asynchronous read and write operations for buffer sequences
+    that are arrays (`boost::array` or `std::array`) of exactly two buffers.
+* Fixed a compile error in the regex overload of `async_read_until`
+  ([@https://svn.boost.org/trac/boost/ticket/5688 #5688]).
+* Fixed a Windows-specific compile error by explicitly specifying the
+  `signal()` function from the global namespace
+  ([@https://svn.boost.org/trac/boost/ticket/5722 #5722]).
+* Changed the `deadline_timer` implementation so that it does not read the
+  clock unless the timer heap is non-empty.
+* Changed the SSL stream's buffers' sizes so that they are large enough to hold
+  a complete TLS record ([@https://svn.boost.org/trac/boost/ticket/5854 #5854]).
+* Fixed the behaviour of the synchronous `null_buffers` operations so that they
+  obey the user's non-blocking setting
+  ([@https://svn.boost.org/trac/boost/ticket/5756 #5756]).
+* Changed to set the size of the select `fd_set` at runtime when using Windows.
+* Disabled an MSVC warning due to const qualifier being applied to function type.
+* Fixed a crash due that occurs when using the Intel C++ compiler
+  ([@https://svn.boost.org/trac/boost/ticket/5763 #5763]).
+* Changed the initialisation of the OpenSSL library so that it supports all
+  available algorithms.
+* Fixed the SSL error mapping used when the session is gracefully shut down.
+* Added some latency test programs.
+* Clarified that a read operation ends when the buffer is full
+  ([@https://svn.boost.org/trac/boost/ticket/5999 #5999]).
+* Fixed an exception safety issue in `epoll_reactor` initialisation
+  ([@https://svn.boost.org/trac/boost/ticket/6006 #6006]).
+* Made the number of strand implementations configurable by defining
+  `BOOST_ASIO_STRAND_IMPLEMENTATIONS` to the desired number.
+* Added support for a new `BOOST_ASIO_ENABLE_SEQUENTIAL_STRAND_ALLOCATION` flag
+  which switches the allocation of strand implementations to use a round-robin
+  approach rather than hashing.
+* Fixed potential strand starvation issue that can occur when `strand.post()`
+  is used.
+
 [heading Asio 1.6.0 / Boost 1.47]
 
 * Added support for signal handling, using a new class called `signal_set`.