include/boost/corosio/native/detail/reactor/reactor_op.hpp

81.5% Lines (203/249) 88.6% List of functions (234/264) 11.4% Branches (74/648)

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

reactor_op.hpp

f(x) Functions (264)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Steve Gerbino
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_REACTOR_REACTOR_OP_HPP
11			#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
12
13			#include <boost/corosio/native/detail/reactor/reactor_events.hpp>
14			#include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
15			#include <boost/corosio/io/io_object.hpp>
16			#include <boost/corosio/endpoint.hpp>
17			#include <boost/capy/ex/executor_ref.hpp>
18
19			#include <atomic>
20			#include <cstddef>
21			#include <optional>
22			#include <stop_token>
23
24			#include <errno.h>
25			#include <poll.h>
26
27			#include <netinet/in.h>
28			#include <sys/socket.h>
29			#include <sys/uio.h>
30
31			namespace boost::corosio::detail {
32
33			/** Base operation for reactor-based backends.
34
35			Holds per-operation state that depends on the concrete backend
36			socket/acceptor types: coroutine handle, executor, output
37			pointers, file descriptor, stop_callback, and type-specific
38			impl pointers.
39
40			Fields shared across all backends (errn, bytes_transferred,
41			cancelled, impl_ptr, perform_io, complete) live in
42			reactor_op_base so the scheduler and descriptor_state can
43			access them without template instantiation.
44
45			@tparam Socket The backend socket impl type (forward-declared).
46			@tparam Acceptor The backend acceptor impl type (forward-declared).
47			*/
48			template<class Socket, class Acceptor>
49			struct reactor_op : reactor_op_base
50			{
51			// The op envelope — coroutine handle h, cont, executor ex, ec_out,
52			// bytes_out, cancelled, stop_cb (+ its canceller), impl_ptr — lives in
53			// coro_op (via reactor_op_base) and is shared with io_uring/IOCP.
54			// reactor_op adds only the reactor-specific routing state below.
55
56			/// File descriptor this operation targets.
57		152200x	int fd = -1;
58
59			/// Owning socket impl (for stop_token cancellation routing).
60		152200x	Socket* socket_impl_ = nullptr;
61
62			/// Owning acceptor impl (for stop_token cancellation routing).
63		152200x	Acceptor* acceptor_impl_ = nullptr;
64
65		304400x	reactor_op() = default;
66
67			/// Reset operation state for reuse.
68		1661303x	void reset() noexcept
69			{
70		1661303x	fd = -1;
71		1661303x	errn = 0;
72		1661303x	bytes_transferred = 0;
73		1661303x	cancelled.store(false, std::memory_order_relaxed);
74		1661303x	impl_ptr.reset();
75		1661303x	socket_impl_ = nullptr;
76		1661303x	acceptor_impl_ = nullptr;
77		1661303x	}
78
79			/// Return true if this is a read-direction operation.
80		164846x	virtual bool is_read_operation() const noexcept
81			{
82		164846x	return false;
83			}
84
85			/// Cancel this operation via the owning impl.
86			virtual void cancel() noexcept = 0;
87
88			/// coro_op cancellation hook (fired by the shared canceller when the
89			/// stop_token requests cancellation): route to the impl-specific cancel().
90		1075x	void on_cancel() noexcept override
91			{
92		1075x	cancel();
93		1075x	}
94
95			/// Destroy without invoking.
96		16x	void destroy() override
97			{
98		16x	stop_cb.reset();
99		16x	reactor_op_base::destroy();
100		16x	}
101
102			/// Arm the stop-token callback for a socket operation.
103		392627x	void start(std::stop_token const& token, Socket* impl)
104			{
105		392627x	socket_impl_ = impl;
106		392627x	acceptor_impl_ = nullptr;
107		392627x	coro_op::start(token);
108		392627x	}
109
110			/// Arm the stop-token callback for an acceptor operation.
111		7628x	void start(std::stop_token const& token, Acceptor* impl)
112			{
113		7628x	socket_impl_ = nullptr;
114		7628x	acceptor_impl_ = impl;
115		7628x	coro_op::start(token);
116		7628x	}
117			};
118
119			/** Shared connect operation.
120
121			Checks SO_ERROR for connect completion status. The operator()()
122			and cancel() are provided by the concrete backend type.
123
124			@tparam Base The backend's base op type.
125			@tparam Endpoint The endpoint type (endpoint or local_endpoint).
126			*/
127			template<class Base, class Endpoint = endpoint>
128			struct reactor_connect_op : Base
129			{
130			/// Endpoint to connect to.
131			Endpoint target_endpoint;
132
133			/// Reset operation state for reuse.
134		7606x	void reset() noexcept
135			{
136		7606x	Base::reset();
137		7606x	target_endpoint = Endpoint{};
138		7606x	}
139
140		7523x	void perform_io() noexcept override
141			{
142			// A readiness notification does not prove the handshake
143			// finished: fresh sockets raise a spurious writable event,
144			// and a cached edge can trigger this check while the connect
145			// is still in flight — where SO_ERROR also reads 0. Probe
146			// writability first and report EAGAIN to stay parked;
147			// SO_ERROR decides only once the socket is actually writable.
148		7523x	pollfd pfd{};
149		7523x	pfd.fd = this->fd;
150		7523x	pfd.events = POLLOUT;
151			int r;
152		7523x	do
153			{
154	2/16 ✓ Branch 0 taken 5837 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 1686 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	7523x	r = ::poll(&pfd, 1, 0);
155		15046x	}
156	2/32 ✗ Branch 0 not taken. ✓ Branch 1 taken 5837 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✓ Branch 5 taken 1686 times. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken. ✗ Branch 16 not taken. ✗ Branch 17 not taken. ✗ Branch 18 not taken. ✗ Branch 19 not taken. ✗ Branch 20 not taken. ✗ Branch 21 not taken. ✗ Branch 22 not taken. ✗ Branch 23 not taken. ✗ Branch 24 not taken. ✗ Branch 25 not taken. ✗ Branch 26 not taken. ✗ Branch 27 not taken. ✗ Branch 28 not taken. ✗ Branch 29 not taken. ✗ Branch 30 not taken. ✗ Branch 31 not taken.	7523x	while (r < 0 && errno == EINTR);
157
158	2/16 ✗ Branch 0 not taken. ✓ Branch 1 taken 5837 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 1686 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	7523x	if (r == 0)
159			{
160		✗	this->complete(EAGAIN, 0);
161		✗	return;
162			}
163	2/16 ✗ Branch 0 not taken. ✓ Branch 1 taken 5837 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 1686 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	7523x	if (r < 0)
164			{
165			// EAGAIN must not escape: it is the stay-parked sentinel.
166		✗	this->complete(
167		✗	(errno == EAGAIN \|\| errno == EWOULDBLOCK) ? ENOMEM
168		✗	: errno,
169			0);
170		✗	return;
171			}
172
173		7523x	int err = 0;
174		7523x	socklen_t len = sizeof(err);
175	4/32 ✓ Branch 0 taken 5837 times. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✓ Branch 3 taken 5837 times. ✓ Branch 4 taken 1686 times. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✓ Branch 7 taken 1686 times. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken. ✗ Branch 16 not taken. ✗ Branch 17 not taken. ✗ Branch 18 not taken. ✗ Branch 19 not taken. ✗ Branch 20 not taken. ✗ Branch 21 not taken. ✗ Branch 22 not taken. ✗ Branch 23 not taken. ✗ Branch 24 not taken. ✗ Branch 25 not taken. ✗ Branch 26 not taken. ✗ Branch 27 not taken. ✗ Branch 28 not taken. ✗ Branch 29 not taken. ✗ Branch 30 not taken. ✗ Branch 31 not taken.	7523x	if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
176		✗	err = errno;
177		7523x	this->complete(err, 0);
178		7523x	}
179			};
180
181			/** Readiness-only wait operation.
182
183			Completion is decided by probing the descriptor with a
184			zero-timeout `poll()`, never by the reactor's cached edge
185			events: a speculative read can drain the socket without
186			touching the reactor (stale edge), and a short read can
187			consume the edge while data remains buffered (missing edge).
188			`perform_io()` runs the probe and reports `EAGAIN` when the
189			condition does not currently hold, which keeps the op parked.
190
191			@tparam Base The backend's base op type.
192			*/
193			template<class Base>
194		77718x	struct reactor_wait_op : Base
195			{
196			/// Which event bit this wait targets (reactor_event_read/write/error).
197		77718x	std::uint32_t wait_event = 0;
198
199		120x	void reset() noexcept
200			{
201		120x	Base::reset();
202		120x	wait_event = 0;
203		120x	}
204
205		✗	bool is_read_operation() const noexcept override
206			{
207		✗	return wait_event == reactor_event_read;
208			}
209
210			/** Check whether the waited-for condition currently holds.
211
212			Zero-timeout `poll()` probe. `POLLERR`/`POLLHUP` count as
213			ready for every wait type: the wait must not park on a
214			socket whose next I/O would fail immediately. The probe is
215			side-effect free — in particular it never reads `SO_ERROR`,
216			which is consume-on-read and belongs to whichever operation
217			observes the failure next.
218
219			@param fd The descriptor to probe.
220			@param event The event bit to probe for (read/write/error).
221			@param err Receives the probe failure, if any.
222
223			@return `true` if the condition holds or the probe failed.
224			*/
225		220x	static bool probe(int fd, std::uint32_t event, int& err) noexcept
226			{
227			// poll() silently ignores negative fds; without this guard a
228			// wait on a never-opened or closed socket parks forever.
229	4/16 ✓ Branch 0 taken 108 times. ✓ Branch 1 taken 8 times. ✓ Branch 2 taken 96 times. ✓ Branch 3 taken 8 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	220x	if (fd < 0)
230			{
231		16x	err = EBADF;
232		16x	return true;
233			}
234
235		204x	pollfd pfd{};
236		204x	pfd.fd = fd;
237	4/16 ✓ Branch 0 taken 67 times. ✓ Branch 1 taken 41 times. ✓ Branch 2 taken 55 times. ✓ Branch 3 taken 41 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	204x	if (event == reactor_event_read)
238		122x	pfd.events = POLLIN;
239	4/16 ✓ Branch 0 taken 15 times. ✓ Branch 1 taken 26 times. ✓ Branch 2 taken 15 times. ✓ Branch 3 taken 26 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	82x	else if (event == reactor_event_write)
240		30x	pfd.events = POLLOUT;
241			else
242		52x	pfd.events = POLLPRI;
243
244			int r;
245		204x	do
246			{
247	2/16 ✓ Branch 0 taken 108 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 96 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	204x	r = ::poll(&pfd, 1, 0);
248		408x	}
249	2/32 ✗ Branch 0 not taken. ✓ Branch 1 taken 108 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✓ Branch 5 taken 96 times. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken. ✗ Branch 16 not taken. ✗ Branch 17 not taken. ✗ Branch 18 not taken. ✗ Branch 19 not taken. ✗ Branch 20 not taken. ✗ Branch 21 not taken. ✗ Branch 22 not taken. ✗ Branch 23 not taken. ✗ Branch 24 not taken. ✗ Branch 25 not taken. ✗ Branch 26 not taken. ✗ Branch 27 not taken. ✗ Branch 28 not taken. ✗ Branch 29 not taken. ✗ Branch 30 not taken. ✗ Branch 31 not taken.	204x	while (r < 0 && errno == EINTR);
250
251	2/16 ✗ Branch 0 not taken. ✓ Branch 1 taken 108 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 96 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	204x	if (r < 0)
252			{
253			// Complete with the probe failure rather than park forever.
254			// EAGAIN must not escape here: callers treat it as the
255			// stay-parked sentinel, and poll() can fail with it on
256			// BSD/macOS under transient resource pressure.
257		✗	err = (errno == EAGAIN \|\| errno == EWOULDBLOCK)
258			? ENOMEM
259		✗	: errno;
260		✗	return true;
261			}
262		204x	return r != 0;
263		220x	}
264
265		85x	void perform_io() noexcept override
266			{
267		85x	int err = 0;
268	4/16 ✓ Branch 0 taken 12 times. ✓ Branch 1 taken 34 times. ✓ Branch 2 taken 10 times. ✓ Branch 3 taken 29 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	85x	if (probe(this->fd, wait_event, err))
269		22x	this->complete(err, 0);
270			else
271		63x	this->complete(EAGAIN, 0);
272		85x	}
273			};
274
275			/** Shared scatter-read operation.
276
277			Uses readv() with an EINTR retry loop.
278
279			@tparam Base The backend's base op type.
280			*/
281			template<class Base>
282		23370x	struct reactor_read_op : Base
283			{
284			/// Maximum scatter-gather buffer count.
285			static constexpr std::size_t max_buffers = 16;
286
287			/// Scatter-gather I/O vectors.
288			iovec iovecs[max_buffers];
289
290			/// Number of active I/O vectors.
291		23370x	int iovec_count = 0;
292
293			/// True for zero-length reads (completed immediately).
294		23370x	bool empty_buffer_read = false;
295
296			/// Return true (this is a read-direction operation).
297		219981x	bool is_read_operation() const noexcept override
298			{
299		219981x	return !empty_buffer_read;
300			}
301
302		760889x	void reset() noexcept
303			{
304		760889x	Base::reset();
305		760889x	iovec_count = 0;
306		760889x	empty_buffer_read = false;
307		760889x	}
308
309		88520x	void perform_io() noexcept override
310			{
311			ssize_t n;
312		88520x	do
313			{
314	2/8 ✓ Branch 0 taken 49814 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 38706 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	88520x	n = ::readv(this->fd, iovecs, iovec_count);
315		177040x	}
316	6/16 ✓ Branch 0 taken 855 times. ✓ Branch 1 taken 48959 times. ✓ Branch 2 taken 855 times. ✗ Branch 3 not taken. ✓ Branch 4 taken 749 times. ✓ Branch 5 taken 37957 times. ✓ Branch 6 taken 749 times. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	88520x	while (n < 0 && errno == EINTR);
317
318	4/8 ✓ Branch 0 taken 855 times. ✓ Branch 1 taken 48959 times. ✓ Branch 2 taken 749 times. ✓ Branch 3 taken 37957 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	88520x	if (n >= 0)
319		86916x	this->complete(0, static_cast<std::size_t>(n));
320			else
321	2/8 ✓ Branch 0 taken 855 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 749 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	1604x	this->complete(errno, 0);
322		88520x	}
323			};
324
325			/** Shared gather-write operation.
326
327			Delegates the actual syscall to WritePolicy::write(fd, iovecs, count),
328			which returns ssize_t (bytes written or -1 with errno set).
329
330			@tparam Base The backend's base op type.
331			@tparam WritePolicy Provides `static ssize_t write(int, iovec*, int)`.
332			*/
333			template<class Base, class WritePolicy>
334		23370x	struct reactor_write_op : Base
335			{
336			/// The write syscall policy type.
337			using write_policy = WritePolicy;
338
339			/// Maximum scatter-gather buffer count.
340			static constexpr std::size_t max_buffers = 16;
341
342			/// Scatter-gather I/O vectors.
343			iovec iovecs[max_buffers];
344
345			/// Number of active I/O vectors.
346		23370x	int iovec_count = 0;
347
348		884823x	void reset() noexcept
349			{
350		884823x	Base::reset();
351		884823x	iovec_count = 0;
352		884823x	}
353
354		1019x	void perform_io() noexcept override
355			{
356		1019x	ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
357	3/8 ✓ Branch 0 taken 508 times. ✓ Branch 1 taken 3 times. ✓ Branch 2 taken 508 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	1019x	if (n >= 0)
358		1016x	this->complete(0, static_cast<std::size_t>(n));
359			else
360	1/8 ✓ Branch 0 taken 3 times. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	3x	this->complete(errno, 0);
361		1019x	}
362			};
363
364			/** Shared accept operation.
365
366			Delegates the actual syscall to AcceptPolicy::do_accept(fd, peer_storage),
367			which returns the accepted fd or -1 with errno set.
368
369			@tparam Base The backend's base op type.
370			@tparam AcceptPolicy Provides `static int do_accept(int, sockaddr_storage&)`.
371			*/
372			template<class Base, class AcceptPolicy>
373		2077x	struct reactor_accept_op : Base
374			{
375			/// File descriptor of the accepted connection.
376		2077x	int accepted_fd = -1;
377
378			/// Pointer to the peer socket implementation.
379		2077x	io_object::implementation* peer_impl = nullptr;
380
381			/// Output pointer for the accepted implementation.
382		2077x	io_object::implementation** impl_out = nullptr;
383
384			/// Peer address storage filled by accept.
385		2077x	sockaddr_storage peer_storage{};
386
387			/// Peer address length returned by accept.
388		2077x	socklen_t peer_addrlen = 0;
389
390		7597x	void reset() noexcept
391			{
392		7597x	Base::reset();
393		7597x	accepted_fd = -1;
394		7597x	peer_impl = nullptr;
395		7597x	impl_out = nullptr;
396		7597x	peer_storage = {};
397		7597x	peer_addrlen = 0;
398		7597x	}
399
400		7528x	void perform_io() noexcept override
401			{
402		7528x	int new_fd = AcceptPolicy::do_accept(
403		7528x	this->fd, peer_storage, peer_addrlen);
404	3/8 ✓ Branch 0 taken 5844 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 1683 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✓ Branch 7 taken 1 time.	7528x	if (new_fd >= 0)
405			{
406		7527x	accepted_fd = new_fd;
407		7527x	this->complete(0, 0);
408		7527x	}
409			else
410			{
411	1/8 ✗ Branch 0 not taken. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✓ Branch 6 taken 1 time. ✗ Branch 7 not taken.	1x	this->complete(errno, 0);
412			}
413		7528x	}
414			};
415
416			/** Shared connected send operation for datagram sockets.
417
418			Uses sendmsg() with msg_name=nullptr (connected mode).
419
420			@tparam Base The backend's base op type.
421			*/
422			template<class Base>
423		459x	struct reactor_send_op : Base
424			{
425			/// Maximum scatter-gather buffer count.
426			static constexpr std::size_t max_buffers = 16;
427
428			/// Scatter-gather I/O vectors.
429			iovec iovecs[max_buffers];
430
431			/// Number of active I/O vectors.
432		459x	int iovec_count = 0;
433
434			/// User-supplied message flags.
435		459x	int msg_flags = 0;
436
437		45x	void reset() noexcept
438			{
439		45x	Base::reset();
440		45x	iovec_count = 0;
441		45x	msg_flags = 0;
442		45x	}
443
444		✗	void perform_io() noexcept override
445			{
446		✗	msghdr msg{};
447		✗	msg.msg_iov = iovecs;
448		✗	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
449
450			#ifdef MSG_NOSIGNAL
451		✗	int send_flags = msg_flags \| MSG_NOSIGNAL;
452			#else
453			int send_flags = msg_flags;
454			#endif
455
456			ssize_t n;
457		✗	do
458			{
459		✗	n = ::sendmsg(this->fd, &msg, send_flags);
460		✗	}
461		✗	while (n < 0 && errno == EINTR);
462
463		✗	if (n >= 0)
464		✗	this->complete(0, static_cast<std::size_t>(n));
465			else
466		✗	this->complete(errno, 0);
467		✗	}
468			};
469
470			/** Shared connected recv operation for datagram sockets.
471
472			Uses recvmsg() with msg_name=nullptr (connected mode).
473			Unlike reactor_read_op, does not map n==0 to EOF
474			(zero-length datagrams are valid).
475
476			@tparam Base The backend's base op type.
477			*/
478			template<class Base>
479		459x	struct reactor_recv_op : Base
480			{
481			/// Maximum scatter-gather buffer count.
482			static constexpr std::size_t max_buffers = 16;
483
484			/// Scatter-gather I/O vectors.
485			iovec iovecs[max_buffers];
486
487			/// Number of active I/O vectors.
488		459x	int iovec_count = 0;
489
490			/// User-supplied message flags.
491		459x	int msg_flags = 0;
492
493			/// Return true (this is a read-direction operation).
494		✗	bool is_read_operation() const noexcept override
495			{
496		✗	return true;
497			}
498
499		47x	void reset() noexcept
500			{
501		47x	Base::reset();
502		47x	iovec_count = 0;
503		47x	msg_flags = 0;
504		47x	}
505
506		7x	void perform_io() noexcept override
507			{
508		7x	msghdr msg{};
509		7x	msg.msg_iov = iovecs;
510		7x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
511
512			ssize_t n;
513		7x	do
514			{
515	2/8 ✓ Branch 0 taken 3 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 4 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	7x	n = ::recvmsg(this->fd, &msg, msg_flags);
516		14x	}
517	2/16 ✗ Branch 0 not taken. ✓ Branch 1 taken 3 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✓ Branch 5 taken 4 times. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✗ Branch 9 not taken. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✗ Branch 13 not taken. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	7x	while (n < 0 && errno == EINTR);
518
519	2/8 ✗ Branch 0 not taken. ✓ Branch 1 taken 3 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 4 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	7x	if (n >= 0)
520		7x	this->complete(0, static_cast<std::size_t>(n));
521			else
522		✗	this->complete(errno, 0);
523		7x	}
524			};
525
526			/** Shared send_to operation for datagram sockets.
527
528			Uses sendmsg() with the destination endpoint in msg_name.
529
530			@tparam Base The backend's base op type.
531			*/
532			template<class Base>
533		459x	struct reactor_send_to_op : Base
534			{
535			/// Maximum scatter-gather buffer count.
536			static constexpr std::size_t max_buffers = 16;
537
538			/// Scatter-gather I/O vectors.
539			iovec iovecs[max_buffers];
540
541			/// Number of active I/O vectors.
542		459x	int iovec_count = 0;
543
544			/// Destination address storage.
545		459x	sockaddr_storage dest_storage{};
546
547			/// Destination address length.
548		459x	socklen_t dest_len = 0;
549
550			/// User-supplied message flags.
551		459x	int msg_flags = 0;
552
553		81x	void reset() noexcept
554			{
555		81x	Base::reset();
556		81x	iovec_count = 0;
557		81x	dest_storage = {};
558		81x	dest_len = 0;
559		81x	msg_flags = 0;
560		81x	}
561
562		✗	void perform_io() noexcept override
563			{
564		✗	msghdr msg{};
565		✗	msg.msg_name = &dest_storage;
566		✗	msg.msg_namelen = dest_len;
567		✗	msg.msg_iov = iovecs;
568		✗	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
569
570			#ifdef MSG_NOSIGNAL
571		✗	int send_flags = msg_flags \| MSG_NOSIGNAL;
572			#else
573			int send_flags = msg_flags;
574			#endif
575
576			ssize_t n;
577		✗	do
578			{
579		✗	n = ::sendmsg(this->fd, &msg, send_flags);
580		✗	}
581		✗	while (n < 0 && errno == EINTR);
582
583		✗	if (n >= 0)
584		✗	this->complete(0, static_cast<std::size_t>(n));
585			else
586		✗	this->complete(errno, 0);
587		✗	}
588			};
589
590			/** Shared recv_from operation for datagram sockets.
591
592			Uses recvmsg() with msg_name to capture the source endpoint.
593
594			@tparam Base The backend's base op type.
595			@tparam Endpoint The endpoint type (endpoint or local_endpoint).
596			*/
597			template<class Base, class Endpoint = endpoint>
598		459x	struct reactor_recv_from_op : Base
599			{
600			/// Maximum scatter-gather buffer count.
601			static constexpr std::size_t max_buffers = 16;
602
603			/// Scatter-gather I/O vectors.
604			iovec iovecs[max_buffers];
605
606			/// Number of active I/O vectors.
607		459x	int iovec_count = 0;
608
609			/// Source address storage filled by recvmsg.
610		459x	sockaddr_storage source_storage{};
611
612			/// Actual source address length returned by recvmsg.
613		459x	socklen_t source_addrlen = 0;
614
615			/// Output pointer for the source endpoint (set by do_recv_from).
616		459x	Endpoint* source_out = nullptr;
617
618			/// User-supplied message flags.
619		459x	int msg_flags = 0;
620
621			/// Return true (this is a read-direction operation).
622		✗	bool is_read_operation() const noexcept override
623			{
624		✗	return true;
625			}
626
627		95x	void reset() noexcept
628			{
629		95x	Base::reset();
630		95x	iovec_count = 0;
631		95x	source_storage = {};
632		95x	source_addrlen = 0;
633		95x	source_out = nullptr;
634		95x	msg_flags = 0;
635		95x	}
636
637		26x	void perform_io() noexcept override
638			{
639		26x	msghdr msg{};
640		26x	msg.msg_name = &source_storage;
641		26x	msg.msg_namelen = sizeof(source_storage);
642		26x	msg.msg_iov = iovecs;
643		26x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
644
645			ssize_t n;
646		26x	do
647			{
648	4/8 ✓ Branch 0 taken 7 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 16 times. ✗ Branch 3 not taken. ✓ Branch 4 taken 2 times. ✗ Branch 5 not taken. ✓ Branch 6 taken 1 time. ✗ Branch 7 not taken.	26x	n = ::recvmsg(this->fd, &msg, msg_flags);
649		52x	}
650	4/16 ✗ Branch 0 not taken. ✓ Branch 1 taken 7 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✓ Branch 5 taken 16 times. ✗ Branch 6 not taken. ✗ Branch 7 not taken. ✗ Branch 8 not taken. ✓ Branch 9 taken 2 times. ✗ Branch 10 not taken. ✗ Branch 11 not taken. ✗ Branch 12 not taken. ✓ Branch 13 taken 1 time. ✗ Branch 14 not taken. ✗ Branch 15 not taken.	26x	while (n < 0 && errno == EINTR);
651
652	4/8 ✗ Branch 0 not taken. ✓ Branch 1 taken 7 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 16 times. ✗ Branch 4 not taken. ✓ Branch 5 taken 2 times. ✗ Branch 6 not taken. ✓ Branch 7 taken 1 time.	26x	if (n >= 0)
653			{
654		26x	source_addrlen = msg.msg_namelen;
655		26x	this->complete(0, static_cast<std::size_t>(n));
656		26x	}
657			else
658		✗	this->complete(errno, 0);
659		26x	}
660			};
661
662			} // namespace boost::corosio::detail
663
664			#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
665