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),