2
// win_iocp_io_service.hpp
3
// ~~~~~~~~~~~~~~~~~~~~~~~
5
// Copyright (c) 2003-2008 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 ASIO_DETAIL_WIN_IOCP_IO_SERVICE_HPP
12
#define ASIO_DETAIL_WIN_IOCP_IO_SERVICE_HPP
14
#if defined(_MSC_VER) && (_MSC_VER >= 1200)
16
#endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
18
#include "asio/detail/push_options.hpp"
20
#include "asio/detail/win_iocp_io_service_fwd.hpp"
22
#if defined(ASIO_HAS_IOCP)
24
#include "asio/detail/push_options.hpp"
26
#include <boost/throw_exception.hpp>
27
#include "asio/detail/pop_options.hpp"
29
#include "asio/io_service.hpp"
30
#include "asio/system_error.hpp"
31
#include "asio/detail/call_stack.hpp"
32
#include "asio/detail/handler_alloc_helpers.hpp"
33
#include "asio/detail/handler_invoke_helpers.hpp"
34
#include "asio/detail/service_base.hpp"
35
#include "asio/detail/socket_types.hpp"
36
#include "asio/detail/timer_queue.hpp"
37
#include "asio/detail/mutex.hpp"
42
class win_iocp_io_service
43
: public asio::detail::service_base<win_iocp_io_service>
46
// Base class for all operations. A function pointer is used instead of
47
// virtual functions to avoid the associated overhead.
49
// This class inherits from OVERLAPPED so that we can downcast to get back to
50
// the operation pointer from the LPOVERLAPPED out parameter of
51
// GetQueuedCompletionStatus.
56
typedef void (*invoke_func_type)(operation*, DWORD, size_t);
57
typedef void (*destroy_func_type)(operation*);
59
operation(win_iocp_io_service& iocp_service,
60
invoke_func_type invoke_func, destroy_func_type destroy_func)
61
: outstanding_operations_(&iocp_service.outstanding_operations_),
62
invoke_func_(invoke_func),
63
destroy_func_(destroy_func)
71
::InterlockedIncrement(outstanding_operations_);
74
void do_completion(DWORD last_error, size_t bytes_transferred)
76
invoke_func_(this, last_error, bytes_transferred);
85
// Prevent deletion through this type.
88
::InterlockedDecrement(outstanding_operations_);
92
long* outstanding_operations_;
93
invoke_func_type invoke_func_;
94
destroy_func_type destroy_func_;
99
win_iocp_io_service(asio::io_service& io_service)
100
: asio::detail::service_base<win_iocp_io_service>(io_service),
102
outstanding_work_(0),
103
outstanding_operations_(0),
107
timer_interrupt_issued_(false)
111
void init(size_t concurrency_hint)
113
iocp_.handle = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0,
114
static_cast<DWORD>((std::min<size_t>)(concurrency_hint, DWORD(~0))));
117
DWORD last_error = ::GetLastError();
118
asio::system_error e(
119
asio::error_code(last_error,
120
asio::error::get_system_category()),
122
boost::throw_exception(e);
126
// Destroy all user-defined handler objects owned by the service.
127
void shutdown_service()
129
::InterlockedExchange(&shutdown_, 1);
131
while (::InterlockedExchangeAdd(&outstanding_operations_, 0) > 0)
133
DWORD bytes_transferred = 0;
134
#if (WINVER < 0x0500)
135
DWORD completion_key = 0;
137
DWORD_PTR completion_key = 0;
139
LPOVERLAPPED overlapped = 0;
140
::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
141
&completion_key, &overlapped, INFINITE);
143
static_cast<operation*>(overlapped)->destroy();
146
for (std::size_t i = 0; i < timer_queues_.size(); ++i)
147
timer_queues_[i]->destroy_timers();
148
timer_queues_.clear();
151
// Register a handle with the IO completion port.
152
asio::error_code register_handle(
153
HANDLE handle, asio::error_code& ec)
155
if (::CreateIoCompletionPort(handle, iocp_.handle, 0, 0) == 0)
157
DWORD last_error = ::GetLastError();
158
ec = asio::error_code(last_error,
159
asio::error::get_system_category());
163
ec = asio::error_code();
168
// Run the event loop until stopped or no more work.
169
size_t run(asio::error_code& ec)
171
if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
173
ec = asio::error_code();
177
call_stack<win_iocp_io_service>::context ctx(this);
180
while (do_one(true, ec))
181
if (n != (std::numeric_limits<size_t>::max)())
186
// Run until stopped or one operation is performed.
187
size_t run_one(asio::error_code& ec)
189
if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
191
ec = asio::error_code();
195
call_stack<win_iocp_io_service>::context ctx(this);
197
return do_one(true, ec);
200
// Poll for operations without blocking.
201
size_t poll(asio::error_code& ec)
203
if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
205
ec = asio::error_code();
209
call_stack<win_iocp_io_service>::context ctx(this);
212
while (do_one(false, ec))
213
if (n != (std::numeric_limits<size_t>::max)())
218
// Poll for one operation without blocking.
219
size_t poll_one(asio::error_code& ec)
221
if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
223
ec = asio::error_code();
227
call_stack<win_iocp_io_service>::context ctx(this);
229
return do_one(false, ec);
232
// Stop the event processing loop.
235
if (::InterlockedExchange(&stopped_, 1) == 0)
237
if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0))
239
DWORD last_error = ::GetLastError();
240
asio::system_error e(
241
asio::error_code(last_error,
242
asio::error::get_system_category()),
244
boost::throw_exception(e);
249
// Reset in preparation for a subsequent run invocation.
252
::InterlockedExchange(&stopped_, 0);
255
// Notify that some work has started.
258
::InterlockedIncrement(&outstanding_work_);
261
// Notify that some work has finished.
264
if (::InterlockedDecrement(&outstanding_work_) == 0)
268
// Request invocation of the given handler.
269
template <typename Handler>
270
void dispatch(Handler handler)
272
if (call_stack<win_iocp_io_service>::contains(this))
273
asio_handler_invoke_helpers::invoke(handler, &handler);
278
// Request invocation of the given handler and return immediately.
279
template <typename Handler>
280
void post(Handler handler)
282
// If the service has been shut down we silently discard the handler.
283
if (::InterlockedExchangeAdd(&shutdown_, 0) != 0)
286
// Allocate and construct an operation to wrap the handler.
287
typedef handler_operation<Handler> value_type;
288
typedef handler_alloc_traits<Handler, value_type> alloc_traits;
289
raw_handler_ptr<alloc_traits> raw_ptr(handler);
290
handler_ptr<alloc_traits> ptr(raw_ptr, *this, handler);
292
// Enqueue the operation on the I/O completion port.
293
if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, ptr.get()))
295
DWORD last_error = ::GetLastError();
296
asio::system_error e(
297
asio::error_code(last_error,
298
asio::error::get_system_category()),
300
boost::throw_exception(e);
303
// Operation has been successfully posted.
307
// Request invocation of the given OVERLAPPED-derived operation.
308
void post_completion(operation* op, DWORD op_last_error,
309
DWORD bytes_transferred)
311
// Enqueue the operation on the I/O completion port.
312
if (!::PostQueuedCompletionStatus(iocp_.handle,
313
bytes_transferred, op_last_error, op))
315
DWORD last_error = ::GetLastError();
316
asio::system_error e(
317
asio::error_code(last_error,
318
asio::error::get_system_category()),
320
boost::throw_exception(e);
324
// Add a new timer queue to the service.
325
template <typename Time_Traits>
326
void add_timer_queue(timer_queue<Time_Traits>& timer_queue)
328
asio::detail::mutex::scoped_lock lock(timer_mutex_);
329
timer_queues_.push_back(&timer_queue);
332
// Remove a timer queue from the service.
333
template <typename Time_Traits>
334
void remove_timer_queue(timer_queue<Time_Traits>& timer_queue)
336
asio::detail::mutex::scoped_lock lock(timer_mutex_);
337
for (std::size_t i = 0; i < timer_queues_.size(); ++i)
339
if (timer_queues_[i] == &timer_queue)
341
timer_queues_.erase(timer_queues_.begin() + i);
347
// Schedule a timer in the given timer queue to expire at the specified
348
// absolute time. The handler object will be invoked when the timer expires.
349
template <typename Time_Traits, typename Handler>
350
void schedule_timer(timer_queue<Time_Traits>& timer_queue,
351
const typename Time_Traits::time_type& time, Handler handler, void* token)
353
// If the service has been shut down we silently discard the timer.
354
if (::InterlockedExchangeAdd(&shutdown_, 0) != 0)
357
asio::detail::mutex::scoped_lock lock(timer_mutex_);
358
if (timer_queue.enqueue_timer(time, handler, token))
360
if (!timer_interrupt_issued_)
362
timer_interrupt_issued_ = true;
364
::PostQueuedCompletionStatus(iocp_.handle,
365
0, steal_timer_dispatching, 0);
370
// Cancel the timer associated with the given token. Returns the number of
371
// handlers that have been posted or dispatched.
372
template <typename Time_Traits>
373
std::size_t cancel_timer(timer_queue<Time_Traits>& timer_queue, void* token)
375
// If the service has been shut down we silently ignore the cancellation.
376
if (::InterlockedExchangeAdd(&shutdown_, 0) != 0)
379
asio::detail::mutex::scoped_lock lock(timer_mutex_);
380
std::size_t n = timer_queue.cancel_timer(token);
381
if (n > 0 && !timer_interrupt_issued_)
383
timer_interrupt_issued_ = true;
385
::PostQueuedCompletionStatus(iocp_.handle,
386
0, steal_timer_dispatching, 0);
392
// Dequeues at most one operation from the I/O completion port, and then
393
// executes it. Returns the number of operations that were dequeued (i.e.
395
size_t do_one(bool block, asio::error_code& ec)
397
long this_thread_id = static_cast<long>(::GetCurrentThreadId());
401
// Try to acquire responsibility for dispatching timers.
402
bool dispatching_timers = (::InterlockedCompareExchange(
403
&timer_thread_, this_thread_id, 0) == 0);
405
// Calculate timeout for GetQueuedCompletionStatus call.
406
DWORD timeout = max_timeout;
407
if (dispatching_timers)
409
asio::detail::mutex::scoped_lock lock(timer_mutex_);
410
timer_interrupt_issued_ = false;
411
timeout = get_timeout();
414
// Get the next operation from the queue.
415
DWORD bytes_transferred = 0;
416
#if (WINVER < 0x0500)
417
DWORD completion_key = 0;
419
DWORD_PTR completion_key = 0;
421
LPOVERLAPPED overlapped = 0;
423
BOOL ok = ::GetQueuedCompletionStatus(iocp_.handle, &bytes_transferred,
424
&completion_key, &overlapped, block ? timeout : 0);
425
DWORD last_error = ::GetLastError();
427
// Dispatch any pending timers.
428
if (dispatching_timers)
432
asio::detail::mutex::scoped_lock lock(timer_mutex_);
433
timer_queues_copy_ = timer_queues_;
434
for (std::size_t i = 0; i < timer_queues_copy_.size(); ++i)
436
timer_queues_copy_[i]->dispatch_timers();
437
timer_queues_copy_[i]->dispatch_cancellations();
438
timer_queues_copy_[i]->complete_timers();
443
// Transfer responsibility for dispatching timers to another thread.
444
if (::InterlockedCompareExchange(&timer_thread_,
445
0, this_thread_id) == this_thread_id)
447
::PostQueuedCompletionStatus(iocp_.handle,
448
0, transfer_timer_dispatching, 0);
455
if (!ok && overlapped == 0)
457
if (block && last_error == WAIT_TIMEOUT)
459
// Relinquish responsibility for dispatching timers.
460
if (dispatching_timers)
462
::InterlockedCompareExchange(&timer_thread_, 0, this_thread_id);
468
// Transfer responsibility for dispatching timers to another thread.
469
if (dispatching_timers && ::InterlockedCompareExchange(
470
&timer_thread_, 0, this_thread_id) == this_thread_id)
472
::PostQueuedCompletionStatus(iocp_.handle,
473
0, transfer_timer_dispatching, 0);
476
ec = asio::error_code();
481
// We may have been passed a last_error value in the completion_key.
484
last_error = completion_key;
487
// Transfer responsibility for dispatching timers to another thread.
488
if (dispatching_timers && ::InterlockedCompareExchange(
489
&timer_thread_, 0, this_thread_id) == this_thread_id)
491
::PostQueuedCompletionStatus(iocp_.handle,
492
0, transfer_timer_dispatching, 0);
495
// Ensure that the io_service does not exit due to running out of work
496
// while we make the upcall.
497
auto_work work(*this);
499
// Dispatch the operation.
500
operation* op = static_cast<operation*>(overlapped);
501
op->do_completion(last_error, bytes_transferred);
503
ec = asio::error_code();
506
else if (completion_key == transfer_timer_dispatching)
508
// Woken up to try to acquire responsibility for dispatching timers.
509
::InterlockedCompareExchange(&timer_thread_, 0, this_thread_id);
511
else if (completion_key == steal_timer_dispatching)
513
// Woken up to steal responsibility for dispatching timers.
514
::InterlockedExchange(&timer_thread_, 0);
518
// Relinquish responsibility for dispatching timers. If the io_service
519
// is not being stopped then the thread will get an opportunity to
520
// reacquire timer responsibility on the next loop iteration.
521
if (dispatching_timers)
523
::InterlockedCompareExchange(&timer_thread_, 0, this_thread_id);
526
// The stopped_ flag is always checked to ensure that any leftover
527
// interrupts from a previous run invocation are ignored.
528
if (::InterlockedExchangeAdd(&stopped_, 0) != 0)
530
// Wake up next thread that is blocked on GetQueuedCompletionStatus.
531
if (!::PostQueuedCompletionStatus(iocp_.handle, 0, 0, 0))
533
DWORD last_error = ::GetLastError();
534
ec = asio::error_code(last_error,
535
asio::error::get_system_category());
539
ec = asio::error_code();
546
// Check if all timer queues are empty.
547
bool all_timer_queues_are_empty() const
549
for (std::size_t i = 0; i < timer_queues_.size(); ++i)
550
if (!timer_queues_[i]->empty())
555
// Get the timeout value for the GetQueuedCompletionStatus call. The timeout
556
// value is returned as a number of milliseconds. We will wait no longer than
557
// 1000 milliseconds.
560
if (all_timer_queues_are_empty())
563
boost::posix_time::time_duration minimum_wait_duration
564
= boost::posix_time::milliseconds(max_timeout);
566
for (std::size_t i = 0; i < timer_queues_.size(); ++i)
568
boost::posix_time::time_duration wait_duration
569
= timer_queues_[i]->wait_duration();
570
if (wait_duration < minimum_wait_duration)
571
minimum_wait_duration = wait_duration;
574
if (minimum_wait_duration > boost::posix_time::time_duration())
576
int milliseconds = minimum_wait_duration.total_milliseconds();
577
return static_cast<DWORD>(milliseconds > 0 ? milliseconds : 1);
587
auto_work(win_iocp_io_service& io_service)
588
: io_service_(io_service)
590
io_service_.work_started();
595
io_service_.work_finished();
599
win_iocp_io_service& io_service_;
602
template <typename Handler>
603
struct handler_operation
606
handler_operation(win_iocp_io_service& io_service,
608
: operation(io_service, &handler_operation<Handler>::do_completion_impl,
609
&handler_operation<Handler>::destroy_impl),
610
io_service_(io_service),
613
io_service_.work_started();
618
io_service_.work_finished();
622
// Prevent copying and assignment.
623
handler_operation(const handler_operation&);
624
void operator=(const handler_operation&);
626
static void do_completion_impl(operation* op, DWORD, size_t)
628
// Take ownership of the operation object.
629
typedef handler_operation<Handler> op_type;
630
op_type* handler_op(static_cast<op_type*>(op));
631
typedef handler_alloc_traits<Handler, op_type> alloc_traits;
632
handler_ptr<alloc_traits> ptr(handler_op->handler_, handler_op);
634
// Make a copy of the handler so that the memory can be deallocated before
635
// the upcall is made.
636
Handler handler(handler_op->handler_);
638
// Free the memory associated with the handler.
642
asio_handler_invoke_helpers::invoke(handler, &handler);
645
static void destroy_impl(operation* op)
647
// Take ownership of the operation object.
648
typedef handler_operation<Handler> op_type;
649
op_type* handler_op(static_cast<op_type*>(op));
650
typedef handler_alloc_traits<Handler, op_type> alloc_traits;
651
handler_ptr<alloc_traits> ptr(handler_op->handler_, handler_op);
653
// A sub-object of the handler may be the true owner of the memory
654
// associated with the handler. Consequently, a local copy of the handler
655
// is required to ensure that any owning sub-object remains valid until
656
// after we have deallocated the memory here.
657
Handler handler(handler_op->handler_);
660
// Free the memory associated with the handler.
664
win_iocp_io_service& io_service_;
668
// The IO completion port used for queueing operations.
672
iocp_holder() : handle(0) {}
673
~iocp_holder() { if (handle) ::CloseHandle(handle); }
676
// The count of unfinished work.
677
long outstanding_work_;
679
// The count of unfinished operations.
680
long outstanding_operations_;
681
friend class operation;
683
// Flag to indicate whether the event loop has been stopped.
686
// Flag to indicate whether the service has been shut down.
691
// Maximum GetQueuedCompletionStatus timeout, in milliseconds.
694
// Completion key value to indicate that responsibility for dispatching
695
// timers is being cooperatively transferred from one thread to another.
696
transfer_timer_dispatching = 1,
698
// Completion key value to indicate that responsibility for dispatching
699
// timers should be stolen from another thread.
700
steal_timer_dispatching = 2
703
// The thread that's currently in charge of dispatching timers.
706
// Mutex for protecting access to the timer queues.
709
// Whether a thread has been interrupted to process a new timeout.
710
bool timer_interrupt_issued_;
713
std::vector<timer_queue_base*> timer_queues_;
715
// A copy of the timer queues, used when dispatching, cancelling and cleaning
716
// up timers. The copy is stored as a class data member to avoid unnecessary
717
// memory allocation.
718
std::vector<timer_queue_base*> timer_queues_copy_;
721
} // namespace detail
724
#endif // defined(ASIO_HAS_IOCP)
726
#include "asio/detail/pop_options.hpp"
728
#endif // ASIO_DETAIL_WIN_IOCP_IO_SERVICE_HPP
added
removed
portable/libtorrent/include/libtorrent/asio/detail/win_iocp_io_service.hpp
