boost/asio/detail/win_iocp_io_service.hpp
// // win_iocp_io_service.hpp // ~~~~~~~~~~~~~~~~~~~~~~~ // // Copyright (c) 2003-2010 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_IO_SERVICE_HPP #define BOOST_ASIO_DETAIL_WIN_IOCP_IO_SERVICE_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/push_options.hpp> #include <boost/limits.hpp> #include <boost/throw_exception.hpp> #include <boost/system/system_error.hpp> #include <boost/asio/detail/pop_options.hpp> #include <boost/asio/io_service.hpp> #include <boost/asio/detail/call_stack.hpp> #include <boost/asio/detail/completion_handler.hpp> #include <boost/asio/detail/fenced_block.hpp> #include <boost/asio/detail/handler_alloc_helpers.hpp> #include <boost/asio/detail/handler_invoke_helpers.hpp> #include <boost/asio/detail/mutex.hpp> #include <boost/asio/detail/op_queue.hpp> #include <boost/asio/detail/service_base.hpp> #include <boost/asio/detail/socket_types.hpp> #include <boost/asio/detail/timer_op.hpp> #include <boost/asio/detail/timer_queue_base.hpp> #include <boost/asio/detail/timer_queue_fwd.hpp> #include <boost/asio/detail/timer_queue_set.hpp> #include <boost/asio/detail/win_iocp_operation.hpp> namespace boost { namespace asio { namespace detail { class timer_op; class win_iocp_io_service : public boost::asio::detail::service_base<win_iocp_io_service> { public: typedef win_iocp_operation operation; // 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), stopped_(0), shutdown_(0), timer_thread_(0), timer_interrupt_issued_(false) { } void 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::system_error e( boost::system::error_code(last_error, boost::asio::error::get_system_category()), "iocp"); boost::throw_exception(e); } } // Destroy all user-defined handler objects owned by the service. void shutdown_service() { ::InterlockedExchange(&shutdown_, 1); while (::InterlockedExchangeAdd(&outstanding_work_, 0) > 0) { op_queue<operation> ops; timer_queues_.get_all_timers(ops); ops.push(completed_ops_); if (!ops.empty()) { while (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, max_timeout); if (overlapped) { ::InterlockedDecrement(&outstanding_work_); static_cast<operation*>(overlapped)->destroy(); } } } } // Initialise the task. Nothing to do here. void init_task() { } // Register a handle with the IO completion port. boost::system::error_code 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; } // Run the event loop until stopped or no more work. size_t 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; } // Run until stopped or one operation is performed. size_t 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); } // Poll for operations without blocking. size_t 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; } // Poll for one operation without blocking. size_t 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); } // Stop the event processing loop. void stop() { if (::InterlockedExchange(&stopped_, 1) == 0) { if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0)) { DWORD last_error = ::GetLastError(); boost::system::system_error e( boost::system::error_code(last_error, boost::asio::error::get_system_category()), "pqcs"); boost::throw_exception(e); } } } // Reset in preparation for a subsequent run invocation. void reset() { ::InterlockedExchange(&stopped_, 0); } // Notify that some work has started. void work_started() { ::InterlockedIncrement(&outstanding_work_); } // Notify that some work has finished. void work_finished() { if (::InterlockedDecrement(&outstanding_work_) == 0) stop(); } // Request invocation of the given handler. template <typename Handler> void dispatch(Handler handler) { if (call_stack<win_iocp_io_service>::contains(this)) { boost::asio::detail::fenced_block b; boost_asio_handler_invoke_helpers::invoke(handler, handler); } else post(handler); } // Request invocation of the given handler and return immediately. template <typename Handler> void post(Handler handler) { // Allocate and construct an operation to wrap the handler. typedef completion_handler<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, handler); post_immediate_completion(ptr.get()); ptr.release(); } // Request invocation of the given operation and return immediately. Assumes // that work_started() has not yet been called for the operation. void post_immediate_completion(operation* op) { work_started(); post_deferred_completion(op); } // Request invocation of the given operation and return immediately. Assumes // that work_started() was previously called for the operation. void post_deferred_completion(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. boost::asio::detail::mutex::scoped_lock lock(timer_mutex_); completed_ops_.push(op); } } // Request invocation of the given operation and return immediately. Assumes // that work_started() was previously called for the operations. void post_deferred_completions(op_queue<operation>& ops) { while (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. boost::asio::detail::mutex::scoped_lock lock(timer_mutex_); completed_ops_.push(op); completed_ops_.push(ops); } } } // Called after starting an overlapped I/O operation that did not complete // immediately. The caller must have already called work_started() prior to // starting the operation. void on_pending(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. boost::asio::detail::mutex::scoped_lock lock(timer_mutex_); completed_ops_.push(op); } } } // Called after starting an overlapped I/O operation that completed // immediately. The caller must have already called work_started() prior to // starting the operation. void on_completion(operation* op, DWORD last_error = 0, DWORD bytes_transferred = 0) { // 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. boost::asio::detail::mutex::scoped_lock lock(timer_mutex_); completed_ops_.push(op); } } // Called after starting an overlapped I/O operation that completed // immediately. The caller must have already called work_started() prior to // starting the operation. void on_completion(operation* op, const boost::system::error_code& ec, DWORD bytes_transferred = 0) { // 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. boost::asio::detail::mutex::scoped_lock lock(timer_mutex_); completed_ops_.push(op); } } // 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_.insert(&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_); timer_queues_.erase(&timer_queue); } // Schedule a new operation in the given timer queue to expire at the // specified absolute time. template <typename Time_Traits> void schedule_timer(timer_queue<Time_Traits>& timer_queue, const typename Time_Traits::time_type& time, timer_op* op, 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_); bool interrupt = timer_queue.enqueue_timer(time, op, token); work_started(); if (interrupt && !timer_interrupt_issued_) { timer_interrupt_issued_ = true; lock.unlock(); ::PostQueuedCompletionStatus(iocp_.handle, 0, steal_timer_dispatching, 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_); op_queue<operation> ops; std::size_t n = timer_queue.cancel_timer(token, ops); post_deferred_completions(ops); if (n > 0 && !timer_interrupt_issued_) { timer_interrupt_issued_ = true; lock.unlock(); ::PostQueuedCompletionStatus(iocp_.handle, 0, steal_timer_dispatching, 0); } return n; } private: #if defined(WINVER) && (WINVER < 0x0500) typedef DWORD dword_ptr_t; typedef ULONG ulong_ptr_t; #else // defined(WINVER) && (WINVER < 0x0500) typedef DWORD_PTR dword_ptr_t; typedef ULONG_PTR ulong_ptr_t; #endif // defined(WINVER) && (WINVER < 0x0500) // 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) { long this_thread_id = static_cast<long>(::GetCurrentThreadId()); for (;;) { // Try to acquire responsibility for dispatching timers. bool dispatching_timers = (::InterlockedCompareExchange( &timer_thread_, this_thread_id, 0) == 0); // Calculate timeout for GetQueuedCompletionStatus call. DWORD timeout = max_timeout; if (dispatching_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; dword_ptr_t completion_key = 0; LPOVERLAPPED overlapped = 0; ::SetLastError(0); BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred, &completion_key, &overlapped, block ? timeout : 0); DWORD last_error = ::GetLastError(); // Dispatch any pending timers. if (dispatching_timers) { boost::asio::detail::mutex::scoped_lock lock(timer_mutex_); op_queue<operation> ops; ops.push(completed_ops_); timer_queues_.get_ready_timers(ops); post_deferred_completions(ops); } if (!ok && overlapped == 0) { if (block && last_error == WAIT_TIMEOUT) { // Relinquish responsibility for dispatching timers. if (dispatching_timers) { ::InterlockedCompareExchange(&timer_thread_, 0, this_thread_id); } continue; } // Transfer responsibility for dispatching timers to another thread. if (dispatching_timers && ::InterlockedCompareExchange( &timer_thread_, 0, this_thread_id) == this_thread_id) { ::PostQueuedCompletionStatus(iocp_.handle, 0, transfer_timer_dispatching, 0); } ec = boost::system::error_code(); return 0; } else if (overlapped) { operation* op = static_cast<operation*>(overlapped); boost::system::error_code result_ec(last_error, boost::asio::error::get_system_category()); // Transfer responsibility for dispatching timers to another thread. if (dispatching_timers && ::InterlockedCompareExchange( &timer_thread_, 0, this_thread_id) == this_thread_id) { ::PostQueuedCompletionStatus(iocp_.handle, 0, transfer_timer_dispatching, 0); } // 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 (completion_key == transfer_timer_dispatching) { // Woken up to try to acquire responsibility for dispatching timers. ::InterlockedCompareExchange(&timer_thread_, 0, this_thread_id); } else if (completion_key == steal_timer_dispatching) { // Woken up to steal responsibility for dispatching timers. ::InterlockedExchange(&timer_thread_, 0); } else { // Relinquish responsibility for dispatching timers. If the io_service // is not being stopped then the thread will get an opportunity to // reacquire timer responsibility on the next loop iteration. if (dispatching_timers) { ::InterlockedCompareExchange(&timer_thread_, 0, this_thread_id); } // 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; } } } } // 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() { return timer_queues_.wait_duration_msec(max_timeout); } // Helper class to call work_finished() on block exit. struct work_finished_on_block_exit { ~work_finished_on_block_exit() { io_service_->work_finished(); } win_iocp_io_service* io_service_; }; // The IO completion port used for queueing operations. struct iocp_holder { HANDLE handle; iocp_holder() : handle(0) {} ~iocp_holder() { if (handle) ::CloseHandle(handle); } } iocp_; // The count of unfinished work. long outstanding_work_; // Flag to indicate whether the event loop has been stopped. long stopped_; // Flag to indicate whether the service has been shut down. long shutdown_; enum { // Maximum GetQueuedCompletionStatus timeout, in milliseconds. max_timeout = 500, // Completion key value to indicate that responsibility for dispatching // timers is being cooperatively transferred from one thread to another. transfer_timer_dispatching = 1, // Completion key value to indicate that responsibility for dispatching // timers should be stolen from another thread. steal_timer_dispatching = 2, // Completion key value to indicate that an operation has posted with the // original last_error and bytes_transferred values stored in the fields of // the OVERLAPPED structure. overlapped_contains_result = 3 }; // 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. timer_queue_set timer_queues_; // The operations that are ready to dispatch. op_queue<operation> completed_ops_; }; } // 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_IO_SERVICE_HPP