include/boost/corosio/native/detail/iocp/win_wait_reactor.hpp

94.2% Lines (196/208) 100.0% List of functions (16/16) 80.5% Branches (132/164)

[ ] prev/next file n p prev/next uncovered line

win_wait_reactor.hpp

f(x) Functions (16)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Michael Vandeberg
3			//
4			// Distributed under the Boost Software License, Version 1.0. (See accompanying
5			// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6			//
7			// Official repository: https://github.com/cppalliance/corosio
8			//
9
10			#ifndef BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_WAIT_REACTOR_HPP
11			#define BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_WAIT_REACTOR_HPP
12
13			#include <boost/corosio/detail/platform.hpp>
14
15			#if BOOST_COROSIO_HAS_IOCP
16
17			// This header is included from the bottom of win_scheduler.hpp after
18			// the scheduler class is fully defined. Including it directly would
19			// circle back into a still-incomplete win_scheduler when the dtor's
20			// unique_ptr<win_wait_reactor>::reset() is parsed. Diagnose that
21			// rather than emitting a confusing "incomplete type" error far away.
22			#ifndef BOOST_COROSIO_DETAIL_IOCP_WIN_SCHEDULER_BODY_DONE
23			#error "Include <boost/corosio/native/detail/iocp/win_scheduler.hpp> \
24			instead of including this header directly."
25			#endif
26
27			#include <boost/corosio/wait_type.hpp>
28			#include <boost/corosio/native/detail/iocp/win_overlapped_op.hpp>
29			#include <boost/corosio/native/detail/iocp/win_scheduler.hpp>
30			#include <boost/corosio/native/detail/iocp/win_windows.hpp>
31			#include <boost/corosio/native/detail/iocp/win_wsa_init.hpp>
32
33			#include <Ws2tcpip.h>
34
35			#include <algorithm>
36			#include <atomic>
37			#include <cstddef>
38			#include <mutex>
39			#include <thread>
40			#include <vector>
41
42			namespace boost::corosio::detail {
43
44			/** Auxiliary select-based reactor for IOCP wait operations.
45
46			IOCP has no native primitive for socket readiness without I/O.
47			For cases where a zero-byte WSARecv won't work (datagram-read,
48			acceptor-read, write-wait, error-wait), this reactor runs a
49			dedicated thread using WSAPoll to detect readiness and posts a
50			synthetic completion to the owning IOCP scheduler via
51			win_scheduler::on_completion().
52
53			The same dispatch path used by overlapped I/O then delivers the
54			completion to the user's coroutine, so the public API is uniform
55			across backends.
56
57			Per-op lifecycle:
58			1. Caller sets up an overlapped_op (h, ex, ec_out, cancelled flag).
59			2. Caller calls register_wait(fd, w, op) and returns
60			std::noop_coroutine. The op is parked in the reactor's table.
61			3. Reactor thread polls. When the fd is ready, the op is removed
62			from the table and posted to the scheduler. The error code
63			delivered to the completion is: ec={} on success; the SO_ERROR
64			value if error revents fired and SO_ERROR is set; or
65			WSAECONNABORTED as a synthesized fallback for wait_type::error
66			when error revents fired but SO_ERROR returned zero.
67			4. On socket cancel(), the user's thread calls cancel_wait(op),
68			which queues a cancel request. The reactor thread removes the
69			op from the table and posts a completion; invoke_handler sees
70			op.cancelled==true and yields capy::cond::canceled.
71
72			A poll the provider refuses on account of one descriptor answers
73			that descriptor's op with the refusal and leaves the rest of the
74			table polling. Such an error reaches the caller only where nothing
75			flagged the op first: close() and cancel() set the cancelled flag
76			before the handle goes, and a flagged op yields canceled whatever
77			the completion carries.
78
79			A refusal none of the entries accounts for is the end of the
80			reactor: the error is latched, the ops it still holds are completed
81			with it, and every later register_wait is completed with it too,
82			since nothing would drain an op parked after that point. Both
83			answers name the reason waits can no longer be served; canceled is
84			reserved for ops close() or cancel() flagged, and for the ops a
85			stop() drain finds parked.
86
87			The constructor builds the wakeup channel and throws if it cannot:
88			a reactor that cannot be woken can never report readiness, so
89			there is no reactor worth handing back. The polling thread is a
90			separate cost, paid by the first register_wait, so a context that
91			never waits never carries one. A thread the system refuses costs
92			only the wait that asked for it: that wait completes with
93			`resource_unavailable_try_again` and the next one tries again.
94
95			Thread-safe: register_wait, cancel_wait, and stop may be called
96			from any thread.
97			*/
98			class win_wait_reactor : private win_wsa_init
99			{
100			public:
101			/** Construct the reactor and its wakeup channel.
102
103			@par Exception Safety
104			Strong guarantee. A channel that cannot be formed leaves no
105			socket open.
106
107			@param sched The scheduler synthetic completions are posted to.
108
109			@throws std::system_error If Winsock could not be started or
110			the wakeup socket pair could not be built.
111			*/
112			explicit win_wait_reactor(win_scheduler& sched);
113
114			/// Stop the reactor thread and close the wakeup channel.
115			~win_wait_reactor();
116
117			win_wait_reactor(win_wait_reactor const&) = delete;
118			win_wait_reactor& operator=(win_wait_reactor const&) = delete;
119
120			/// Park an overlapped_op until @p fd is ready for @p w.
121			void register_wait(SOCKET fd, wait_type w, overlapped_op* op);
122
123			/// Remove a parked op and post a completion. Idempotent.
124			void cancel_wait(overlapped_op* op);
125
126			/// Stop the reactor thread and drain remaining ops as cancelled.
127			void stop();
128
129			private:
130			struct entry
131			{
132			SOCKET fd = INVALID_SOCKET;
133			wait_type w = wait_type::read;
134			overlapped_op* op = nullptr;
135			};
136
137			void run();
138			bool drop_refused_entries();
139			DWORD queue_register(entry const& e);
140			void wake_self() noexcept;
141			DWORD make_wakeup_pair() noexcept;
142			void close_wakeup_pair() noexcept;
143
144			// A failed call that left a zero last error would answer "no
145			// error" and put the reactor straight back on the silent path.
146		137x	static DWORD last_error() noexcept
147			{
148		137x	DWORD const err = ::WSAGetLastError();
149	1/2 ✓ Branch 3 → 4 taken 137 times. ✗ Branch 3 → 5 not taken.	137x	return err != 0 ? err : static_cast<DWORD>(WSAEINVAL);
150			}
151
152		43x	static SHORT events_for_wait(wait_type w) noexcept
153			{
154	3/3 ✓ Branch 2 → 3 taken 21 times. ✓ Branch 2 → 4 taken 13 times. ✓ Branch 2 → 5 taken 9 times.	43x	switch (w)
155			{
156		21x	case wait_type::read:
157		21x	return POLLRDNORM;
158		13x	case wait_type::write:
159		13x	return POLLWRNORM;
160			// The Microsoft provider does not implement POLLPRI and
161			// refuses the whole call when it is asked for, which would
162			// take every other registration in the set with it. The band
163			// it does implement carries the same out-of-band meaning, and
164			// the error conditions an error wait is really after arrive in
165			// revents whether or not they were asked for.
166		9x	default:
167		9x	return POLLRDBAND;
168			}
169			}
170
171		27x	static bool ready_for_wait(wait_type w, SHORT revents) noexcept
172			{
173		27x	constexpr SHORT err_bits = POLLERR \| POLLHUP \| POLLNVAL;
174	3/3 ✓ Branch 2 → 3 taken 11 times. ✓ Branch 2 → 4 taken 13 times. ✓ Branch 2 → 5 taken 3 times.	27x	switch (w)
175			{
176		11x	case wait_type::read:
177		11x	return (revents & (POLLRDNORM \| POLLRDBAND \| err_bits)) != 0;
178		13x	case wait_type::write:
179		13x	return (revents & (POLLWRNORM \| POLLWRBAND \| err_bits)) != 0;
180		3x	default:
181		3x	return (revents & (POLLRDBAND \| err_bits)) != 0;
182			}
183			}
184
185			win_scheduler& sched_;
186
187			// Built by the constructor and closed by the destructor, so every
188			// other member function can assume a usable channel.
189			SOCKET wakeup_read_ = INVALID_SOCKET;
190			SOCKET wakeup_write_ = INVALID_SOCKET;
191
192			// Also guards thread_ against a start racing the stop that joins it.
193			std::mutex mutex_;
194			std::vector<entry> pending_register_;
195			std::vector<overlapped_op*> pending_cancel_;
196			std::atomic<bool> stop_{false};
197			std::atomic<bool> wake_pending_{false};
198
199			// What the exit told the ops it was still holding, kept so a later
200			// register_wait can be told the same thing instead of inventing a
201			// cancellation nobody asked for. Non-zero is also what says the
202			// polling thread left on its own after a WSAPoll error: stop_ must
203			// not be used for that exit -- stop() reads it as "already
204			// stopped" and would skip the join that keeps the thread from
205			// being destroyed joinable. Nothing latches a zero here, because
206			// a failed call that left a zero last error is substituted for.
207			DWORD dead_err_ = 0;
208
209			std::vector<entry> registered_; // reactor-thread-only
210
211			std::thread thread_;
212			};
213
214		1383x	inline win_wait_reactor::win_wait_reactor(win_scheduler& sched) : sched_(sched)
215			{
216			// The win_wsa_init base is what makes the sockets below legal, and
217			// holding the reference rather than borrowing someone else's is
218			// what keeps them legal to the end: a base is constructed before
219			// this body and released after ~win_wait_reactor has closed the
220			// pair, so WSACleanup can never land between the two.
221			//
222			// A reactor that cannot be woken would park every op it is handed
223			// forever, so it refuses to exist rather than being handed out
224			// broken. The polling thread waits for the first register_wait.
225	2/2 ✓ Branch 11 → 12 taken 136 times. ✓ Branch 11 → 14 taken 1247 times.	1383x	if (DWORD const err = make_wakeup_pair(); err != 0)
226		136x	detail::throw_system_error(make_err(err), "win_wait_reactor");
227		2063x	}
228
229		1247x	inline win_wait_reactor::~win_wait_reactor()
230			{
231		1247x	stop();
232		1247x	close_wakeup_pair();
233		1247x	}
234
235			inline DWORD
236		1383x	win_wait_reactor::make_wakeup_pair() noexcept
237			{
238			// Build a pair of connected loopback sockets to use as a wakeup
239			// channel. Winsock has no socketpair(2), so we listen on
240			// 127.0.0.1:0, connect a peer, then accept it.
241			//
242			// Every failure path reads the last error before closing
243			// anything: closesocket() overwrites it.
244		1383x	SOCKET listener = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
245	2/2 ✓ Branch 3 → 4 taken 17 times. ✓ Branch 3 → 5 taken 1366 times.	1383x	if (listener == INVALID_SOCKET)
246		17x	return last_error();
247
248		1366x	sockaddr_in addr{};
249		1366x	addr.sin_family = AF_INET;
250		1366x	addr.sin_addr.s_addr = ::htonl(INADDR_LOOPBACK);
251		1366x	addr.sin_port = 0;
252
253		1366x	int len = sizeof(addr);
254		1366x	if (::bind(listener, reinterpret_cast<sockaddr*>(&addr), len) ==
255		1349x	SOCKET_ERROR \|\|
256	6/6 ✓ Branch 7 → 8 taken 1349 times. ✓ Branch 7 → 12 taken 17 times. ✓ Branch 9 → 10 taken 1332 times. ✓ Branch 9 → 12 taken 17 times. ✓ Branch 14 → 15 taken 51 times. ✓ Branch 14 → 18 taken 1315 times.	2698x	::listen(listener, 1) == SOCKET_ERROR \|\|
257	2/2 ✓ Branch 11 → 12 taken 17 times. ✓ Branch 11 → 13 taken 1315 times.	1332x	::getsockname(listener, reinterpret_cast<sockaddr*>(&addr), &len) ==
258			SOCKET_ERROR)
259			{
260		51x	DWORD const err = last_error();
261		51x	::closesocket(listener);
262		51x	return err;
263			}
264
265		1315x	wakeup_write_ = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
266	2/2 ✓ Branch 19 → 20 taken 17 times. ✓ Branch 19 → 23 taken 1298 times.	1315x	if (wakeup_write_ == INVALID_SOCKET)
267			{
268		17x	DWORD const err = last_error();
269		17x	::closesocket(listener);
270		17x	return err;
271			}
272
273	2/2 ✓ Branch 24 → 25 taken 17 times. ✓ Branch 24 → 29 taken 1281 times.	1298x	if (::connect(wakeup_write_, reinterpret_cast<sockaddr*>(&addr), len) ==
274			SOCKET_ERROR)
275			{
276		17x	DWORD const err = last_error();
277		17x	::closesocket(wakeup_write_);
278		17x	wakeup_write_ = INVALID_SOCKET;
279		17x	::closesocket(listener);
280		17x	return err;
281			}
282
283		1281x	wakeup_read_ = ::accept(listener, nullptr, nullptr);
284	2/2 ✓ Branch 30 → 31 taken 17 times. ✓ Branch 30 → 35 taken 1264 times.	1281x	if (wakeup_read_ == INVALID_SOCKET)
285			{
286		17x	DWORD const err = last_error();
287		17x	::closesocket(listener);
288		17x	::closesocket(wakeup_write_);
289		17x	wakeup_write_ = INVALID_SOCKET;
290		17x	return err;
291			}
292		1264x	::closesocket(listener);
293
294			// The drain loop in run() calls recv() until it returns <= 0.
295			// With a blocking socket that second recv() would block instead
296			// of returning WSAEWOULDBLOCK, deadlocking the reactor thread.
297		1264x	u_long non_blocking = 1;
298	2/2 ✓ Branch 37 → 38 taken 17 times. ✓ Branch 37 → 41 taken 1247 times.	1264x	if (::ioctlsocket(wakeup_read_, FIONBIO, &non_blocking) == SOCKET_ERROR)
299			{
300		17x	DWORD const err = last_error();
301		17x	close_wakeup_pair();
302		17x	return err;
303			}
304		1247x	return 0;
305			}
306
307			inline void
308		1264x	win_wait_reactor::close_wakeup_pair() noexcept
309			{
310	1/2 ✓ Branch 2 → 3 taken 1264 times. ✗ Branch 2 → 5 not taken.	1264x	if (wakeup_read_ != INVALID_SOCKET)
311			{
312		1264x	::closesocket(wakeup_read_);
313		1264x	wakeup_read_ = INVALID_SOCKET;
314			}
315	1/2 ✓ Branch 5 → 6 taken 1264 times. ✗ Branch 5 → 8 not taken.	1264x	if (wakeup_write_ != INVALID_SOCKET)
316			{
317		1264x	::closesocket(wakeup_write_);
318		1264x	wakeup_write_ = INVALID_SOCKET;
319			}
320		1264x	}
321
322			inline void
323		31358x	win_wait_reactor::wake_self() noexcept
324			{
325			// Coalesce wakes: only send a byte if no wake is already pending.
326		31358x	bool expected = false;
327	2/2 ✓ Branch 3 → 4 taken 29919 times. ✓ Branch 3 → 5 taken 1439 times.	31358x	if (!wake_pending_.compare_exchange_strong(
328			expected, true, std::memory_order_acq_rel))
329		29919x	return;
330
331		1439x	char b = 0;
332	2/2 ✓ Branch 6 → 7 taken 2 times. ✓ Branch 6 → 8 taken 1437 times.	1439x	if (::send(wakeup_write_, &b, 1, 0) == SOCKET_ERROR)
333			{
334			// The flag is what coalesces later wakes into a byte already
335			// in the channel; a send that failed put no byte there, so
336			// leaving it latched would swallow every wake that follows.
337			// Disarming keeps the cost to the wakes already in flight: the
338			// next register, cancel or stop sends its own byte and the
339			// reactor learns about both.
340		2x	wake_pending_.store(false, std::memory_order_release);
341			}
342			}
343
344			inline void
345		48x	win_wait_reactor::register_wait(SOCKET fd, wait_type w, overlapped_op* op)
346			{
347			// If the op was already cancelled (e.g. pre-cancelled stop_token
348			// fired synchronously before this call), complete immediately
349			// instead of registering. Otherwise the reactor would park the
350			// op forever because the earlier cancel_wait() found nothing to
351			// cancel in registered_.
352	2/2 ✓ Branch 3 → 4 taken 4 times. ✓ Branch 3 → 6 taken 44 times.	48x	if (op->cancelled.load(std::memory_order_acquire))
353			{
354		4x	sched_.on_completion(op, 0, 0);
355		4x	return;
356			}
357
358	3/3 ✓ Branch 6 → 7 taken 44 times. ✓ Branch 7 → 8 taken 1 time. ✓ Branch 7 → 10 taken 43 times.	44x	if (DWORD const err = queue_register(entry{fd, w, op}); err != 0)
359			{
360			// Nothing would ever drain a parked op, and the refusal knows
361			// why: the abort a stop drains with, the error the polling
362			// thread died of, or the system declining a thread.
363		1x	sched_.on_completion(op, err, 0);
364		1x	return;
365			}
366		43x	wake_self();
367			}
368
369			inline DWORD
370		44x	win_wait_reactor::queue_register(entry const& e)
371			{
372	1/1 ✓ Branch 2 → 3 taken 44 times.	44x	std::lock_guard lock(mutex_);
373			// stop() sets the flag before it takes the thread out from under
374			// this lock, and never joins again. Checking the flag and queueing
375			// in one critical section is what keeps both halves honest: a
376			// thread started after that join would be destroyed still joinable,
377			// which ends the process, and an op queued after it would have no
378			// drainer. Queueing under the flag instead leaves the op for the
379			// drain run() performs on its way out.
380			// dead_err_ says the same thing for the other exit: a thread that
381			// left on a WSAPoll error drained what it held and will not poll
382			// again, so a register queued after it would wait on nobody. It
383			// answers with what killed it, so the caller learns why its waits
384			// stopped being served instead of hearing that something cancelled
385			// them.
386	1/2 ✗ Branch 4 → 5 not taken. ✓ Branch 4 → 6 taken 44 times.	44x	if (stop_.load(std::memory_order_acquire))
387		✗	return ERROR_OPERATION_ABORTED;
388	2/2 ✓ Branch 6 → 7 taken 1 time. ✓ Branch 6 → 8 taken 43 times.	44x	if (dead_err_)
389		1x	return dead_err_;
390
391			// A polling thread costs a thread per context, and a context that
392			// never waits never pays for one; the first wait is what starts it.
393			// A system that will not give one refuses this wait alone: nothing
394			// has been queued yet, so the reactor is left exactly as it was and
395			// the next register_wait asks again.
396	2/2 ✓ Branch 9 → 10 taken 36 times. ✓ Branch 9 → 14 taken 7 times.	43x	if (!thread_.joinable())
397			{
398			try
399			{
400	1/1 ✓ Branch 10 → 11 taken 36 times.	72x	thread_ = std::thread([this] { run(); });
401			}
402		✗	catch (...)
403			{
404		✗	return ERROR_MAX_THRDS_REACHED;
405		✗	}
406			}
407
408	1/1 ✓ Branch 14 → 15 taken 43 times.	43x	pending_register_.push_back(e);
409		43x	return 0;
410		44x	}
411
412			inline void
413		30080x	win_wait_reactor::cancel_wait(overlapped_op* op)
414			{
415			{
416	1/1 ✓ Branch 2 → 3 taken 30080 times.	30080x	std::lock_guard lock(mutex_);
417			// Same refusal queue_register makes, for the same reason: once
418			// the polling thread is gone nothing reads pending_cancel_
419			// again, so a cancel queued here would sit there for good. It
420			// has nothing to cancel either -- the drain those exits run
421			// already answered whatever was parked, and a register that
422			// arrived after them was refused at its caller.
423	6/6 ✓ Branch 4 → 5 taken 30074 times. ✓ Branch 4 → 6 taken 6 times. ✓ Branch 5 → 6 taken 6 times. ✓ Branch 5 → 7 taken 30068 times. ✓ Branch 8 → 9 taken 12 times. ✓ Branch 8 → 10 taken 30068 times.	30080x	if (stop_.load(std::memory_order_acquire) \|\| dead_err_)
424		12x	return;
425	1/1 ✓ Branch 10 → 11 taken 30068 times.	30068x	pending_cancel_.push_back(op);
426		30080x	}
427		30068x	wake_self();
428			}
429
430			inline void
431		3741x	win_wait_reactor::stop()
432			{
433	2/2 ✓ Branch 3 → 4 taken 2494 times. ✓ Branch 3 → 5 taken 1247 times.	3741x	if (stop_.exchange(true, std::memory_order_acq_rel))
434		2494x	return;
435		1247x	wake_self();
436			// Moved out under the lock, then joined without it: the reactor
437			// thread takes the same lock on every pass, so joining while
438			// holding it would deadlock. A context that never waited has no
439			// thread here at all.
440		1247x	std::thread t;
441			{
442	1/1 ✓ Branch 7 → 8 taken 1247 times.	1247x	std::lock_guard lock(mutex_);
443		1247x	t = std::move(thread_);
444		1247x	}
445	2/2 ✓ Branch 12 → 13 taken 36 times. ✓ Branch 12 → 14 taken 1211 times.	1247x	if (t.joinable())
446	1/1 ✓ Branch 13 → 14 taken 36 times.	36x	t.join();
447		1247x	}
448
449			inline bool
450		1x	win_wait_reactor::drop_refused_entries()
451			{
452			// A descriptor the provider no longer recognises either comes back
453			// POLLNVAL on its own entry, which the revents walk already
454			// answers, or refuses the whole call and reaches here. Asking
455			// about each entry alone covers both without depending on which
456			// one this provider does. A socket closed under a parked wait is
457			// the ordinary way to get here: the reactor is told to drop the
458			// entry, but the handle can go before that ask is read.
459		1x	bool dropped = false;
460	2/2 ✓ Branch 16 → 3 taken 1 time. ✓ Branch 16 → 17 taken 1 time.	2x	for (std::size_t i = registered_.size(); i > 0; --i)
461			{
462		1x	auto const& e = registered_[i - 1];
463		1x	WSAPOLLFD pfd{e.fd, events_for_wait(e.w), 0};
464	2/3 ✓ Branch 5 → 6 taken 1 time. ✓ Branch 6 → 7 taken 1 time. ✗ Branch 6 → 8 not taken.	1x	if (::WSAPoll(&pfd, 1, 0) != SOCKET_ERROR)
465		1x	continue;
466
467		✗	sched_.on_completion(e.op, last_error(), 0);
468		✗	registered_.erase(registered_.begin() + (i - 1));
469		✗	dropped = true;
470			}
471		1x	return dropped;
472			}
473
474			inline void
475		36x	win_wait_reactor::run()
476			{
477		36x	std::vector<WSAPOLLFD> pollfds;
478
479			// What the ops still parked here are told on the way out, and what
480			// every later register_wait is told too. Ending on a poll the
481			// provider refused is not a cancellation, and an op that reports
482			// one hides the reason its wait could not be kept.
483		36x	DWORD drain_err = ERROR_OPERATION_ABORTED;
484
485	4/4 ✓ Branch 108 → 3 taken 224 times. ✓ Branch 108 → 109 taken 34 times. ✓ Branch 110 → 3 taken 9 times. ✓ Branch 110 → 111 taken 1 time.	268x	while (!stop_.load(std::memory_order_acquire))
486			{
487			// Drain pending register/cancel under the lock.
488		233x	std::vector<entry> to_add;
489		233x	std::vector<overlapped_op*> to_cancel;
490			{
491	1/1 ✓ Branch 3 → 4 taken 233 times.	233x	std::lock_guard lock(mutex_);
492		233x	to_add.swap(pending_register_);
493		233x	to_cancel.swap(pending_cancel_);
494		233x	}
495
496	2/2 ✓ Branch 13 → 9 taken 43 times. ✓ Branch 13 → 14 taken 233 times.	276x	for (auto& e : to_add)
497	1/1 ✓ Branch 10 → 11 taken 43 times.	43x	registered_.push_back(e);
498
499	2/2 ✓ Branch 33 → 16 taken 421 times. ✓ Branch 33 → 34 taken 233 times.	654x	for (auto* op : to_cancel)
500			{
501			// A socket's close() and cancel() ask the reactor to drop
502			// their wait op whether or not one is parked -- open() goes
503			// through close_socket() before it has a socket at all --
504			// and such an ask can outlive the op's next reset().
505			// Acting on it then would complete the wait that reset
506			// started, the moment it is registered. Every real cancel
507			// flags the op before queueing the ask, and only reset()
508			// clears the flag, so an unflagged op is one of those
509			// stale asks.
510	2/2 ✓ Branch 18 → 19 taken 37 times. ✓ Branch 18 → 20 taken 384 times.	421x	if (!op->cancelled.load(std::memory_order_acquire))
511		37x	continue;
512
513	1/1 ✓ Branch 22 → 23 taken 384 times.	384x	auto it = std::find_if(
514			registered_.begin(), registered_.end(),
515		147x	[op](entry const& e) { return e.op == op; });
516	2/2 ✓ Branch 25 → 26 taken 15 times. ✓ Branch 25 → 30 taken 369 times.	384x	if (it != registered_.end())
517			{
518			// The op's cancelled flag has already been set by
519			// request_cancel; invoke_handler will translate it.
520	1/1 ✓ Branch 26 → 27 taken 15 times.	15x	sched_.on_completion(op, 0, 0);
521	1/1 ✓ Branch 28 → 29 taken 15 times.	15x	registered_.erase(it);
522			}
523			// If not in registered_, the op already fired — no-op.
524			}
525
526			// Build the poll set. Slot 0 is the wakeup socket.
527		233x	pollfds.clear();
528	1/1 ✓ Branch 36 → 37 taken 233 times.	233x	pollfds.reserve(registered_.size() + 1);
529	1/1 ✓ Branch 37 → 38 taken 233 times.	233x	pollfds.push_back({wakeup_read_, POLLRDNORM, 0});
530	2/2 ✓ Branch 45 → 40 taken 42 times. ✓ Branch 45 → 46 taken 233 times.	275x	for (auto& e : registered_)
531	1/1 ✓ Branch 42 → 43 taken 42 times.	42x	pollfds.push_back({e.fd, events_for_wait(e.w), 0});
532
533			// Block until the self-pipe (slot 0) is poked by a register,
534			// cancel, or stop, or a watched socket becomes ready. No
535			// periodic timeout, so an idle reactor consumes no CPU: the
536			// self-pipe is the only thing that ends this wait, and
537			// wake_self() leaves the channel free for the next poke when
538			// its own send fails, so a lost wake costs the wakes already in
539			// flight rather than every wake after it.
540	1/1 ✓ Branch 48 → 49 taken 233 times.	233x	int n = ::WSAPoll(
541		233x	pollfds.data(), static_cast<ULONG>(pollfds.size()),
542			-1 /* infinite */);
543	2/2 ✓ Branch 49 → 50 taken 1 time. ✓ Branch 49 → 60 taken 232 times.	233x	if (n == SOCKET_ERROR)
544			{
545			// A refusal one entry accounts for costs that entry its
546			// wait and nothing else; the rest of the set keeps being
547			// polled, and a later register still finds a live reactor.
548			// Only a failure no entry explains ends the loop.
549		1x	DWORD const err = last_error();
550	4/7 ✓ Branch 52 → 53 taken 1 time. ✗ Branch 52 → 55 not taken. ✓ Branch 53 → 54 taken 1 time. ✓ Branch 54 → 55 taken 1 time. ✗ Branch 54 → 56 not taken. ✓ Branch 57 → 58 taken 1 time. ✗ Branch 57 → 59 not taken.	1x	if (registered_.empty() \|\| !drop_refused_entries())
551			{
552		1x	drain_err = err;
553		1x	break;
554			}
555		✗	continue;
556		✗	}
557
558			// Drain the wakeup socket so it stops reporting readable.
559	2/3 ✓ Branch 61 → 62 taken 225 times. ✓ Branch 61 → 68 taken 7 times. ✗ Branch 61 → 69 not taken.	232x	if (pollfds[0].revents != 0)
560			{
561			char buf[64];
562			for (;;)
563			{
564	2/2 ✓ Branch 62 → 63 taken 432 times. ✓ Branch 63 → 64 taken 18 times.	450x	int r = ::recv(wakeup_read_, buf, sizeof(buf), 0);
565	4/4 ✓ Branch 63 → 64 taken 216 times. ✓ Branch 63 → 65 taken 216 times. ✓ Branch 64 → 65 taken 9 times. ✓ Branch 64 → 66 taken 9 times.	450x	if (r <= 0)
566		225x	break;
567		225x	}
568		225x	wake_pending_.store(false, std::memory_order_release);
569			}
570
571			// Walk events in reverse so erases don't invalidate later indices.
572	4/4 ✓ Branch 94 → 69 taken 40 times. ✓ Branch 94 → 95 taken 223 times. ✓ Branch 95 → 70 taken 1 time. ✓ Branch 95 → 96 taken 9 times.	273x	for (std::size_t i = pollfds.size(); i > 1; --i)
573			{
574		41x	auto const& pfd = pollfds[i - 1];
575	3/4 ✓ Branch 70 → 71 taken 14 times. ✓ Branch 70 → 72 taken 26 times. ✗ Branch 71 → 72 not taken. ✓ Branch 71 → 73 taken 1 time.	41x	if (pfd.revents == 0)
576		14x	continue;
577
578		27x	auto const& e = registered_[i - 2];
579	2/4 ✗ Branch 74 → 75 not taken. ✓ Branch 74 → 76 taken 26 times. ✗ Branch 75 → 76 not taken. ✓ Branch 75 → 77 taken 1 time.	27x	if (!ready_for_wait(e.w, pfd.revents))
580		✗	continue;
581
582		27x	DWORD err = 0;
583		27x	constexpr SHORT err_bits = POLLERR \| POLLHUP \| POLLNVAL;
584	3/4 ✓ Branch 76 → 77 taken 5 times. ✓ Branch 76 → 87 taken 21 times. ✗ Branch 77 → 78 not taken. ✓ Branch 77 → 88 taken 1 time.	27x	if (pfd.revents & err_bits)
585			{
586		5x	int so_err = 0;
587		5x	int sz = sizeof(so_err);
588		15x	if (::getsockopt(
589	1/2 ✓ Branch 77 → 78 taken 5 times. ✗ Branch 78 → 79 not taken.	5x	e.fd, SOL_SOCKET, SO_ERROR,
590	3/8 ✓ Branch 78 → 79 taken 3 times. ✓ Branch 78 → 81 taken 2 times. ✗ Branch 79 → 80 not taken. ✗ Branch 79 → 82 not taken. ✗ Branch 82 → 83 not taken. ✓ Branch 82 → 84 taken 5 times. ✗ Branch 83 → 84 not taken. ✗ Branch 83 → 85 not taken.	8x	reinterpret_cast<char*>(&so_err), &sz) == 0 &&
591	1/4 ✗ Branch 79 → 80 not taken. ✓ Branch 79 → 81 taken 3 times. ✗ Branch 80 → 81 not taken. ✗ Branch 80 → 82 not taken.	3x	so_err != 0)
592			{
593		✗	err = static_cast<DWORD>(so_err);
594			}
595	2/4 ✓ Branch 84 → 85 taken 3 times. ✓ Branch 84 → 86 taken 2 times. ✗ Branch 85 → 86 not taken. ✗ Branch 85 → 87 not taken.	5x	else if (e.w == wait_type::error)
596			{
597			// wait_type::error fires on the error condition;
598			// the contract is to report a non-zero error_code.
599		3x	err = WSAECONNABORTED;
600			}
601			}
602
603	2/2 ✓ Branch 87 → 88 taken 26 times. ✓ Branch 88 → 89 taken 1 time.	27x	sched_.on_completion(e.op, err, 0);
604	2/2 ✓ Branch 91 → 92 taken 26 times. ✓ Branch 92 → 93 taken 1 time.	27x	registered_.erase(registered_.begin() + (i - 2));
605			}
606		234x	}
607
608			// Drain remaining ops on the way out. This must cover both the
609			// active set and anything still queued by user threads that hasn't
610			// been moved into registered_ yet, otherwise those ops leak
611			// work_started credit and stall scheduler shutdown.
612			{
613	2/2 ✓ Branch 109 → 110 taken 35 times. ✓ Branch 111 → 112 taken 1 time.	36x	std::lock_guard lock(mutex_);
614			// Closing the door and taking what is behind it in one critical
615			// section is what leaves no register in between: one that got
616			// in is drained here, one that arrives after is refused by
617			// queue_register and completes with the same code at its
618			// caller.
619		36x	dead_err_ = drain_err;
620	2/4 ✗ Branch 116 → 112 not taken. ✓ Branch 116 → 117 taken 35 times. ✗ Branch 118 → 114 not taken. ✓ Branch 118 → 119 taken 1 time.	36x	for (auto& e : pending_register_)
621		✗	registered_.push_back(e);
622		36x	pending_register_.clear();
623		36x	pending_cancel_.clear();
624		36x	}
625	3/4 ✓ Branch 126 → 122 taken 1 time. ✓ Branch 126 → 127 taken 35 times. ✗ Branch 128 → 124 not taken. ✓ Branch 128 → 129 taken 1 time.	37x	for (auto& e : registered_)
626	1/2 ✓ Branch 123 → 124 taken 1 time. ✗ Branch 125 → 126 not taken.	1x	sched_.on_completion(e.op, drain_err, 0);
627		36x	registered_.clear();
628		36x	}
629
630			} // namespace boost::corosio::detail
631
632			#endif // BOOST_COROSIO_HAS_IOCP
633
634			#endif // BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_WAIT_REACTOR_HPP
635