2 // detail/impl/win_iocp_io_service.ipp
3 // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
5 // Copyright (c) 2003-2016 Christopher M. Kohlhoff (chris at kohlhoff dot com)
7 // Distributed under the Boost Software License, Version 1.0. (See accompanying
8 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
11 #ifndef BOOST_ASIO_DETAIL_IMPL_WIN_IOCP_IO_SERVICE_IPP
12 #define BOOST_ASIO_DETAIL_IMPL_WIN_IOCP_IO_SERVICE_IPP
14 #if defined(_MSC_VER) && (_MSC_VER >= 1200)
16 #endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
18 #include <boost/asio/detail/config.hpp>
20 #if defined(BOOST_ASIO_HAS_IOCP)
22 #include <boost/asio/error.hpp>
23 #include <boost/asio/io_service.hpp>
24 #include <boost/asio/detail/cstdint.hpp>
25 #include <boost/asio/detail/handler_alloc_helpers.hpp>
26 #include <boost/asio/detail/handler_invoke_helpers.hpp>
27 #include <boost/asio/detail/limits.hpp>
28 #include <boost/asio/detail/throw_error.hpp>
29 #include <boost/asio/detail/win_iocp_io_service.hpp>
31 #include <boost/asio/detail/push_options.hpp>
37 struct win_iocp_io_service::work_finished_on_block_exit
39 ~work_finished_on_block_exit()
41 io_service_->work_finished();
44 win_iocp_io_service* io_service_;
47 struct win_iocp_io_service::timer_thread_function
51 while (::InterlockedExchangeAdd(&io_service_->shutdown_, 0) == 0)
53 if (::WaitForSingleObject(io_service_->waitable_timer_.handle,
54 INFINITE) == WAIT_OBJECT_0)
56 ::InterlockedExchange(&io_service_->dispatch_required_, 1);
57 ::PostQueuedCompletionStatus(io_service_->iocp_.handle,
58 0, wake_for_dispatch, 0);
63 win_iocp_io_service* io_service_;
66 win_iocp_io_service::win_iocp_io_service(
67 boost::asio::io_service& io_service, size_t concurrency_hint)
68 : boost::asio::detail::service_base<win_iocp_io_service>(io_service),
72 stop_event_posted_(0),
74 gqcs_timeout_(get_gqcs_timeout()),
77 BOOST_ASIO_HANDLER_TRACKING_INIT;
79 iocp_.handle = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0,
80 static_cast<DWORD>(concurrency_hint < DWORD(~0)
81 ? concurrency_hint : DWORD(~0)));
84 DWORD last_error = ::GetLastError();
85 boost::system::error_code ec(last_error,
86 boost::asio::error::get_system_category());
87 boost::asio::detail::throw_error(ec, "iocp");
91 void win_iocp_io_service::shutdown_service()
93 ::InterlockedExchange(&shutdown_, 1);
95 if (timer_thread_.get())
97 LARGE_INTEGER timeout;
99 ::SetWaitableTimer(waitable_timer_.handle, &timeout, 1, 0, 0, FALSE);
102 while (::InterlockedExchangeAdd(&outstanding_work_, 0) > 0)
104 op_queue<win_iocp_operation> ops;
105 timer_queues_.get_all_timers(ops);
106 ops.push(completed_ops_);
109 while (win_iocp_operation* op = ops.front())
112 ::InterlockedDecrement(&outstanding_work_);
118 DWORD bytes_transferred = 0;
119 dword_ptr_t completion_key = 0;
120 LPOVERLAPPED overlapped = 0;
121 ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
122 &completion_key, &overlapped, gqcs_timeout_);
125 ::InterlockedDecrement(&outstanding_work_);
126 static_cast<win_iocp_operation*>(overlapped)->destroy();
131 if (timer_thread_.get())
132 timer_thread_->join();
135 boost::system::error_code win_iocp_io_service::register_handle(
136 HANDLE handle, boost::system::error_code& ec)
138 if (::CreateIoCompletionPort(handle, iocp_.handle, 0, 0) == 0)
140 DWORD last_error = ::GetLastError();
141 ec = boost::system::error_code(last_error,
142 boost::asio::error::get_system_category());
146 ec = boost::system::error_code();
151 size_t win_iocp_io_service::run(boost::system::error_code& ec)
153 if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
156 ec = boost::system::error_code();
160 win_iocp_thread_info this_thread;
161 thread_call_stack::context ctx(this, this_thread);
164 while (do_one(true, ec))
165 if (n != (std::numeric_limits<size_t>::max)())
170 size_t win_iocp_io_service::run_one(boost::system::error_code& ec)
172 if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
175 ec = boost::system::error_code();
179 win_iocp_thread_info this_thread;
180 thread_call_stack::context ctx(this, this_thread);
182 return do_one(true, ec);
185 size_t win_iocp_io_service::poll(boost::system::error_code& ec)
187 if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
190 ec = boost::system::error_code();
194 win_iocp_thread_info this_thread;
195 thread_call_stack::context ctx(this, this_thread);
198 while (do_one(false, ec))
199 if (n != (std::numeric_limits<size_t>::max)())
204 size_t win_iocp_io_service::poll_one(boost::system::error_code& ec)
206 if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
209 ec = boost::system::error_code();
213 win_iocp_thread_info this_thread;
214 thread_call_stack::context ctx(this, this_thread);
216 return do_one(false, ec);
219 void win_iocp_io_service::stop()
221 if (::InterlockedExchange(&stopped_, 1) == 0)
223 if (::InterlockedExchange(&stop_event_posted_, 1) == 0)
225 if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0))
227 DWORD last_error = ::GetLastError();
228 boost::system::error_code ec(last_error,
229 boost::asio::error::get_system_category());
230 boost::asio::detail::throw_error(ec, "pqcs");
236 void win_iocp_io_service::post_deferred_completion(win_iocp_operation* op)
238 // Flag the operation as ready.
241 // Enqueue the operation on the I/O completion port.
242 if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, op))
244 // Out of resources. Put on completed queue instead.
245 mutex::scoped_lock lock(dispatch_mutex_);
246 completed_ops_.push(op);
247 ::InterlockedExchange(&dispatch_required_, 1);
251 void win_iocp_io_service::post_deferred_completions(
252 op_queue<win_iocp_operation>& ops)
254 while (win_iocp_operation* op = ops.front())
258 // Flag the operation as ready.
261 // Enqueue the operation on the I/O completion port.
262 if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, op))
264 // Out of resources. Put on completed queue instead.
265 mutex::scoped_lock lock(dispatch_mutex_);
266 completed_ops_.push(op);
267 completed_ops_.push(ops);
268 ::InterlockedExchange(&dispatch_required_, 1);
273 void win_iocp_io_service::abandon_operations(
274 op_queue<win_iocp_operation>& ops)
276 while (win_iocp_operation* op = ops.front())
279 ::InterlockedDecrement(&outstanding_work_);
284 void win_iocp_io_service::on_pending(win_iocp_operation* op)
286 if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1)
288 // Enqueue the operation on the I/O completion port.
289 if (!::PostQueuedCompletionStatus(iocp_.handle,
290 0, overlapped_contains_result, op))
292 // Out of resources. Put on completed queue instead.
293 mutex::scoped_lock lock(dispatch_mutex_);
294 completed_ops_.push(op);
295 ::InterlockedExchange(&dispatch_required_, 1);
300 void win_iocp_io_service::on_completion(win_iocp_operation* op,
301 DWORD last_error, DWORD bytes_transferred)
303 // Flag that the operation is ready for invocation.
306 // Store results in the OVERLAPPED structure.
307 op->Internal = reinterpret_cast<ulong_ptr_t>(
308 &boost::asio::error::get_system_category());
309 op->Offset = last_error;
310 op->OffsetHigh = bytes_transferred;
312 // Enqueue the operation on the I/O completion port.
313 if (!::PostQueuedCompletionStatus(iocp_.handle,
314 0, overlapped_contains_result, op))
316 // Out of resources. Put on completed queue instead.
317 mutex::scoped_lock lock(dispatch_mutex_);
318 completed_ops_.push(op);
319 ::InterlockedExchange(&dispatch_required_, 1);
323 void win_iocp_io_service::on_completion(win_iocp_operation* op,
324 const boost::system::error_code& ec, DWORD bytes_transferred)
326 // Flag that the operation is ready for invocation.
329 // Store results in the OVERLAPPED structure.
330 op->Internal = reinterpret_cast<ulong_ptr_t>(&ec.category());
331 op->Offset = ec.value();
332 op->OffsetHigh = bytes_transferred;
334 // Enqueue the operation on the I/O completion port.
335 if (!::PostQueuedCompletionStatus(iocp_.handle,
336 0, overlapped_contains_result, op))
338 // Out of resources. Put on completed queue instead.
339 mutex::scoped_lock lock(dispatch_mutex_);
340 completed_ops_.push(op);
341 ::InterlockedExchange(&dispatch_required_, 1);
345 size_t win_iocp_io_service::do_one(bool block, boost::system::error_code& ec)
349 // Try to acquire responsibility for dispatching timers and completed ops.
350 if (::InterlockedCompareExchange(&dispatch_required_, 0, 1) == 1)
352 mutex::scoped_lock lock(dispatch_mutex_);
354 // Dispatch pending timers and operations.
355 op_queue<win_iocp_operation> ops;
356 ops.push(completed_ops_);
357 timer_queues_.get_ready_timers(ops);
358 post_deferred_completions(ops);
362 // Get the next operation from the queue.
363 DWORD bytes_transferred = 0;
364 dword_ptr_t completion_key = 0;
365 LPOVERLAPPED overlapped = 0;
367 BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
368 &completion_key, &overlapped, block ? gqcs_timeout_ : 0);
369 DWORD last_error = ::GetLastError();
373 win_iocp_operation* op = static_cast<win_iocp_operation*>(overlapped);
374 boost::system::error_code result_ec(last_error,
375 boost::asio::error::get_system_category());
377 // We may have been passed the last_error and bytes_transferred in the
378 // OVERLAPPED structure itself.
379 if (completion_key == overlapped_contains_result)
381 result_ec = boost::system::error_code(static_cast<int>(op->Offset),
382 *reinterpret_cast<boost::system::error_category*>(op->Internal));
383 bytes_transferred = op->OffsetHigh;
386 // Otherwise ensure any result has been saved into the OVERLAPPED
390 op->Internal = reinterpret_cast<ulong_ptr_t>(&result_ec.category());
391 op->Offset = result_ec.value();
392 op->OffsetHigh = bytes_transferred;
395 // Dispatch the operation only if ready. The operation may not be ready
396 // if the initiating function (e.g. a call to WSARecv) has not yet
397 // returned. This is because the initiating function still wants access
398 // to the operation's OVERLAPPED structure.
399 if (::InterlockedCompareExchange(&op->ready_, 1, 0) == 1)
401 // Ensure the count of outstanding work is decremented on block exit.
402 work_finished_on_block_exit on_exit = { this };
405 op->complete(*this, result_ec, bytes_transferred);
406 ec = boost::system::error_code();
412 if (last_error != WAIT_TIMEOUT)
414 ec = boost::system::error_code(last_error,
415 boost::asio::error::get_system_category());
419 // If we're not polling we need to keep going until we get a real handler.
423 ec = boost::system::error_code();
426 else if (completion_key == wake_for_dispatch)
428 // We have been woken up to try to acquire responsibility for dispatching
429 // timers and completed operations.
433 // Indicate that there is no longer an in-flight stop event.
434 ::InterlockedExchange(&stop_event_posted_, 0);
436 // The stopped_ flag is always checked to ensure that any leftover
437 // stop events from a previous run invocation are ignored.
438 if (::InterlockedExchangeAdd(&stopped_, 0) != 0)
440 // Wake up next thread that is blocked on GetQueuedCompletionStatus.
441 if (::InterlockedExchange(&stop_event_posted_, 1) == 0)
443 if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0))
445 last_error = ::GetLastError();
446 ec = boost::system::error_code(last_error,
447 boost::asio::error::get_system_category());
452 ec = boost::system::error_code();
459 DWORD win_iocp_io_service::get_gqcs_timeout()
461 OSVERSIONINFOEX osvi;
462 ZeroMemory(&osvi, sizeof(osvi));
463 osvi.dwOSVersionInfoSize = sizeof(osvi);
464 osvi.dwMajorVersion = 6ul;
466 const uint64_t condition_mask = ::VerSetConditionMask(
467 0, VER_MAJORVERSION, VER_GREATER_EQUAL);
469 if (!!::VerifyVersionInfo(&osvi, VER_MAJORVERSION, condition_mask))
472 return default_gqcs_timeout;
475 void win_iocp_io_service::do_add_timer_queue(timer_queue_base& queue)
477 mutex::scoped_lock lock(dispatch_mutex_);
479 timer_queues_.insert(&queue);
481 if (!waitable_timer_.handle)
483 waitable_timer_.handle = ::CreateWaitableTimer(0, FALSE, 0);
484 if (waitable_timer_.handle == 0)
486 DWORD last_error = ::GetLastError();
487 boost::system::error_code ec(last_error,
488 boost::asio::error::get_system_category());
489 boost::asio::detail::throw_error(ec, "timer");
492 LARGE_INTEGER timeout;
493 timeout.QuadPart = -max_timeout_usec;
494 timeout.QuadPart *= 10;
495 ::SetWaitableTimer(waitable_timer_.handle,
496 &timeout, max_timeout_msec, 0, 0, FALSE);
499 if (!timer_thread_.get())
501 timer_thread_function thread_function = { this };
502 timer_thread_.reset(new thread(thread_function, 65536));
506 void win_iocp_io_service::do_remove_timer_queue(timer_queue_base& queue)
508 mutex::scoped_lock lock(dispatch_mutex_);
510 timer_queues_.erase(&queue);
513 void win_iocp_io_service::update_timeout()
515 if (timer_thread_.get())
517 // There's no point updating the waitable timer if the new timeout period
518 // exceeds the maximum timeout. In that case, we might as well wait for the
519 // existing period of the timer to expire.
520 long timeout_usec = timer_queues_.wait_duration_usec(max_timeout_usec);
521 if (timeout_usec < max_timeout_usec)
523 LARGE_INTEGER timeout;
524 timeout.QuadPart = -timeout_usec;
525 timeout.QuadPart *= 10;
526 ::SetWaitableTimer(waitable_timer_.handle,
527 &timeout, max_timeout_msec, 0, 0, FALSE);
532 } // namespace detail
536 #include <boost/asio/detail/pop_options.hpp>
538 #endif // defined(BOOST_ASIO_HAS_IOCP)
540 #endif // BOOST_ASIO_DETAIL_IMPL_WIN_IOCP_IO_SERVICE_IPP