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

94.8% Lines (201 / 212) 79.5% Branches (136 / 171) 100.0% Functions (16 / 16)

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