diff options
Diffstat (limited to 'lib/asio/detail/impl/win_iocp_io_context.ipp')
| -rw-r--r-- | lib/asio/detail/impl/win_iocp_io_context.ipp | 554 |
1 files changed, 554 insertions, 0 deletions
diff --git a/lib/asio/detail/impl/win_iocp_io_context.ipp b/lib/asio/detail/impl/win_iocp_io_context.ipp new file mode 100644 index 0000000..c371b86 --- /dev/null +++ b/lib/asio/detail/impl/win_iocp_io_context.ipp @@ -0,0 +1,554 @@ +// +// detail/impl/win_iocp_io_context.ipp +// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +// +// Copyright (c) 2003-2018 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 ASIO_DETAIL_IMPL_WIN_IOCP_IO_CONTEXT_IPP +#define ASIO_DETAIL_IMPL_WIN_IOCP_IO_CONTEXT_IPP + +#if defined(_MSC_VER) && (_MSC_VER >= 1200) +# pragma once +#endif // defined(_MSC_VER) && (_MSC_VER >= 1200) + +#include "asio/detail/config.hpp" + +#if defined(ASIO_HAS_IOCP) + +#include "asio/error.hpp" +#include "asio/detail/cstdint.hpp" +#include "asio/detail/handler_alloc_helpers.hpp" +#include "asio/detail/handler_invoke_helpers.hpp" +#include "asio/detail/limits.hpp" +#include "asio/detail/throw_error.hpp" +#include "asio/detail/win_iocp_io_context.hpp" + +#include "asio/detail/push_options.hpp" + +namespace asio { +namespace detail { + +struct win_iocp_io_context::work_finished_on_block_exit +{ + ~work_finished_on_block_exit() + { + io_context_->work_finished(); + } + + win_iocp_io_context* io_context_; +}; + +struct win_iocp_io_context::timer_thread_function +{ + void operator()() + { + while (::InterlockedExchangeAdd(&io_context_->shutdown_, 0) == 0) + { + if (::WaitForSingleObject(io_context_->waitable_timer_.handle, + INFINITE) == WAIT_OBJECT_0) + { + ::InterlockedExchange(&io_context_->dispatch_required_, 1); + ::PostQueuedCompletionStatus(io_context_->iocp_.handle, + 0, wake_for_dispatch, 0); + } + } + } + + win_iocp_io_context* io_context_; +}; + +win_iocp_io_context::win_iocp_io_context( + asio::execution_context& ctx, int concurrency_hint) + : execution_context_service_base<win_iocp_io_context>(ctx), + iocp_(), + outstanding_work_(0), + stopped_(0), + stop_event_posted_(0), + shutdown_(0), + gqcs_timeout_(get_gqcs_timeout()), + dispatch_required_(0), + concurrency_hint_(concurrency_hint) +{ + ASIO_HANDLER_TRACKING_INIT; + + iocp_.handle = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, + static_cast<DWORD>(concurrency_hint >= 0 ? concurrency_hint : DWORD(~0))); + if (!iocp_.handle) + { + DWORD last_error = ::GetLastError(); + asio::error_code ec(last_error, + asio::error::get_system_category()); + asio::detail::throw_error(ec, "iocp"); + } +} + +void win_iocp_io_context::shutdown() +{ + ::InterlockedExchange(&shutdown_, 1); + + if (timer_thread_.get()) + { + 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_.get()) + timer_thread_->join(); +} + +asio::error_code win_iocp_io_context::register_handle( + HANDLE handle, asio::error_code& ec) +{ + if (::CreateIoCompletionPort(handle, iocp_.handle, 0, 0) == 0) + { + DWORD last_error = ::GetLastError(); + ec = asio::error_code(last_error, + asio::error::get_system_category()); + } + else + { + ec = asio::error_code(); + } + return ec; +} + +size_t win_iocp_io_context::run(asio::error_code& ec) +{ + if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0) + { + stop(); + ec = asio::error_code(); + return 0; + } + + win_iocp_thread_info this_thread; + thread_call_stack::context ctx(this, this_thread); + + size_t n = 0; + while (do_one(INFINITE, ec)) + if (n != (std::numeric_limits<size_t>::max)()) + ++n; + return n; +} + +size_t win_iocp_io_context::run_one(asio::error_code& ec) +{ + if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0) + { + stop(); + ec = asio::error_code(); + return 0; + } + + win_iocp_thread_info this_thread; + thread_call_stack::context ctx(this, this_thread); + + return do_one(INFINITE, ec); +} + +size_t win_iocp_io_context::wait_one(long usec, asio::error_code& ec) +{ + if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0) + { + stop(); + ec = asio::error_code(); + return 0; + } + + win_iocp_thread_info this_thread; + thread_call_stack::context ctx(this, this_thread); + + return do_one(usec < 0 ? INFINITE : ((usec - 1) / 1000 + 1), ec); +} + +size_t win_iocp_io_context::poll(asio::error_code& ec) +{ + if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0) + { + stop(); + ec = asio::error_code(); + return 0; + } + + win_iocp_thread_info this_thread; + thread_call_stack::context ctx(this, this_thread); + + size_t n = 0; + while (do_one(0, ec)) + if (n != (std::numeric_limits<size_t>::max)()) + ++n; + return n; +} + +size_t win_iocp_io_context::poll_one(asio::error_code& ec) +{ + if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0) + { + stop(); + ec = asio::error_code(); + return 0; + } + + win_iocp_thread_info this_thread; + thread_call_stack::context ctx(this, this_thread); + + return do_one(0, ec); +} + +void win_iocp_io_context::stop() +{ + if (::InterlockedExchange(&stopped_, 1) == 0) + { + if (::InterlockedExchange(&stop_event_posted_, 1) == 0) + { + if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0)) + { + DWORD last_error = ::GetLastError(); + asio::error_code ec(last_error, + asio::error::get_system_category()); + asio::detail::throw_error(ec, "pqcs"); + } + } + } +} + +void win_iocp_io_context::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, 0, 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_context::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, 0, 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_context::abandon_operations( + op_queue<win_iocp_operation>& ops) +{ + while (win_iocp_operation* op = ops.front()) + { + ops.pop(); + ::InterlockedDecrement(&outstanding_work_); + op->destroy(); + } +} + +void win_iocp_io_context::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_context::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>( + &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_context::on_completion(win_iocp_operation* op, + const asio::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_context::do_one(DWORD msec, asio::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, + msec < gqcs_timeout_ ? msec : gqcs_timeout_); + DWORD last_error = ::GetLastError(); + + if (overlapped) + { + win_iocp_operation* op = static_cast<win_iocp_operation*>(overlapped); + asio::error_code result_ec(last_error, + 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 = asio::error_code(static_cast<int>(op->Offset), + *reinterpret_cast<asio::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 = asio::error_code(); + return 1; + } + } + else if (!ok) + { + if (last_error != WAIT_TIMEOUT) + { + ec = asio::error_code(last_error, + asio::error::get_system_category()); + return 0; + } + + // If we're waiting indefinitely we need to keep going until we get a + // real handler. + if (msec == INFINITE) + continue; + + ec = asio::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 + { + // Indicate that there is no longer an in-flight stop event. + ::InterlockedExchange(&stop_event_posted_, 0); + + // The stopped_ flag is always checked to ensure that any leftover + // stop events from a previous run invocation are ignored. + if (::InterlockedExchangeAdd(&stopped_, 0) != 0) + { + // Wake up next thread that is blocked on GetQueuedCompletionStatus. + if (::InterlockedExchange(&stop_event_posted_, 1) == 0) + { + if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0)) + { + last_error = ::GetLastError(); + ec = asio::error_code(last_error, + asio::error::get_system_category()); + return 0; + } + } + + ec = asio::error_code(); + return 0; + } + } + } +} + +DWORD win_iocp_io_context::get_gqcs_timeout() +{ + OSVERSIONINFOEX osvi; + ZeroMemory(&osvi, sizeof(osvi)); + osvi.dwOSVersionInfoSize = sizeof(osvi); + osvi.dwMajorVersion = 6ul; + + const uint64_t condition_mask = ::VerSetConditionMask( + 0, VER_MAJORVERSION, VER_GREATER_EQUAL); + + if (!!::VerifyVersionInfo(&osvi, VER_MAJORVERSION, condition_mask)) + return INFINITE; + + return default_gqcs_timeout; +} + +void win_iocp_io_context::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(); + asio::error_code ec(last_error, + asio::error::get_system_category()); + 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_.get()) + { + timer_thread_function thread_function = { this }; + timer_thread_.reset(new thread(thread_function, 65536)); + } +} + +void win_iocp_io_context::do_remove_timer_queue(timer_queue_base& queue) +{ + mutex::scoped_lock lock(dispatch_mutex_); + + timer_queues_.erase(&queue); +} + +void win_iocp_io_context::update_timeout() +{ + if (timer_thread_.get()) + { + // 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 + +#include "asio/detail/pop_options.hpp" + +#endif // defined(ASIO_HAS_IOCP) + +#endif // ASIO_DETAIL_IMPL_WIN_IOCP_IO_CONTEXT_IPP |