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

From: chris_at_[hidden]
Date: 2007-11-07 23:10:14

Author: chris_kohlhoff
Date: 2007-11-07 23:10:14 EST (Wed, 07 Nov 2007)
New Revision: 40919
URL: http://svn.boost.org/trac/boost/changeset/40919

Log:
Eliminate the need for an extra thread to perform timer dispatching.

Text files modified: 
   trunk/boost/asio/deadline_timer_service.hpp     |     3                                         
   trunk/boost/asio/detail/win_iocp_io_service.hpp |   197 +++++++++++++++++++++++++++++++++++++++ 
   2 files changed, 195 insertions(+), 5 deletions(-)

Modified: trunk/boost/asio/deadline_timer_service.hpp
==============================================================================
--- trunk/boost/asio/deadline_timer_service.hpp	(original)
+++ trunk/boost/asio/deadline_timer_service.hpp	2007-11-07 23:10:14 EST (Wed, 07 Nov 2007)
@@ -29,6 +29,7 @@
 #include <boost/asio/detail/kqueue_reactor.hpp>
 #include <boost/asio/detail/select_reactor.hpp>
 #include <boost/asio/detail/service_base.hpp>
+#include <boost/asio/detail/win_iocp_io_service.hpp>
 
 namespace boost {
 namespace asio {
@@ -63,7 +64,7 @@
   // The type of the platform-specific implementation.
 #if defined(BOOST_ASIO_HAS_IOCP)
   typedef detail::deadline_timer_service<
-    traits_type, detail::select_reactor<true> > service_impl_type;
+    traits_type, detail::win_iocp_io_service> service_impl_type;
 #elif defined(BOOST_ASIO_HAS_EPOLL)
   typedef detail::deadline_timer_service<
     traits_type, detail::epoll_reactor<false> > service_impl_type;

Modified: trunk/boost/asio/detail/win_iocp_io_service.hpp
==============================================================================
--- trunk/boost/asio/detail/win_iocp_io_service.hpp	(original)
+++ trunk/boost/asio/detail/win_iocp_io_service.hpp	2007-11-07 23:10:14 EST (Wed, 07 Nov 2007)
@@ -33,7 +33,9 @@
 #include <boost/asio/detail/handler_invoke_helpers.hpp>
 #include <boost/asio/detail/service_base.hpp>
 #include <boost/asio/detail/socket_types.hpp>
+#include <boost/asio/detail/timer_queue.hpp>
 #include <boost/asio/detail/win_iocp_operation.hpp>
+#include <boost/asio/detail/mutex.hpp>
 
 namespace boost {
 namespace asio {
@@ -52,7 +54,9 @@
       iocp_(),
       outstanding_work_(0),
       stopped_(0),
-      shutdown_(0)
+      shutdown_(0),
+      timer_thread_(0),
+      timer_interrupt_issued_(false)
   {
   }
 
@@ -94,6 +98,10 @@
       if (overlapped)
         static_cast<operation*>(overlapped)->destroy();
     }
+
+    for (std::size_t i = 0; i < timer_queues_.size(); ++i)
+      timer_queues_[i]->destroy_timers();
+    timer_queues_.clear();
   }
 
   // Register a handle with the IO completion port.
@@ -258,14 +266,96 @@
     }
   }
 
+  // Add a new timer queue to the service.
+  template <typename Time_Traits>
+  void add_timer_queue(timer_queue<Time_Traits>& timer_queue)
+  {
+    boost::asio::detail::mutex::scoped_lock lock(timer_mutex_);
+    timer_queues_.push_back(&timer_queue);
+  }
+
+  // Remove a timer queue from the service.
+  template <typename Time_Traits>
+  void remove_timer_queue(timer_queue<Time_Traits>& timer_queue)
+  {
+    boost::asio::detail::mutex::scoped_lock lock(timer_mutex_);
+    for (std::size_t i = 0; i < timer_queues_.size(); ++i)
+    {
+      if (timer_queues_[i] == &timer_queue)
+      {
+        timer_queues_.erase(timer_queues_.begin() + i);
+        return;
+      }
+    }
+  }
+
+  // Schedule a timer in the given timer queue to expire at the specified
+  // absolute time. The handler object will be invoked when the timer expires.
+  template <typename Time_Traits, typename Handler>
+  void schedule_timer(timer_queue<Time_Traits>& timer_queue,
+      const typename Time_Traits::time_type& time, Handler handler, void* token)
+  {
+    // If the service has been shut down we silently discard the timer.
+    if (::InterlockedExchangeAdd(&shutdown_, 0) != 0)
+      return;
+
+    boost::asio::detail::mutex::scoped_lock lock(timer_mutex_);
+    if (timer_queue.enqueue_timer(time, handler, token))
+    {
+      if (!timer_interrupt_issued_)
+      {
+        timer_interrupt_issued_ = true;
+        lock.unlock();
+        ::PostQueuedCompletionStatus(iocp_.handle, 0, 1, 0);
+      }
+    }
+  }
+
+  // Cancel the timer associated with the given token. Returns the number of
+  // handlers that have been posted or dispatched.
+  template <typename Time_Traits>
+  std::size_t cancel_timer(timer_queue<Time_Traits>& timer_queue, void* token)
+  {
+    // If the service has been shut down we silently ignore the cancellation.
+    if (::InterlockedExchangeAdd(&shutdown_, 0) != 0)
+      return 0;
+
+    boost::asio::detail::mutex::scoped_lock lock(timer_mutex_);
+    std::size_t n = timer_queue.cancel_timer(token);
+    if (n > 0 && !timer_interrupt_issued_)
+    {
+      timer_interrupt_issued_ = true;
+      lock.unlock();
+      ::PostQueuedCompletionStatus(iocp_.handle, 0, 1, 0);
+    }
+    return n;
+  }
+
 private:
   // Dequeues at most one operation from the I/O completion port, and then
   // executes it. Returns the number of operations that were dequeued (i.e.
   // either 0 or 1).
   size_t do_one(bool block, boost::system::error_code& ec)
   {
+    bool doing_timers = false;
     for (;;)
     {
+      // Try to become the timer thread.
+      if (!doing_timers)
+      {
+        doing_timers = (InterlockedCompareExchange(&timer_thread_,
+              static_cast<long>(GetCurrentThreadId()), 0) == 0);
+      }
+
+      // Calculate timeout for GetQueuedCompletionStatus call.
+      DWORD timeout = 1000;
+      if (doing_timers)
+      {
+        boost::asio::detail::mutex::scoped_lock lock(timer_mutex_);
+        timer_interrupt_issued_ = false;
+        timeout = get_timeout();
+      }
+
       // Get the next operation from the queue.
       DWORD bytes_transferred = 0;
 #if (WINVER < 0x0500)
@@ -276,18 +366,45 @@
       LPOVERLAPPED overlapped = 0;
       ::SetLastError(0);
       BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
-          &completion_key, &overlapped, block ? 1000 : 0);
+          &completion_key, &overlapped, block ? timeout : 0);
       DWORD last_error = ::GetLastError();
 
+      // Dispatch any pending timers.
+      if (doing_timers)
+      {
+        boost::asio::detail::mutex::scoped_lock lock(timer_mutex_);
+        for (std::size_t i = 0; i < timer_queues_.size(); ++i)
+        {
+          timer_queues_[i]->dispatch_timers();
+          timer_queues_[i]->dispatch_cancellations();
+        }
+
+        // Clean up timers. We must not hold the lock while cleaning up timers
+        // since the destructors may make calls back into this service. We make
+        // a copy of the vector of timer queues since the original may be
+        // modified while the lock is not held.
+        timer_queues_for_cleanup_ = timer_queues_;
+        lock.unlock();
+        for (std::size_t i = 0; i < timer_queues_for_cleanup_.size(); ++i)
+          timer_queues_for_cleanup_[i]->cleanup_timers();
+      }
+
       if (!ok && overlapped == 0)
       {
         if (block && last_error == WAIT_TIMEOUT)
           continue;
+
+        // Pass responsibility for timers to another thread.
+        if (doing_timers)
+        {
+          ::InterlockedExchange(&timer_thread_, 0);
+          ::PostQueuedCompletionStatus(iocp_.handle, 0, 1, 0);
+        }
+
         ec = boost::system::error_code();
         return 0;
       }
-
-      if (overlapped)
+      else if (overlapped)
       {
         // We may have been passed a last_error value in the completion_key.
         if (last_error == 0)
@@ -295,6 +412,13 @@
           last_error = completion_key;
         }
 
+        // Pass responsibility for timers to another thread.
+        if (doing_timers)
+        {
+          ::InterlockedExchange(&timer_thread_, 0);
+          ::PostQueuedCompletionStatus(iocp_.handle, 0, 1, 0);
+        }
+
         // Ensure that the io_service does not exit due to running out of work
         // while we make the upcall.
         auto_work work(*this);
@@ -306,12 +430,22 @@
         ec = boost::system::error_code();
         return 1;
       }
+      else if (completion_key == 1)
+      {
+        // Woken up to try to become the timer thread.
+      }
       else
       {
         // The stopped_ flag is always checked to ensure that any leftover
         // interrupts from a previous run invocation are ignored.
         if (::InterlockedExchangeAdd(&stopped_, 0) != 0)
         {
+          // Pass responsibility for timers to another thread.
+          if (doing_timers)
+          {
+            ::InterlockedExchange(&timer_thread_, 0);
+          }
+
           // Wake up next thread that is blocked on GetQueuedCompletionStatus.
           if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0))
           {
@@ -328,6 +462,45 @@
     }
   }
 
+  // Check if all timer queues are empty.
+  bool all_timer_queues_are_empty() const
+  {
+    for (std::size_t i = 0; i < timer_queues_.size(); ++i)
+      if (!timer_queues_[i]->empty())
+        return false;
+    return true;
+  }
+
+  // Get the timeout value for the GetQueuedCompletionStatus call. The timeout
+  // value is returned as a number of milliseconds. We will wait no longer than
+  // 1000 milliseconds.
+  DWORD get_timeout()
+  {
+    if (all_timer_queues_are_empty())
+      return 1000;
+
+    boost::posix_time::time_duration minimum_wait_duration
+      = boost::posix_time::seconds(1);
+
+    for (std::size_t i = 0; i < timer_queues_.size(); ++i)
+    {
+      boost::posix_time::time_duration wait_duration
+        = timer_queues_[i]->wait_duration();
+      if (wait_duration < minimum_wait_duration)
+        minimum_wait_duration = wait_duration;
+    }
+
+    if (minimum_wait_duration > boost::posix_time::time_duration())
+    {
+      int milliseconds = minimum_wait_duration.total_milliseconds();
+      return static_cast<DWORD>(milliseconds > 0 ? milliseconds : 1);
+    }
+    else
+    {
+      return 0;
+    }
+  }
+
   struct auto_work
   {
     auto_work(win_iocp_io_service& io_service)
@@ -417,6 +590,22 @@
 
   // Flag to indicate whether the service has been shut down.
   long shutdown_;
+
+  // The thread that's currently in charge of dispatching timers.
+  long timer_thread_;
+
+  // Mutex for protecting access to the timer queues.
+  mutex timer_mutex_;
+
+  // Whether a thread has been interrupted to process a new timeout.
+  bool timer_interrupt_issued_;
+
+  // The timer queues.
+  std::vector<timer_queue_base*> timer_queues_;
+
+  // A copy of the timer queues, used when cleaning up timers. The copy is
+  // stored as a class data member to avoid unnecessary memory allocation.
+  std::vector<timer_queue_base*> timer_queues_for_cleanup_;
 };
 
 } // namespace detail