// | |
// detail/impl/win_iocp_io_service.ipp | |
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
// | |
// Copyright (c) 2003-2011 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_IMPL_WIN_IOCP_IO_SERVICE_IPP | |
#define BOOST_ASIO_DETAIL_IMPL_WIN_IOCP_IO_SERVICE_IPP | |
#if defined(_MSC_VER) && (_MSC_VER >= 1200) | |
# pragma once | |
#endif // defined(_MSC_VER) && (_MSC_VER >= 1200) | |
#include <boost/asio/detail/config.hpp> | |
#if defined(BOOST_ASIO_HAS_IOCP) | |
#include <boost/limits.hpp> | |
#include <boost/asio/error.hpp> | |
#include <boost/asio/io_service.hpp> | |
#include <boost/asio/detail/handler_alloc_helpers.hpp> | |
#include <boost/asio/detail/handler_invoke_helpers.hpp> | |
#include <boost/asio/detail/throw_error.hpp> | |
#include <boost/asio/detail/win_iocp_io_service.hpp> | |
#include <boost/asio/detail/push_options.hpp> | |
namespace boost { | |
namespace asio { | |
namespace detail { | |
struct win_iocp_io_service::work_finished_on_block_exit | |
{ | |
~work_finished_on_block_exit() | |
{ | |
io_service_->work_finished(); | |
} | |
win_iocp_io_service* io_service_; | |
}; | |
struct win_iocp_io_service::timer_thread_function | |
{ | |
void operator()() | |
{ | |
while (::InterlockedExchangeAdd(&io_service_->shutdown_, 0) == 0) | |
{ | |
if (::WaitForSingleObject(io_service_->waitable_timer_.handle, | |
INFINITE) == WAIT_OBJECT_0) | |
{ | |
::InterlockedExchange(&io_service_->dispatch_required_, 1); | |
::PostQueuedCompletionStatus(io_service_->iocp_.handle, | |
0, wake_for_dispatch, 0); | |
} | |
} | |
} | |
win_iocp_io_service* io_service_; | |
}; | |
win_iocp_io_service::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), | |
stopped_(0), | |
shutdown_(0), | |
dispatch_required_(0) | |
{ | |
} | |
void win_iocp_io_service::init(size_t concurrency_hint) | |
{ | |
iocp_.handle = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, | |
static_cast<DWORD>((std::min<size_t>)(concurrency_hint, DWORD(~0)))); | |
if (!iocp_.handle) | |
{ | |
DWORD last_error = ::GetLastError(); | |
boost::system::error_code ec(last_error, | |
boost::asio::error::get_system_category()); | |
boost::asio::detail::throw_error(ec, "iocp"); | |
} | |
} | |
void win_iocp_io_service::shutdown_service() | |
{ | |
::InterlockedExchange(&shutdown_, 1); | |
if (timer_thread_) | |
{ | |
LARGE_INTEGER timeout; | |
timeout.QuadPart = 1; | |
::SetWaitableTimer(waitable_timer_.handle, &timeout, 1, 0, 0, FALSE); | |
} | |
while (::InterlockedExchangeAdd(&outstanding_work_, 0) > 0) | |
{ | |
op_queue<win_iocp_operation> ops; | |
timer_queues_.get_all_timers(ops); | |
ops.push(completed_ops_); | |
if (!ops.empty()) | |
{ | |
while (win_iocp_operation* op = ops.front()) | |
{ | |
ops.pop(); | |
::InterlockedDecrement(&outstanding_work_); | |
op->destroy(); | |
} | |
} | |
else | |
{ | |
DWORD bytes_transferred = 0; | |
dword_ptr_t completion_key = 0; | |
LPOVERLAPPED overlapped = 0; | |
::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred, | |
&completion_key, &overlapped, gqcs_timeout); | |
if (overlapped) | |
{ | |
::InterlockedDecrement(&outstanding_work_); | |
static_cast<win_iocp_operation*>(overlapped)->destroy(); | |
} | |
} | |
} | |
if (timer_thread_) | |
timer_thread_->join(); | |
} | |
boost::system::error_code win_iocp_io_service::register_handle( | |
HANDLE handle, boost::system::error_code& ec) | |
{ | |
if (::CreateIoCompletionPort(handle, iocp_.handle, 0, 0) == 0) | |
{ | |
DWORD last_error = ::GetLastError(); | |
ec = boost::system::error_code(last_error, | |
boost::asio::error::get_system_category()); | |
} | |
else | |
{ | |
ec = boost::system::error_code(); | |
} | |
return ec; | |
} | |
size_t win_iocp_io_service::run(boost::system::error_code& ec) | |
{ | |
if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0) | |
{ | |
stop(); | |
ec = boost::system::error_code(); | |
return 0; | |
} | |
call_stack<win_iocp_io_service>::context ctx(this); | |
size_t n = 0; | |
while (do_one(true, ec)) | |
if (n != (std::numeric_limits<size_t>::max)()) | |
++n; | |
return n; | |
} | |
size_t win_iocp_io_service::run_one(boost::system::error_code& ec) | |
{ | |
if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0) | |
{ | |
stop(); | |
ec = boost::system::error_code(); | |
return 0; | |
} | |
call_stack<win_iocp_io_service>::context ctx(this); | |
return do_one(true, ec); | |
} | |
size_t win_iocp_io_service::poll(boost::system::error_code& ec) | |
{ | |
if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0) | |
{ | |
stop(); | |
ec = boost::system::error_code(); | |
return 0; | |
} | |
call_stack<win_iocp_io_service>::context ctx(this); | |
size_t n = 0; | |
while (do_one(false, ec)) | |
if (n != (std::numeric_limits<size_t>::max)()) | |
++n; | |
return n; | |
} | |
size_t win_iocp_io_service::poll_one(boost::system::error_code& ec) | |
{ | |
if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0) | |
{ | |
stop(); | |
ec = boost::system::error_code(); | |
return 0; | |
} | |
call_stack<win_iocp_io_service>::context ctx(this); | |
return do_one(false, ec); | |
} | |
void win_iocp_io_service::stop() | |
{ | |
if (::InterlockedExchange(&stopped_, 1) == 0) | |
{ | |
if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0)) | |
{ | |
DWORD last_error = ::GetLastError(); | |
boost::system::error_code ec(last_error, | |
boost::asio::error::get_system_category()); | |
boost::asio::detail::throw_error(ec, "pqcs"); | |
} | |
} | |
} | |
void win_iocp_io_service::post_deferred_completion(win_iocp_operation* op) | |
{ | |
// Flag the operation as ready. | |
op->ready_ = 1; | |
// Enqueue the operation on the I/O completion port. | |
if (!::PostQueuedCompletionStatus(iocp_.handle, | |
0, overlapped_contains_result, op)) | |
{ | |
// Out of resources. Put on completed queue instead. | |
mutex::scoped_lock lock(dispatch_mutex_); | |
completed_ops_.push(op); | |
::InterlockedExchange(&dispatch_required_, 1); | |
} | |
} | |
void win_iocp_io_service::post_deferred_completions( | |
op_queue<win_iocp_operation>& ops) | |
{ | |
while (win_iocp_operation* op = ops.front()) | |
{ | |
ops.pop(); | |
// Flag the operation as ready. | |
op->ready_ = 1; | |
// Enqueue the operation on the I/O completion port. | |
if (!::PostQueuedCompletionStatus(iocp_.handle, | |
0, overlapped_contains_result, op)) | |
{ | |
// Out of resources. Put on completed queue instead. | |
mutex::scoped_lock lock(dispatch_mutex_); | |
completed_ops_.push(op); | |
completed_ops_.push(ops); | |
::InterlockedExchange(&dispatch_required_, 1); | |
} | |
} | |
} | |
void win_iocp_io_service::on_pending(win_iocp_operation* op) | |
{ | |
if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1) | |
{ | |
// Enqueue the operation on the I/O completion port. | |
if (!::PostQueuedCompletionStatus(iocp_.handle, | |
0, overlapped_contains_result, op)) | |
{ | |
// Out of resources. Put on completed queue instead. | |
mutex::scoped_lock lock(dispatch_mutex_); | |
completed_ops_.push(op); | |
::InterlockedExchange(&dispatch_required_, 1); | |
} | |
} | |
} | |
void win_iocp_io_service::on_completion(win_iocp_operation* op, | |
DWORD last_error, DWORD bytes_transferred) | |
{ | |
// Flag that the operation is ready for invocation. | |
op->ready_ = 1; | |
// Store results in the OVERLAPPED structure. | |
op->Internal = reinterpret_cast<ulong_ptr_t>( | |
&boost::asio::error::get_system_category()); | |
op->Offset = last_error; | |
op->OffsetHigh = bytes_transferred; | |
// Enqueue the operation on the I/O completion port. | |
if (!::PostQueuedCompletionStatus(iocp_.handle, | |
0, overlapped_contains_result, op)) | |
{ | |
// Out of resources. Put on completed queue instead. | |
mutex::scoped_lock lock(dispatch_mutex_); | |
completed_ops_.push(op); | |
::InterlockedExchange(&dispatch_required_, 1); | |
} | |
} | |
void win_iocp_io_service::on_completion(win_iocp_operation* op, | |
const boost::system::error_code& ec, DWORD bytes_transferred) | |
{ | |
// Flag that the operation is ready for invocation. | |
op->ready_ = 1; | |
// Store results in the OVERLAPPED structure. | |
op->Internal = reinterpret_cast<ulong_ptr_t>(&ec.category()); | |
op->Offset = ec.value(); | |
op->OffsetHigh = bytes_transferred; | |
// Enqueue the operation on the I/O completion port. | |
if (!::PostQueuedCompletionStatus(iocp_.handle, | |
0, overlapped_contains_result, op)) | |
{ | |
// Out of resources. Put on completed queue instead. | |
mutex::scoped_lock lock(dispatch_mutex_); | |
completed_ops_.push(op); | |
::InterlockedExchange(&dispatch_required_, 1); | |
} | |
} | |
size_t win_iocp_io_service::do_one(bool block, boost::system::error_code& ec) | |
{ | |
for (;;) | |
{ | |
// Try to acquire responsibility for dispatching timers and completed ops. | |
if (::InterlockedCompareExchange(&dispatch_required_, 0, 1) == 1) | |
{ | |
mutex::scoped_lock lock(dispatch_mutex_); | |
// Dispatch pending timers and operations. | |
op_queue<win_iocp_operation> ops; | |
ops.push(completed_ops_); | |
timer_queues_.get_ready_timers(ops); | |
post_deferred_completions(ops); | |
update_timeout(); | |
} | |
// Get the next operation from the queue. | |
DWORD bytes_transferred = 0; | |
dword_ptr_t completion_key = 0; | |
LPOVERLAPPED overlapped = 0; | |
::SetLastError(0); | |
BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred, | |
&completion_key, &overlapped, block ? gqcs_timeout : 0); | |
DWORD last_error = ::GetLastError(); | |
if (overlapped) | |
{ | |
win_iocp_operation* op = static_cast<win_iocp_operation*>(overlapped); | |
boost::system::error_code result_ec(last_error, | |
boost::asio::error::get_system_category()); | |
// We may have been passed the last_error and bytes_transferred in the | |
// OVERLAPPED structure itself. | |
if (completion_key == overlapped_contains_result) | |
{ | |
result_ec = boost::system::error_code(static_cast<int>(op->Offset), | |
*reinterpret_cast<boost::system::error_category*>(op->Internal)); | |
bytes_transferred = op->OffsetHigh; | |
} | |
// Otherwise ensure any result has been saved into the OVERLAPPED | |
// structure. | |
else | |
{ | |
op->Internal = reinterpret_cast<ulong_ptr_t>(&result_ec.category()); | |
op->Offset = result_ec.value(); | |
op->OffsetHigh = bytes_transferred; | |
} | |
// Dispatch the operation only if ready. The operation may not be ready | |
// if the initiating function (e.g. a call to WSARecv) has not yet | |
// returned. This is because the initiating function still wants access | |
// to the operation's OVERLAPPED structure. | |
if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1) | |
{ | |
// Ensure the count of outstanding work is decremented on block exit. | |
work_finished_on_block_exit on_exit = { this }; | |
(void)on_exit; | |
op->complete(*this, result_ec, bytes_transferred); | |
ec = boost::system::error_code(); | |
return 1; | |
} | |
} | |
else if (!ok) | |
{ | |
if (last_error != WAIT_TIMEOUT) | |
{ | |
ec = boost::system::error_code(last_error, | |
boost::asio::error::get_system_category()); | |
return 0; | |
} | |
// If we're not polling we need to keep going until we get a real handler. | |
if (block) | |
continue; | |
ec = boost::system::error_code(); | |
return 0; | |
} | |
else if (completion_key == wake_for_dispatch) | |
{ | |
// We have been woken up to try to acquire responsibility for dispatching | |
// timers and completed operations. | |
} | |
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) | |
{ | |
// Wake up next thread that is blocked on GetQueuedCompletionStatus. | |
if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0)) | |
{ | |
last_error = ::GetLastError(); | |
ec = boost::system::error_code(last_error, | |
boost::asio::error::get_system_category()); | |
return 0; | |
} | |
ec = boost::system::error_code(); | |
return 0; | |
} | |
} | |
} | |
} | |
void win_iocp_io_service::do_add_timer_queue(timer_queue_base& queue) | |
{ | |
mutex::scoped_lock lock(dispatch_mutex_); | |
timer_queues_.insert(&queue); | |
if (!waitable_timer_.handle) | |
{ | |
waitable_timer_.handle = ::CreateWaitableTimer(0, FALSE, 0); | |
if (waitable_timer_.handle == 0) | |
{ | |
DWORD last_error = ::GetLastError(); | |
boost::system::error_code ec(last_error, | |
boost::asio::error::get_system_category()); | |
boost::asio::detail::throw_error(ec, "timer"); | |
} | |
LARGE_INTEGER timeout; | |
timeout.QuadPart = -max_timeout_usec; | |
timeout.QuadPart *= 10; | |
::SetWaitableTimer(waitable_timer_.handle, | |
&timeout, max_timeout_msec, 0, 0, FALSE); | |
} | |
if (!timer_thread_) | |
{ | |
timer_thread_function thread_function = { this }; | |
timer_thread_.reset(new thread(thread_function, 65536)); | |
} | |
} | |
void win_iocp_io_service::do_remove_timer_queue(timer_queue_base& queue) | |
{ | |
mutex::scoped_lock lock(dispatch_mutex_); | |
timer_queues_.erase(&queue); | |
} | |
void win_iocp_io_service::update_timeout() | |
{ | |
if (timer_thread_) | |
{ | |
// There's no point updating the waitable timer if the new timeout period | |
// exceeds the maximum timeout. In that case, we might as well wait for the | |
// existing period of the timer to expire. | |
long timeout_usec = timer_queues_.wait_duration_usec(max_timeout_usec); | |
if (timeout_usec < max_timeout_usec) | |
{ | |
LARGE_INTEGER timeout; | |
timeout.QuadPart = -timeout_usec; | |
timeout.QuadPart *= 10; | |
::SetWaitableTimer(waitable_timer_.handle, | |
&timeout, max_timeout_msec, 0, 0, FALSE); | |
} | |
} | |
} | |
} // namespace detail | |
} // namespace asio | |
} // namespace boost | |
#include <boost/asio/detail/pop_options.hpp> | |
#endif // defined(BOOST_ASIO_HAS_IOCP) | |
#endif // BOOST_ASIO_DETAIL_IMPL_WIN_IOCP_IO_SERVICE_IPP |