From: chris_at_[hidden]
Date: 2008-04-28 09:56:08
Author: chris_kohlhoff
Date: 2008-04-28 09:56:07 EDT (Mon, 28 Apr 2008)
New Revision: 44851
URL: http://svn.boost.org/trac/boost/changeset/44851
Log:
Add an experimental two-lock queue implementation for task_io_service.
Added:
trunk/boost/asio/detail/indirect_handler_queue.hpp (contents, props changed)
trunk/boost/asio/detail/task_io_service_2lock.hpp (contents, props changed)
Text files modified:
trunk/boost/asio/detail/task_io_service.hpp | 6 ++++++
1 files changed, 6 insertions(+), 0 deletions(-)
Added: trunk/boost/asio/detail/indirect_handler_queue.hpp
==============================================================================
--- (empty file)
+++ trunk/boost/asio/detail/indirect_handler_queue.hpp 2008-04-28 09:56:07 EDT (Mon, 28 Apr 2008)
@@ -0,0 +1,265 @@
+//
+// indirect_handler_queue.hpp
+// ~~~~~~~~~~~~~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2008 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// 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_ASIO_DETAIL_INDIRECT_HANDLER_QUEUE_HPP
+#define BOOST_ASIO_DETAIL_INDIRECT_HANDLER_QUEUE_HPP
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1200)
+# pragma once
+#endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
+
+#include <boost/asio/detail/push_options.hpp>
+
+#include <boost/asio/detail/handler_alloc_helpers.hpp>
+#include <boost/asio/detail/handler_invoke_helpers.hpp>
+#include <boost/asio/detail/noncopyable.hpp>
+
+namespace boost {
+namespace asio {
+namespace detail {
+
+class indirect_handler_queue
+ : private noncopyable
+{
+public:
+ class handler;
+
+ // Element for a node in the queue.
+ class node
+ {
+ public:
+ node()
+ : version_(0),
+ handler_(0),
+ next_(0)
+ {
+ }
+
+ private:
+ friend class indirect_handler_queue;
+ unsigned long version_;
+ handler* handler_;
+ node* next_;
+ };
+
+ // Base class for handlers in the queue.
+ class handler
+ : private noncopyable
+ {
+ public:
+ void invoke()
+ {
+ invoke_func_(this);
+ }
+
+ void destroy()
+ {
+ destroy_func_(this);
+ }
+
+ protected:
+ typedef void (*invoke_func_type)(handler*);
+ typedef void (*destroy_func_type)(handler*);
+
+ handler(invoke_func_type invoke_func,
+ destroy_func_type destroy_func)
+ : node_(new node),
+ invoke_func_(invoke_func),
+ destroy_func_(destroy_func)
+ {
+ }
+
+ ~handler()
+ {
+ if (node_)
+ delete node_;
+ }
+
+ private:
+ friend class indirect_handler_queue;
+ node* node_;
+ invoke_func_type invoke_func_;
+ destroy_func_type destroy_func_;
+ };
+
+ // Smart point to manager handler lifetimes.
+ class scoped_ptr
+ : private noncopyable
+ {
+ public:
+ explicit scoped_ptr(handler* h)
+ : handler_(h)
+ {
+ }
+
+ ~scoped_ptr()
+ {
+ if (handler_)
+ handler_->destroy();
+ }
+
+ handler* get() const
+ {
+ return handler_;
+ }
+
+ handler* release()
+ {
+ handler* tmp = handler_;
+ handler_ = 0;
+ return tmp;
+ }
+
+ private:
+ handler* handler_;
+ };
+
+ // Constructor.
+ indirect_handler_queue()
+ : front_(new node),
+ back_(front_),
+ next_version_(1)
+ {
+ }
+
+ // Destructor.
+ ~indirect_handler_queue()
+ {
+ while (front_)
+ {
+ node* tmp = front_;
+ front_ = front_->next_;
+ delete tmp;
+ }
+ }
+
+ // Wrap a handler to be pushed into the queue.
+ template <typename Handler>
+ static handler* wrap(Handler h)
+ {
+ // Allocate and construct an object to wrap the handler.
+ typedef handler_wrapper<Handler> value_type;
+ typedef handler_alloc_traits<Handler, value_type> alloc_traits;
+ raw_handler_ptr<alloc_traits> raw_ptr(h);
+ handler_ptr<alloc_traits> ptr(raw_ptr, h);
+ return ptr.release();
+ }
+
+ // Determine whether the queue has something ready to pop.
+ bool poppable()
+ {
+ return front_->next_ != 0;
+ }
+
+ // The version number at the front of the queue.
+ unsigned long front_version()
+ {
+ return front_->version_;
+ }
+
+ // The version number at the back of the queue.
+ unsigned long back_version()
+ {
+ return back_->version_;
+ }
+
+ // Pop a handler from the front of the queue.
+ handler* pop()
+ {
+ node* n = front_;
+ node* new_front = n->next_;
+ if (new_front)
+ {
+ handler* h = new_front->handler_;
+ h->node_ = n;
+ new_front->handler_ = 0;
+ front_ = new_front;
+ return h;
+ }
+ return 0;
+ }
+
+ // Push a handler on to the back of the queue.
+ void push(handler* h)
+ {
+ node* n = h->node_;
+ h->node_ = 0;
+ n->version_ = next_version_;
+ next_version_ += 2;
+ n->handler_ = h;
+ n->next_ = 0;
+ back_->next_ = n;
+ back_ = n;
+ }
+
+private:
+ // Template wrapper for handlers.
+ template <typename Handler>
+ class handler_wrapper
+ : public handler
+ {
+ public:
+ handler_wrapper(Handler h)
+ : handler(
+ &handler_wrapper<Handler>::do_call,
+ &handler_wrapper<Handler>::do_destroy),
+ handler_(h)
+ {
+ }
+
+ static void do_call(handler* base)
+ {
+ // Take ownership of the handler object.
+ typedef handler_wrapper<Handler> this_type;
+ this_type* h(static_cast<this_type*>(base));
+ typedef handler_alloc_traits<Handler, this_type> alloc_traits;
+ handler_ptr<alloc_traits> ptr(h->handler_, h);
+
+ // Make a copy of the handler so that the memory can be deallocated before
+ // the upcall is made.
+ Handler handler(h->handler_);
+
+ // Free the memory associated with the handler.
+ ptr.reset();
+
+ // Make the upcall.
+ boost_asio_handler_invoke_helpers::invoke(handler, &handler);
+ }
+
+ static void do_destroy(handler* base)
+ {
+ // Take ownership of the handler object.
+ typedef handler_wrapper<Handler> this_type;
+ this_type* h(static_cast<this_type*>(base));
+ typedef handler_alloc_traits<Handler, this_type> alloc_traits;
+ handler_ptr<alloc_traits> ptr(h->handler_, h);
+ }
+
+ private:
+ Handler handler_;
+ };
+
+ // The front of the queue.
+ node* front_;
+
+ // The back of the queue.
+ node* back_;
+
+ // The next version counter to be assigned to a node.
+ unsigned long next_version_;
+};
+
+} // namespace detail
+} // namespace asio
+} // namespace boost
+
+#include <boost/asio/detail/pop_options.hpp>
+
+#endif // BOOST_ASIO_DETAIL_INDIRECT_HANDLER_QUEUE_HPP
Modified: trunk/boost/asio/detail/task_io_service.hpp
==============================================================================
--- trunk/boost/asio/detail/task_io_service.hpp (original)
+++ trunk/boost/asio/detail/task_io_service.hpp 2008-04-28 09:56:07 EDT (Mon, 28 Apr 2008)
@@ -15,6 +15,10 @@
# pragma once
#endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
+#if defined(BOOST_ASIO_ENABLE_TWO_LOCK_QUEUE)
+#include <boost/asio/detail/task_io_service_2lock.hpp>
+#else // defined(BOOST_ASIO_ENABLE_TWO_LOCK_QUEUE)
+
#include <boost/asio/detail/push_options.hpp>
#include <boost/asio/io_service.hpp>
@@ -418,4 +422,6 @@
#include <boost/system/error_code.hpp>
#include <boost/asio/detail/pop_options.hpp>
+#endif // defined(BOOST_ASIO_ENABLE_TWO_LOCK_QUEUE)
+
#endif // BOOST_ASIO_DETAIL_TASK_IO_SERVICE_HPP
Added: trunk/boost/asio/detail/task_io_service_2lock.hpp
==============================================================================
--- (empty file)
+++ trunk/boost/asio/detail/task_io_service_2lock.hpp 2008-04-28 09:56:07 EDT (Mon, 28 Apr 2008)
@@ -0,0 +1,464 @@
+//
+// task_io_service_2lock.hpp
+// ~~~~~~~~~~~~~~~~~~~~~~~~~
+//
+// Copyright (c) 2003-2008 Christopher M. Kohlhoff (chris at kohlhoff dot com)
+//
+// 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_ASIO_DETAIL_TASK_IO_SERVICE_2LOCK_HPP
+#define BOOST_ASIO_DETAIL_TASK_IO_SERVICE_2LOCK_HPP
+
+#if defined(_MSC_VER) && (_MSC_VER >= 1200)
+# pragma once
+#endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
+
+#include <boost/asio/detail/push_options.hpp>
+
+#include <boost/asio/io_service.hpp>
+#include <boost/asio/detail/call_stack.hpp>
+#include <boost/asio/detail/event.hpp>
+#include <boost/asio/detail/handler_alloc_helpers.hpp>
+#include <boost/asio/detail/handler_invoke_helpers.hpp>
+#include <boost/asio/detail/indirect_handler_queue.hpp>
+#include <boost/asio/detail/mutex.hpp>
+#include <boost/asio/detail/service_base.hpp>
+#include <boost/asio/detail/task_io_service_fwd.hpp>
+
+#include <boost/asio/detail/push_options.hpp>
+#include <boost/detail/atomic_count.hpp>
+#include <boost/system/error_code.hpp>
+#include <boost/asio/detail/pop_options.hpp>
+
+namespace boost {
+namespace asio {
+namespace detail {
+
+// An alternative task_io_service implementation based on a two-lock queue.
+
+template <typename Task>
+class task_io_service
+ : public boost::asio::detail::service_base<task_io_service<Task> >
+{
+public:
+ typedef indirect_handler_queue handler_queue;
+
+ // Constructor.
+ task_io_service(boost::asio::io_service& io_service)
+ : boost::asio::detail::service_base<task_io_service<Task> >(io_service),
+ front_mutex_(),
+ back_mutex_(),
+ task_(use_service<Task>(io_service)),
+ outstanding_work_(0),
+ front_stopped_(false),
+ back_stopped_(false),
+ back_shutdown_(false),
+ back_first_idle_thread_(0),
+ back_task_thread_(0)
+ {
+ handler_queue_.push(&task_handler_);
+ }
+
+ void init(size_t /*concurrency_hint*/)
+ {
+ }
+
+ // Destroy all user-defined handler objects owned by the service.
+ void shutdown_service()
+ {
+ boost::asio::detail::mutex::scoped_lock back_lock(back_mutex_);
+ back_shutdown_ = true;
+ back_lock.unlock();
+
+ // Destroy handler objects.
+ while (handler_queue::handler* h = handler_queue_.pop())
+ if (h != &task_handler_)
+ h->destroy();
+
+ // Reset handler queue to initial state.
+ handler_queue_.push(&task_handler_);
+ }
+
+ // Run the event loop until interrupted or no more work.
+ size_t run(boost::system::error_code& ec)
+ {
+ if (outstanding_work_ == 0)
+ {
+ stop();
+ ec = boost::system::error_code();
+ return 0;
+ }
+
+ typename call_stack<task_io_service>::context ctx(this);
+
+ idle_thread_info this_idle_thread;
+ this_idle_thread.next = 0;
+
+ size_t n = 0;
+ while (do_one(&this_idle_thread, ec))
+ if (n != (std::numeric_limits<size_t>::max)())
+ ++n;
+ return n;
+ }
+
+ // Run until interrupted or one operation is performed.
+ size_t run_one(boost::system::error_code& ec)
+ {
+ if (outstanding_work_ == 0)
+ {
+ stop();
+ ec = boost::system::error_code();
+ return 0;
+ }
+
+ typename call_stack<task_io_service>::context ctx(this);
+
+ idle_thread_info this_idle_thread;
+ this_idle_thread.next = 0;
+
+ return do_one(&this_idle_thread, ec);
+ }
+
+ // Poll for operations without blocking.
+ size_t poll(boost::system::error_code& ec)
+ {
+ if (outstanding_work_ == 0)
+ {
+ stop();
+ ec = boost::system::error_code();
+ return 0;
+ }
+
+ typename call_stack<task_io_service>::context ctx(this);
+
+ size_t n = 0;
+ while (do_one(0, ec))
+ if (n != (std::numeric_limits<size_t>::max)())
+ ++n;
+ return n;
+ }
+
+ // Poll for one operation without blocking.
+ size_t poll_one(boost::system::error_code& ec)
+ {
+ if (outstanding_work_ == 0)
+ {
+ stop();
+ ec = boost::system::error_code();
+ return 0;
+ }
+
+ typename call_stack<task_io_service>::context ctx(this);
+
+ return do_one(0, ec);
+ }
+
+ // Interrupt the event processing loop.
+ void stop()
+ {
+ boost::asio::detail::mutex::scoped_lock front_lock(front_mutex_);
+ front_stopped_ = true;
+ front_lock.unlock();
+
+ boost::asio::detail::mutex::scoped_lock back_lock(back_mutex_);
+ back_stopped_ = true;
+ interrupt_all_idle_threads(back_lock);
+ }
+
+ // Reset in preparation for a subsequent run invocation.
+ void reset()
+ {
+ boost::asio::detail::mutex::scoped_lock front_lock(front_mutex_);
+ front_stopped_ = false;
+ front_lock.unlock();
+
+ boost::asio::detail::mutex::scoped_lock back_lock(back_mutex_);
+ back_stopped_ = false;
+ }
+
+ // Notify that some work has started.
+ void work_started()
+ {
+ ++outstanding_work_;
+ }
+
+ // Notify that some work has finished.
+ void work_finished()
+ {
+ if (--outstanding_work_ == 0)
+ stop();
+ }
+
+ // Request invocation of the given handler.
+ template <typename Handler>
+ void dispatch(Handler handler)
+ {
+ if (call_stack<task_io_service>::contains(this))
+ boost_asio_handler_invoke_helpers::invoke(handler, &handler);
+ else
+ post(handler);
+ }
+
+ // Request invocation of the given handler and return immediately.
+ template <typename Handler>
+ void post(Handler handler)
+ {
+ // Allocate and construct an operation to wrap the handler.
+ handler_queue::scoped_ptr ptr(handler_queue::wrap(handler));
+
+ boost::asio::detail::mutex::scoped_lock back_lock(back_mutex_);
+
+ // If the service has been shut down we silently discard the handler.
+ if (back_shutdown_)
+ return;
+
+ // Add the handler to the end of the queue.
+ handler_queue_.push(ptr.get());
+ ptr.release();
+
+ // An undelivered handler is treated as unfinished work.
+ ++outstanding_work_;
+
+ // Wake up a thread to execute the handler.
+ interrupt_one_idle_thread(back_lock);
+ }
+
+private:
+ struct idle_thread_info;
+
+ size_t do_one(idle_thread_info* this_idle_thread,
+ boost::system::error_code& ec)
+ {
+ bool task_has_run = false;
+ for (;;)
+ {
+ // The front lock must be held before we can pop items from the queue.
+ boost::asio::detail::mutex::scoped_lock front_lock(front_mutex_);
+ if (front_stopped_)
+ {
+ ec = boost::system::error_code();
+ return 0;
+ }
+
+ if (handler_queue::handler* h = handler_queue_.pop())
+ {
+ if (h == &task_handler_)
+ {
+ bool more_handlers = handler_queue_.poppable();
+ unsigned long front_version = handler_queue_.front_version();
+ front_lock.unlock();
+
+ // The task is always added to the back of the queue when we exit
+ // this block.
+ task_cleanup c(*this);
+
+ // If we're polling and the task has already run then we're done.
+ bool polling = !this_idle_thread;
+ if (task_has_run && polling)
+ {
+ ec = boost::system::error_code();
+ return 0;
+ }
+
+ // If we're considering going idle we need to check whether the queue
+ // is still empty. If it is, add the thread to the list of idle
+ // threads.
+ if (!more_handlers && !polling)
+ {
+ boost::asio::detail::mutex::scoped_lock back_lock(back_mutex_);
+ if (back_stopped_)
+ {
+ ec = boost::system::error_code();
+ return 0;
+ }
+ else if (front_version == handler_queue_.back_version())
+ {
+ back_task_thread_ = this_idle_thread;
+ }
+ else
+ {
+ more_handlers = true;
+ }
+ }
+
+ // Run the task. May throw an exception. Only block if the handler
+ // queue is empty and we're not polling, otherwise we want to return
+ // as soon as possible.
+ task_has_run = true;
+ task_.run(!more_handlers && !polling);
+ }
+ else
+ {
+ front_lock.unlock();
+ handler_cleanup c(*this);
+
+ // Invoke the handler. May throw an exception.
+ h->invoke(); // invoke() deletes the handler object
+
+ ec = boost::system::error_code();
+ return 1;
+ }
+ }
+ else if (this_idle_thread)
+ {
+ unsigned long front_version = handler_queue_.front_version();
+ front_lock.unlock();
+
+ // If we're considering going idle we need to check whether the queue
+ // is still empty. If it is, add the thread to the list of idle
+ // threads.
+ boost::asio::detail::mutex::scoped_lock back_lock(back_mutex_);
+ if (back_stopped_)
+ {
+ ec = boost::system::error_code();
+ return 0;
+ }
+ else if (front_version == handler_queue_.back_version())
+ {
+ this_idle_thread->next = back_first_idle_thread_;
+ back_first_idle_thread_ = this_idle_thread;
+ this_idle_thread->wakeup_event.clear(back_lock);
+ this_idle_thread->wakeup_event.wait(back_lock);
+ }
+ }
+ else
+ {
+ ec = boost::system::error_code();
+ return 0;
+ }
+ }
+ }
+
+ // Interrupt a single idle thread.
+ void interrupt_one_idle_thread(
+ boost::asio::detail::mutex::scoped_lock& back_lock)
+ {
+ if (back_first_idle_thread_)
+ {
+ idle_thread_info* idle_thread = back_first_idle_thread_;
+ back_first_idle_thread_ = idle_thread->next;
+ idle_thread->next = 0;
+ idle_thread->wakeup_event.signal(back_lock);
+ }
+ else if (back_task_thread_)
+ {
+ back_task_thread_ = 0;
+ task_.interrupt();
+ }
+ }
+
+ // Interrupt all idle threads.
+ void interrupt_all_idle_threads(
+ boost::asio::detail::mutex::scoped_lock& back_lock)
+ {
+ while (back_first_idle_thread_)
+ {
+ idle_thread_info* idle_thread = back_first_idle_thread_;
+ back_first_idle_thread_ = idle_thread->next;
+ idle_thread->next = 0;
+ idle_thread->wakeup_event.signal(back_lock);
+ }
+
+ if (back_task_thread_)
+ {
+ back_task_thread_ = 0;
+ task_.interrupt();
+ }
+ }
+
+ // Helper class to perform task-related operations on block exit.
+ class task_cleanup;
+ friend class task_cleanup;
+ class task_cleanup
+ {
+ public:
+ task_cleanup(task_io_service& task_io_svc)
+ : task_io_service_(task_io_svc)
+ {
+ }
+
+ ~task_cleanup()
+ {
+ // Reinsert the task at the end of the handler queue.
+ boost::asio::detail::mutex::scoped_lock back_lock(
+ task_io_service_.back_mutex_);
+ task_io_service_.back_task_thread_ = 0;
+ task_io_service_.handler_queue_.push(&task_io_service_.task_handler_);
+ }
+
+ private:
+ task_io_service& task_io_service_;
+ };
+
+ // Helper class to perform handler-related operations on block exit.
+ class handler_cleanup
+ {
+ public:
+ handler_cleanup(task_io_service& task_io_svc)
+ : task_io_service_(task_io_svc)
+ {
+ }
+
+ ~handler_cleanup()
+ {
+ task_io_service_.work_finished();
+ }
+
+ private:
+ task_io_service& task_io_service_;
+ };
+
+ // Mutexes to protect access to internal data.
+ boost::asio::detail::mutex front_mutex_;
+ boost::asio::detail::mutex back_mutex_;
+
+ // The task to be run by this service.
+ Task& task_;
+
+ // Handler object to represent the position of the task in the queue.
+ class task_handler
+ : public handler_queue::handler
+ {
+ public:
+ task_handler()
+ : handler_queue::handler(0, 0)
+ {
+ }
+ } task_handler_;
+
+ // The count of unfinished work.
+ boost::detail::atomic_count outstanding_work_;
+
+ // The queue of handlers that are ready to be delivered.
+ handler_queue handler_queue_;
+
+ // Flag to indicate that the dispatcher has been stopped.
+ bool front_stopped_;
+ bool back_stopped_;
+
+ // Flag to indicate that the dispatcher has been shut down.
+ bool back_shutdown_;
+
+ // Structure containing information about an idle thread.
+ struct idle_thread_info
+ {
+ event wakeup_event;
+ idle_thread_info* next;
+ };
+
+ // The number of threads that are currently idle.
+ idle_thread_info* back_first_idle_thread_;
+
+ // The thread that is currently blocked on the task.
+ idle_thread_info* back_task_thread_;
+};
+
+} // namespace detail
+} // namespace asio
+} // namespace boost
+
+#include <boost/asio/detail/pop_options.hpp>
+
+#endif // BOOST_ASIO_DETAIL_TASK_IO_SERVICE_2LOCK_HPP