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