[Boost-commit] svn:boost r42758 - trunk/boost/asio/detail

From: chris_at_[hidden]
Date: 2008-01-14 08:27:53

Author: chris_kohlhoff
Date: 2008-01-14 08:27:52 EST (Mon, 14 Jan 2008)
New Revision: 42758
URL: http://svn.boost.org/trac/boost/changeset/42758

Log:
Silence some integer truncation warnings.

Only perform the windows-bug workaround where we use a short timeout with
GetQueuedCompletionStatus from one thread, i.e. the timer thread.

Keep track of the number of OVERLAPPED-derived operations to ensure that
they all get cleaned up when the io_service is destroyed.

Removed:
   trunk/boost/asio/detail/win_iocp_operation.hpp
Text files modified: 
   trunk/boost/asio/detail/win_iocp_io_service.hpp     |   104 ++++++++++++++++++++++++++++++++------- 
   trunk/boost/asio/detail/win_iocp_socket_service.hpp |    49 +++++++++---------                      
   2 files changed, 108 insertions(+), 45 deletions(-)

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	2008-01-14 08:27:52 EST (Mon, 14 Jan 2008)
@@ -34,7 +34,6 @@
 #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 {
@@ -45,14 +44,64 @@
   : public boost::asio::detail::service_base<win_iocp_io_service>
 {
 public:
-  // Base class for all operations.
-  typedef win_iocp_operation operation;
+  // Base class for all operations. A function pointer is used instead of
+  // virtual functions to avoid the associated overhead.
+  //
+  // This class inherits from OVERLAPPED so that we can downcast to get back to
+  // the operation pointer from the LPOVERLAPPED out parameter of
+  // GetQueuedCompletionStatus.
+  class operation
+    : public OVERLAPPED
+  {
+  public:
+    typedef void (*invoke_func_type)(operation*, DWORD, size_t);
+    typedef void (*destroy_func_type)(operation*);
+
+    operation(win_iocp_io_service& iocp_service,
+        invoke_func_type invoke_func, destroy_func_type destroy_func)
+      : outstanding_operations_(&iocp_service.outstanding_operations_),
+        invoke_func_(invoke_func),
+        destroy_func_(destroy_func)
+    {
+      Internal = 0;
+      InternalHigh = 0;
+      Offset = 0;
+      OffsetHigh = 0;
+      hEvent = 0;
+
+      ::InterlockedIncrement(outstanding_operations_);
+    }
+
+    void do_completion(DWORD last_error, size_t bytes_transferred)
+    {
+      invoke_func_(this, last_error, bytes_transferred);
+    }
+
+    void destroy()
+    {
+      destroy_func_(this);
+    }
+
+  protected:
+    // Prevent deletion through this type.
+    ~operation()
+    {
+      ::InterlockedDecrement(outstanding_operations_);
+    }
+
+  private:
+    long* outstanding_operations_;
+    invoke_func_type invoke_func_;
+    destroy_func_type destroy_func_;
+  };
+
 
   // Constructor.
   win_iocp_io_service(boost::asio::io_service& io_service)
     : boost::asio::detail::service_base<win_iocp_io_service>(io_service),
       iocp_(),
       outstanding_work_(0),
+      outstanding_operations_(0),
       stopped_(0),
       shutdown_(0),
       timer_thread_(0),
@@ -80,7 +129,7 @@
   {
     ::InterlockedExchange(&shutdown_, 1);
 
-    for (;;)
+    while (::InterlockedExchangeAdd(&outstanding_operations_, 0) > 0)
     {
       DWORD bytes_transferred = 0;
 #if (WINVER < 0x0500)
@@ -89,12 +138,8 @@
       DWORD_PTR completion_key = 0;
 #endif
       LPOVERLAPPED overlapped = 0;
-      ::SetLastError(0);
-      BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle,
-          &bytes_transferred, &completion_key, &overlapped, 0);
-      DWORD last_error = ::GetLastError();
-      if (!ok && overlapped == 0 && last_error == WAIT_TIMEOUT)
-        break;
+      ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
+          &completion_key, &overlapped, INFINITE);
       if (overlapped)
         static_cast<operation*>(overlapped)->destroy();
     }
@@ -250,7 +295,7 @@
   }
 
   // Request invocation of the given OVERLAPPED-derived operation.
-  void post_completion(win_iocp_operation* op, DWORD op_last_error,
+  void post_completion(operation* op, DWORD op_last_error,
       DWORD bytes_transferred)
   {
     // Enqueue the operation on the I/O completion port.
@@ -348,7 +393,7 @@
             &timer_thread_, this_thread_id, 0) == 0);
 
       // Calculate timeout for GetQueuedCompletionStatus call.
-      DWORD timeout = max_timeout;
+      DWORD timeout = INFINITE;
       if (dispatching_timers)
       {
         boost::asio::detail::mutex::scoped_lock lock(timer_mutex_);
@@ -372,13 +417,28 @@
       // Dispatch any pending timers.
       if (dispatching_timers)
       {
-        boost::asio::detail::mutex::scoped_lock lock(timer_mutex_);
-        timer_queues_copy_ = timer_queues_;
-        for (std::size_t i = 0; i < timer_queues_.size(); ++i)
+        try
         {
-          timer_queues_[i]->dispatch_timers();
-          timer_queues_[i]->dispatch_cancellations();
-          timer_queues_[i]->cleanup_timers();
+          boost::asio::detail::mutex::scoped_lock lock(timer_mutex_);
+          timer_queues_copy_ = timer_queues_;
+          for (std::size_t i = 0; i < timer_queues_.size(); ++i)
+          {
+            timer_queues_[i]->dispatch_timers();
+            timer_queues_[i]->dispatch_cancellations();
+            timer_queues_[i]->cleanup_timers();
+          }
+        }
+        catch (...)
+        {
+          // Transfer responsibility for dispatching timers to another thread.
+          if (::InterlockedCompareExchange(&timer_thread_,
+                0, this_thread_id) == this_thread_id)
+          {
+            ::PostQueuedCompletionStatus(iocp_.handle,
+                0, transfer_timer_dispatching, 0);
+          }
+
+          throw;
         }
       }
 
@@ -533,7 +593,7 @@
   {
     handler_operation(win_iocp_io_service& io_service,
         Handler handler)
-      : operation(&handler_operation<Handler>::do_completion_impl,
+      : operation(io_service, &handler_operation<Handler>::do_completion_impl,
           &handler_operation<Handler>::destroy_impl),
         io_service_(io_service),
         handler_(handler)
@@ -594,6 +654,10 @@
   // The count of unfinished work.
   long outstanding_work_;
 
+  // The count of unfinished operations.
+  long outstanding_operations_;
+  friend class operation;
+
   // Flag to indicate whether the event loop has been stopped.
   long stopped_;
 
@@ -603,7 +667,7 @@
   enum
   {
     // Maximum GetQueuedCompletionStatus timeout, in milliseconds.
-    max_timeout = 1000,
+    max_timeout = 500,
 
     // Completion key value to indicate that responsibility for dispatching
     // timers is being cooperatively transferred from one thread to another.

Deleted: trunk/boost/asio/detail/win_iocp_operation.hpp
==============================================================================
--- trunk/boost/asio/detail/win_iocp_operation.hpp	2008-01-14 08:27:52 EST (Mon, 14 Jan 2008)
+++ (empty file)
@@ -1,83 +0,0 @@
-//
-// win_iocp_operation.hpp
-// ~~~~~~~~~~~~~~~~~~~~~~
-//
-// Copyright (c) 2003-2007 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_WIN_IOCP_OPERATION_HPP
-#define BOOST_ASIO_DETAIL_WIN_IOCP_OPERATION_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/win_iocp_io_service_fwd.hpp>
-
-#if defined(BOOST_ASIO_HAS_IOCP)
-
-#include <boost/asio/detail/socket_types.hpp>
-
-namespace boost {
-namespace asio {
-namespace detail {
-
-// Base class for all IOCP operations. A function pointer is used instead of
-// virtual functions to avoid the associated overhead.
-//
-// This class inherits from OVERLAPPED so that we can downcast to get back to
-// the win_iocp_operation pointer from the LPOVERLAPPED out parameter of
-// GetQueuedCompletionStatus.
-struct win_iocp_operation
-  : public OVERLAPPED
-{
-  typedef void (*invoke_func_type)(win_iocp_operation*, DWORD, size_t);
-  typedef void (*destroy_func_type)(win_iocp_operation*);
-
-  win_iocp_operation(invoke_func_type invoke_func,
-      destroy_func_type destroy_func)
-    : invoke_func_(invoke_func),
-      destroy_func_(destroy_func)
-  {
-    Internal = 0;
-    InternalHigh = 0;
-    Offset = 0;
-    OffsetHigh = 0;
-    hEvent = 0;
-  }
-
-  void do_completion(DWORD last_error, size_t bytes_transferred)
-  {
-    invoke_func_(this, last_error, bytes_transferred);
-  }
-
-  void destroy()
-  {
-    destroy_func_(this);
-  }
-
-protected:
-  // Prevent deletion through this type.
-  ~win_iocp_operation()
-  {
-  }
-
-private:
-  invoke_func_type invoke_func_;
-  destroy_func_type destroy_func_;
-};
-
-} // namespace detail
-} // namespace asio
-} // namespace boost
-
-#endif // defined(BOOST_ASIO_HAS_IOCP)
-
-#include <boost/asio/detail/pop_options.hpp>
-
-#endif // BOOST_ASIO_DETAIL_WIN_IOCP_OPERATION_HPP

Modified: trunk/boost/asio/detail/win_iocp_socket_service.hpp
==============================================================================
--- trunk/boost/asio/detail/win_iocp_socket_service.hpp	(original)
+++ trunk/boost/asio/detail/win_iocp_socket_service.hpp	2008-01-14 08:27:52 EST (Mon, 14 Jan 2008)
@@ -57,7 +57,7 @@
   typedef typename Protocol::endpoint endpoint_type;
 
   // Base class for all operations.
-  typedef win_iocp_operation operation;
+  typedef win_iocp_io_service::operation operation;
 
   struct noop_deleter { void operator()(void*) {} };
   typedef boost::shared_ptr<void> shared_cancel_token_type;
@@ -681,13 +681,13 @@
     : public operation
   {
   public:
-    send_operation(boost::asio::io_service& io_service,
+    send_operation(win_iocp_io_service& io_service,
         weak_cancel_token_type cancel_token,
         const ConstBufferSequence& buffers, Handler handler)
-      : operation(
+      : operation(io_service,
           &send_operation<ConstBufferSequence, Handler>::do_completion_impl,
           &send_operation<ConstBufferSequence, Handler>::destroy_impl),
-        work_(io_service),
+        work_(io_service.get_io_service()),
         cancel_token_(cancel_token),
         buffers_(buffers),
         handler_(handler)
@@ -783,8 +783,8 @@
     typedef send_operation<ConstBufferSequence, Handler> value_type;
     typedef handler_alloc_traits<Handler, value_type> alloc_traits;
     raw_handler_ptr<alloc_traits> raw_ptr(handler);
-    handler_ptr<alloc_traits> ptr(raw_ptr,
-        this->get_io_service(), impl.cancel_token_, buffers, handler);
+    handler_ptr<alloc_traits> ptr(raw_ptr, iocp_service_,
+        impl.cancel_token_, buffers, handler);
 
     // Copy buffers into WSABUF array.
     ::WSABUF bufs[max_buffers];
@@ -861,7 +861,7 @@
     // Send the data.
     DWORD bytes_transferred = 0;
     int result = ::WSASendTo(impl.socket_, bufs, i, &bytes_transferred,
-        flags, destination.data(), destination.size(), 0, 0);
+        flags, destination.data(), static_cast<int>(destination.size()), 0, 0);
     if (result != 0)
     {
       DWORD last_error = ::WSAGetLastError();
@@ -881,12 +881,12 @@
     : public operation
   {
   public:
-    send_to_operation(boost::asio::io_service& io_service,
+    send_to_operation(win_iocp_io_service& io_service,
         const ConstBufferSequence& buffers, Handler handler)
-      : operation(
+      : operation(io_service,
           &send_to_operation<ConstBufferSequence, Handler>::do_completion_impl,
           &send_to_operation<ConstBufferSequence, Handler>::destroy_impl),
-        work_(io_service),
+        work_(io_service.get_io_service()),
         buffers_(buffers),
         handler_(handler)
     {
@@ -974,8 +974,7 @@
     typedef send_to_operation<ConstBufferSequence, Handler> value_type;
     typedef handler_alloc_traits<Handler, value_type> alloc_traits;
     raw_handler_ptr<alloc_traits> raw_ptr(handler);
-    handler_ptr<alloc_traits> ptr(raw_ptr,
-        this->get_io_service(), buffers, handler);
+    handler_ptr<alloc_traits> ptr(raw_ptr, iocp_service_, buffers, handler);
 
     // Copy buffers into WSABUF array.
     ::WSABUF bufs[max_buffers];
@@ -992,8 +991,8 @@
 
     // Send the data.
     DWORD bytes_transferred = 0;
-    int result = ::WSASendTo(impl.socket_, bufs, i, &bytes_transferred,
-        flags, destination.data(), destination.size(), ptr.get(), 0);
+    int result = ::WSASendTo(impl.socket_, bufs, i, &bytes_transferred, flags,
+        destination.data(), static_cast<int>(destination.size()), ptr.get(), 0);
     DWORD last_error = ::WSAGetLastError();
 
     // Check if the operation completed immediately.
@@ -1075,15 +1074,15 @@
     : public operation
   {
   public:
-    receive_operation(boost::asio::io_service& io_service,
+    receive_operation(win_iocp_io_service& io_service,
         weak_cancel_token_type cancel_token,
         const MutableBufferSequence& buffers, Handler handler)
-      : operation(
+      : operation(io_service,
           &receive_operation<
             MutableBufferSequence, Handler>::do_completion_impl,
           &receive_operation<
             MutableBufferSequence, Handler>::destroy_impl),
-        work_(io_service),
+        work_(io_service.get_io_service()),
         cancel_token_(cancel_token),
         buffers_(buffers),
         handler_(handler)
@@ -1186,8 +1185,8 @@
     typedef receive_operation<MutableBufferSequence, Handler> value_type;
     typedef handler_alloc_traits<Handler, value_type> alloc_traits;
     raw_handler_ptr<alloc_traits> raw_ptr(handler);
-    handler_ptr<alloc_traits> ptr(raw_ptr,
-        this->get_io_service(), impl.cancel_token_, buffers, handler);
+    handler_ptr<alloc_traits> ptr(raw_ptr, iocp_service_,
+        impl.cancel_token_, buffers, handler);
 
     // Copy buffers into WSABUF array.
     ::WSABUF bufs[max_buffers];
@@ -1291,17 +1290,17 @@
     : public operation
   {
   public:
-    receive_from_operation(boost::asio::io_service& io_service,
+    receive_from_operation(win_iocp_io_service& io_service,
         endpoint_type& endpoint, const MutableBufferSequence& buffers,
         Handler handler)
-      : operation(
+      : operation(io_service,
           &receive_from_operation<
             MutableBufferSequence, Handler>::do_completion_impl,
           &receive_from_operation<
             MutableBufferSequence, Handler>::destroy_impl),
         endpoint_(endpoint),
         endpoint_size_(static_cast<int>(endpoint.capacity())),
-        work_(io_service),
+        work_(io_service.get_io_service()),
         buffers_(buffers),
         handler_(handler)
     {
@@ -1406,8 +1405,8 @@
     typedef receive_from_operation<MutableBufferSequence, Handler> value_type;
     typedef handler_alloc_traits<Handler, value_type> alloc_traits;
     raw_handler_ptr<alloc_traits> raw_ptr(handler);
-    handler_ptr<alloc_traits> ptr(raw_ptr,
-        this->get_io_service(), sender_endp, buffers, handler);
+    handler_ptr<alloc_traits> ptr(raw_ptr, iocp_service_,
+        sender_endp, buffers, handler);
 
     // Copy buffers into WSABUF array.
     ::WSABUF bufs[max_buffers];
@@ -1509,7 +1508,7 @@
         socket_type socket, socket_type new_socket, Socket& peer,
         const protocol_type& protocol, endpoint_type* peer_endpoint,
         bool enable_connection_aborted, Handler handler)
-      : operation(
+      : operation(io_service,
           &accept_operation<Socket, Handler>::do_completion_impl,
           &accept_operation<Socket, Handler>::destroy_impl),
         io_service_(io_service),