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

83.3% Lines (130/156) 100.0% List of functions (13/13) 71.2% Branches (79/111)

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

win_wait_reactor.hpp

f(x) Functions (13)

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
32			#include <Ws2tcpip.h>
33
34			#include <algorithm>
35			#include <atomic>
36			#include <cstddef>
37			#include <mutex>
38			#include <thread>
39			#include <vector>
40
41			namespace boost::corosio::detail {
42
43			/** Auxiliary select-based reactor for IOCP wait operations.
44
45			IOCP has no native primitive for socket readiness without I/O.
46			For cases where a zero-byte WSARecv won't work (datagram-read,
47			acceptor-read, write-wait, error-wait), this reactor runs a
48			dedicated thread using WSAPoll to detect readiness and posts a
49			synthetic completion to the owning IOCP scheduler via
50			win_scheduler::on_completion().
51
52			The same dispatch path used by overlapped I/O then delivers the
53			completion to the user's coroutine, so the public API is uniform
54			across backends.
55
56			Per-op lifecycle:
57			1. Caller sets up an overlapped_op (h, ex, ec_out, cancelled flag).
58			2. Caller calls register_wait(fd, w, op) and returns
59			std::noop_coroutine. The op is parked in the reactor's table.
60			3. Reactor thread polls. When the fd is ready, the op is removed
61			from the table and posted to the scheduler. The error code
62			delivered to the completion is: ec={} on success; the SO_ERROR
63			value if error revents fired and SO_ERROR is set; or
64			WSAECONNABORTED as a synthesized fallback for wait_type::error
65			when error revents fired but SO_ERROR returned zero.
66			4. On socket cancel(), the user's thread calls cancel_wait(op),
67			which queues a cancel request. The reactor thread removes the
68			op from the table and posts a completion; invoke_handler sees
69			op.cancelled==true and yields capy::cond::canceled.
70
71			Thread-safe: register_wait, cancel_wait, and stop may be called
72			from any thread.
73			*/
74			class win_wait_reactor
75			{
76			public:
77			explicit win_wait_reactor(win_scheduler& sched);
78			~win_wait_reactor();
79
80			win_wait_reactor(win_wait_reactor const&) = delete;
81			win_wait_reactor& operator=(win_wait_reactor const&) = delete;
82
83			/// Park an overlapped_op until @p fd is ready for @p w.
84			void register_wait(SOCKET fd, wait_type w, overlapped_op* op);
85
86			/// Remove a parked op and post a completion. Idempotent.
87			void cancel_wait(overlapped_op* op);
88
89			/// Stop the reactor thread and drain remaining ops as cancelled.
90			void stop();
91
92			private:
93			struct entry
94			{
95			SOCKET fd = INVALID_SOCKET;
96			wait_type w = wait_type::read;
97			overlapped_op* op = nullptr;
98			};
99
100			void run();
101			void wake_self() noexcept;
102			void make_wakeup_pair();
103			void close_wakeup_pair() noexcept;
104
105		26x	static SHORT events_for_wait(wait_type w) noexcept
106			{
107	3/3 ✓ Branch 2 → 3 taken 19 times. ✓ Branch 2 → 4 taken 5 times. ✓ Branch 2 → 5 taken 2 times.	26x	switch (w)
108			{
109		19x	case wait_type::read: return POLLRDNORM;
110		5x	case wait_type::write: return POLLWRNORM;
111		2x	default: return POLLPRI;
112			}
113			}
114
115		14x	static bool ready_for_wait(wait_type w, SHORT revents) noexcept
116			{
117		14x	constexpr SHORT err_bits = POLLERR \| POLLHUP \| POLLNVAL;
118	2/3 ✓ Branch 2 → 3 taken 9 times. ✓ Branch 2 → 4 taken 5 times. ✗ Branch 2 → 5 not taken.	14x	switch (w)
119			{
120		9x	case wait_type::read:
121		9x	return (revents & (POLLRDNORM \| POLLRDBAND \| err_bits)) != 0;
122		5x	case wait_type::write:
123		5x	return (revents & (POLLWRNORM \| POLLWRBAND \| err_bits)) != 0;
124		✗	default:
125		✗	return (revents & (POLLPRI \| err_bits)) != 0;
126			}
127			}
128
129			win_scheduler& sched_;
130
131			SOCKET wakeup_read_ = INVALID_SOCKET;
132			SOCKET wakeup_write_ = INVALID_SOCKET;
133
134			std::mutex mutex_;
135			std::vector<entry> pending_register_;
136			std::vector<overlapped_op*> pending_cancel_;
137			std::atomic<bool> stop_{false};
138			std::atomic<bool> wake_pending_{false};
139
140			std::vector<entry> registered_; // reactor-thread-only
141
142			std::thread thread_;
143			};
144
145		24x	inline win_wait_reactor::win_wait_reactor(win_scheduler& sched)
146		24x	: sched_(sched)
147			{
148	1/1 ✓ Branch 9 → 10 taken 24 times.	24x	make_wakeup_pair();
149	1/1 ✓ Branch 10 → 11 taken 24 times.	48x	thread_ = std::thread([this] { run(); });
150		24x	}
151
152		24x	inline win_wait_reactor::~win_wait_reactor()
153			{
154		24x	stop();
155		24x	close_wakeup_pair();
156		24x	}
157
158			inline void
159		24x	win_wait_reactor::make_wakeup_pair()
160			{
161			// Build a pair of connected loopback sockets to use as a wakeup
162			// channel. Winsock has no socketpair(2), so we listen on
163			// 127.0.0.1:0, connect a peer, then accept it.
164	1/1 ✓ Branch 2 → 3 taken 24 times.	24x	SOCKET listener = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
165	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 5 taken 24 times.	24x	if (listener == INVALID_SOCKET)
166		✗	return;
167
168		24x	sockaddr_in addr{};
169		24x	addr.sin_family = AF_INET;
170	1/1 ✓ Branch 5 → 6 taken 24 times.	24x	addr.sin_addr.s_addr = ::htonl(INADDR_LOOPBACK);
171		24x	addr.sin_port = 0;
172
173		24x	int len = sizeof(addr);
174	1/1 ✓ Branch 6 → 7 taken 24 times.	24x	if (::bind(listener, reinterpret_cast<sockaddr*>(&addr), len) ==
175		24x	SOCKET_ERROR \|\|
176	4/7 ✓ Branch 7 → 8 taken 24 times. ✗ Branch 7 → 12 not taken. ✓ Branch 8 → 9 taken 24 times. ✓ Branch 9 → 10 taken 24 times. ✗ Branch 9 → 12 not taken. ✗ Branch 14 → 15 not taken. ✓ Branch 14 → 17 taken 24 times.	48x	::listen(listener, 1) == SOCKET_ERROR \|\|
177	2/3 ✓ Branch 10 → 11 taken 24 times. ✗ Branch 11 → 12 not taken. ✓ Branch 11 → 13 taken 24 times.	24x	::getsockname(listener, reinterpret_cast<sockaddr*>(&addr), &len) ==
178			SOCKET_ERROR)
179			{
180		✗	::closesocket(listener);
181		✗	return;
182			}
183
184	1/1 ✓ Branch 17 → 18 taken 24 times.	24x	wakeup_write_ = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
185	1/2 ✗ Branch 18 → 19 not taken. ✓ Branch 18 → 21 taken 24 times.	24x	if (wakeup_write_ == INVALID_SOCKET)
186			{
187		✗	::closesocket(listener);
188		✗	return;
189			}
190
191	1/1 ✓ Branch 21 → 22 taken 24 times.	24x	if (::connect(
192	1/2 ✗ Branch 22 → 23 not taken. ✓ Branch 22 → 26 taken 24 times.	24x	wakeup_write_, reinterpret_cast<sockaddr*>(&addr), len) ==
193			SOCKET_ERROR)
194			{
195		✗	::closesocket(wakeup_write_);
196		✗	wakeup_write_ = INVALID_SOCKET;
197		✗	::closesocket(listener);
198		✗	return;
199			}
200
201	1/1 ✓ Branch 26 → 27 taken 24 times.	24x	wakeup_read_ = ::accept(listener, nullptr, nullptr);
202	1/1 ✓ Branch 27 → 28 taken 24 times.	24x	::closesocket(listener);
203
204	1/2 ✗ Branch 28 → 29 not taken. ✓ Branch 28 → 31 taken 24 times.	24x	if (wakeup_read_ == INVALID_SOCKET)
205			{
206		✗	::closesocket(wakeup_write_);
207		✗	wakeup_write_ = INVALID_SOCKET;
208		✗	return;
209			}
210
211			// The drain loop in run() calls recv() until it returns <= 0.
212			// With a blocking socket that second recv() would block instead
213			// of returning WSAEWOULDBLOCK, deadlocking the reactor thread.
214		24x	u_long non_blocking = 1;
215	1/1 ✓ Branch 31 → 32 taken 24 times.	24x	::ioctlsocket(wakeup_read_, FIONBIO, &non_blocking);
216			}
217
218			inline void
219		24x	win_wait_reactor::close_wakeup_pair() noexcept
220			{
221	1/2 ✓ Branch 2 → 3 taken 24 times. ✗ Branch 2 → 5 not taken.	24x	if (wakeup_read_ != INVALID_SOCKET)
222			{
223		24x	::closesocket(wakeup_read_);
224		24x	wakeup_read_ = INVALID_SOCKET;
225			}
226	1/2 ✓ Branch 5 → 6 taken 24 times. ✗ Branch 5 → 8 not taken.	24x	if (wakeup_write_ != INVALID_SOCKET)
227			{
228		24x	::closesocket(wakeup_write_);
229		24x	wakeup_write_ = INVALID_SOCKET;
230			}
231		24x	}
232
233			inline void
234		206x	win_wait_reactor::wake_self() noexcept
235			{
236			// Coalesce wakes: only send a byte if no wake is already pending.
237		206x	bool expected = false;
238	2/2 ✓ Branch 3 → 4 taken 68 times. ✓ Branch 3 → 5 taken 138 times.	206x	if (!wake_pending_.compare_exchange_strong(
239			expected, true, std::memory_order_acq_rel))
240		68x	return;
241	1/2 ✓ Branch 5 → 6 taken 138 times. ✗ Branch 5 → 12 not taken.	138x	if (wakeup_write_ != INVALID_SOCKET)
242			{
243		138x	char b = 0;
244	1/2 ✗ Branch 7 → 8 not taken. ✓ Branch 7 → 11 taken 138 times.	138x	if (::send(wakeup_write_, &b, 1, 0) == SOCKET_ERROR)
245			{
246			// The self-pipe byte is the only thing that wakes the reactor
247			// thread from an indefinite poll(); a coalesced lost wakeup
248			// would leave wake_pending_ stuck true and hang the op.
249			// wakeup_write_ is a blocking socket, so a 1-byte send can
250			// only fail on a hard error -- fatal, mirroring a failed
251			// PostQueuedCompletionStatus in win_scheduler.
252		✗	detail::throw_system_error(make_err(::WSAGetLastError()));
253			}
254			}
255			}
256
257			inline void
258		25x	win_wait_reactor::register_wait(
259			SOCKET fd, wait_type w, overlapped_op* op)
260			{
261			// If the op was already cancelled (e.g. pre-cancelled stop_token
262			// fired synchronously before this call), complete immediately
263			// instead of registering. Otherwise the reactor would park the
264			// op forever because the earlier cancel_wait() found nothing to
265			// cancel in registered_.
266	2/2 ✓ Branch 3 → 4 taken 2 times. ✓ Branch 3 → 6 taken 23 times.	25x	if (op->cancelled.load(std::memory_order_acquire))
267			{
268		2x	sched_.on_completion(op, 0, 0);
269		2x	return;
270			}
271			{
272	1/1 ✓ Branch 6 → 7 taken 23 times.	23x	std::lock_guard lock(mutex_);
273	1/1 ✓ Branch 7 → 8 taken 23 times.	23x	pending_register_.push_back(entry{fd, w, op});
274		23x	}
275		23x	wake_self();
276			}
277
278			inline void
279		159x	win_wait_reactor::cancel_wait(overlapped_op* op)
280			{
281			{
282	1/1 ✓ Branch 2 → 3 taken 159 times.	159x	std::lock_guard lock(mutex_);
283	1/1 ✓ Branch 3 → 4 taken 159 times.	159x	pending_cancel_.push_back(op);
284		159x	}
285		159x	wake_self();
286		159x	}
287
288			inline void
289		72x	win_wait_reactor::stop()
290			{
291	2/2 ✓ Branch 3 → 4 taken 48 times. ✓ Branch 3 → 5 taken 24 times.	72x	if (stop_.exchange(true, std::memory_order_acq_rel))
292		48x	return;
293		24x	wake_self();
294	1/2 ✓ Branch 7 → 8 taken 24 times. ✗ Branch 7 → 9 not taken.	24x	if (thread_.joinable())
295		24x	thread_.join();
296			}
297
298			inline void
299		24x	win_wait_reactor::run()
300			{
301		24x	std::vector<WSAPOLLFD> pollfds;
302
303	2/2 ✓ Branch 93 → 3 taken 142 times. ✓ Branch 93 → 94 taken 22 times.	164x	while (!stop_.load(std::memory_order_acquire))
304			{
305			// Drain pending register/cancel under the lock.
306		142x	std::vector<entry> to_add;
307		142x	std::vector<overlapped_op*> to_cancel;
308			{
309	1/1 ✓ Branch 3 → 4 taken 142 times.	142x	std::lock_guard lock(mutex_);
310		142x	to_add.swap(pending_register_);
311		142x	to_cancel.swap(pending_cancel_);
312		142x	}
313
314	2/2 ✓ Branch 13 → 9 taken 23 times. ✓ Branch 13 → 14 taken 142 times.	165x	for (auto& e : to_add)
315	1/1 ✓ Branch 10 → 11 taken 23 times.	23x	registered_.push_back(e);
316
317	2/2 ✓ Branch 29 → 16 taken 149 times. ✓ Branch 29 → 30 taken 142 times.	291x	for (auto* op : to_cancel)
318			{
319	1/1 ✓ Branch 19 → 20 taken 149 times.	149x	auto it = std::find_if(
320			registered_.begin(), registered_.end(),
321		9x	[op](entry const& e) { return e.op == op; });
322	2/2 ✓ Branch 22 → 23 taken 7 times. ✓ Branch 22 → 27 taken 142 times.	149x	if (it != registered_.end())
323			{
324			// The op's cancelled flag has already been set by
325			// request_cancel; invoke_handler will translate it.
326	1/1 ✓ Branch 23 → 24 taken 7 times.	7x	sched_.on_completion(op, 0, 0);
327	1/1 ✓ Branch 25 → 26 taken 7 times.	7x	registered_.erase(it);
328			}
329			// If not in registered_, the op already fired — no-op.
330			}
331
332			// Build the poll set. Slot 0 is the wakeup socket.
333		142x	pollfds.clear();
334	1/1 ✓ Branch 32 → 33 taken 142 times.	142x	pollfds.reserve(registered_.size() + 1);
335	1/1 ✓ Branch 33 → 34 taken 142 times.	142x	pollfds.push_back({wakeup_read_, POLLRDNORM, 0});
336	2/2 ✓ Branch 41 → 36 taken 26 times. ✓ Branch 41 → 42 taken 142 times.	168x	for (auto& e : registered_)
337	1/1 ✓ Branch 38 → 39 taken 26 times.	26x	pollfds.push_back({e.fd, events_for_wait(e.w), 0});
338
339			// Block until the self-pipe (slot 0) is poked by a register,
340			// cancel, or stop, or a watched socket becomes ready. There is
341			// no periodic safety-net timeout: a lost self-pipe wakeup is
342			// fatal in wake_self() (the byte send can only fail on a hard
343			// error), so an idle reactor consumes no CPU.
344	1/1 ✓ Branch 44 → 45 taken 142 times.	142x	int n = ::WSAPoll(
345			pollfds.data(),
346		142x	static_cast<ULONG>(pollfds.size()),
347			-1 /* infinite */);
348	2/2 ✓ Branch 45 → 46 taken 2 times. ✓ Branch 45 → 47 taken 140 times.	142x	if (n == SOCKET_ERROR)
349		2x	break;
350
351			// Drain the wakeup socket so it stops reporting readable.
352	2/2 ✓ Branch 48 → 49 taken 136 times. ✓ Branch 48 → 55 taken 4 times.	140x	if (pollfds[0].revents != 0)
353			{
354			char buf[64];
355			for (;;)
356			{
357	1/1 ✓ Branch 49 → 50 taken 272 times.	272x	int r = ::recv(wakeup_read_, buf, sizeof(buf), 0);
358	2/2 ✓ Branch 50 → 51 taken 136 times. ✓ Branch 50 → 52 taken 136 times.	272x	if (r <= 0)
359		136x	break;
360		136x	}
361		136x	wake_pending_.store(false, std::memory_order_release);
362			}
363
364			// Walk events in reverse so erases don't invalidate later indices.
365	2/2 ✓ Branch 81 → 56 taken 24 times. ✓ Branch 81 → 82 taken 140 times.	164x	for (std::size_t i = pollfds.size(); i > 1; --i)
366			{
367		24x	auto const& pfd = pollfds[i - 1];
368	2/2 ✓ Branch 57 → 58 taken 10 times. ✓ Branch 57 → 59 taken 14 times.	24x	if (pfd.revents == 0)
369		10x	continue;
370
371		14x	auto const& e = registered_[i - 2];
372	1/2 ✗ Branch 61 → 62 not taken. ✓ Branch 61 → 63 taken 14 times.	14x	if (!ready_for_wait(e.w, pfd.revents))
373		✗	continue;
374
375		14x	DWORD err = 0;
376		14x	constexpr SHORT err_bits = POLLERR \| POLLHUP \| POLLNVAL;
377	1/2 ✗ Branch 63 → 64 not taken. ✓ Branch 63 → 74 taken 14 times.	14x	if (pfd.revents & err_bits)
378			{
379		✗	int so_err = 0;
380		✗	int sz = sizeof(so_err);
381		✗	if (::getsockopt(
382		✗	e.fd, SOL_SOCKET, SO_ERROR,
383		✗	reinterpret_cast<char*>(&so_err), &sz) == 0 &&
384		✗	so_err != 0)
385			{
386		✗	err = static_cast<DWORD>(so_err);
387			}
388		✗	else if (e.w == wait_type::error)
389			{
390			// wait_type::error fires on the error condition;
391			// the contract is to report a non-zero error_code.
392		✗	err = WSAECONNABORTED;
393			}
394			}
395
396	1/1 ✓ Branch 74 → 75 taken 14 times.	14x	sched_.on_completion(e.op, err, 0);
397	1/1 ✓ Branch 78 → 79 taken 14 times.	14x	registered_.erase(registered_.begin() + (i - 2));
398			}
399		144x	}
400
401			// Drain remaining ops as cancelled on shutdown. This must cover
402			// both the active set and anything still queued by user threads
403			// that hasn't been moved into registered_ yet, otherwise those
404			// ops leak work_started credit and stall scheduler shutdown.
405			{
406	1/1 ✓ Branch 94 → 95 taken 24 times.	24x	std::lock_guard lock(mutex_);
407	1/2 ✗ Branch 101 → 97 not taken. ✓ Branch 101 → 102 taken 24 times.	24x	for (auto& e : pending_register_)
408		✗	registered_.push_back(e);
409		24x	pending_register_.clear();
410		24x	pending_cancel_.clear();
411		24x	}
412	2/2 ✓ Branch 111 → 107 taken 2 times. ✓ Branch 111 → 112 taken 24 times.	26x	for (auto& e : registered_)
413	1/1 ✓ Branch 108 → 109 taken 2 times.	2x	sched_.on_completion(e.op, ERROR_OPERATION_ABORTED, 0);
414		24x	registered_.clear();
415		24x	}
416
417			} // namespace boost::corosio::detail
418
419			#endif // BOOST_COROSIO_HAS_IOCP
420
421			#endif // BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_WAIT_REACTOR_HPP
422