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

80.5% Lines (169/210) 85.2% List of functions (218/256) 12.9% Branches (36/280)

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

reactor_op.hpp

f(x) Functions (256)

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_op_base.hpp>
14			#include <boost/corosio/io/io_object.hpp>
15			#include <boost/corosio/endpoint.hpp>
16			#include <boost/capy/ex/executor_ref.hpp>
17
18			#include <atomic>
19			#include <cstddef>
20			#include <optional>
21			#include <stop_token>
22
23			#include <errno.h>
24
25			#include <netinet/in.h>
26			#include <sys/socket.h>
27			#include <sys/uio.h>
28
29			namespace boost::corosio::detail {
30
31			/** Base operation for reactor-based backends.
32
33			Holds per-operation state that depends on the concrete backend
34			socket/acceptor types: coroutine handle, executor, output
35			pointers, file descriptor, stop_callback, and type-specific
36			impl pointers.
37
38			Fields shared across all backends (errn, bytes_transferred,
39			cancelled, impl_ptr, perform_io, complete) live in
40			reactor_op_base so the scheduler and descriptor_state can
41			access them without template instantiation.
42
43			@tparam Socket The backend socket impl type (forward-declared).
44			@tparam Acceptor The backend acceptor impl type (forward-declared).
45			*/
46			template<class Socket, class Acceptor>
47			struct reactor_op : reactor_op_base
48			{
49			// The op envelope — coroutine handle h, cont, executor ex, ec_out,
50			// bytes_out, cancelled, stop_cb (+ its canceller), impl_ptr — lives in
51			// coro_op (via reactor_op_base) and is shared with io_uring/IOCP.
52			// reactor_op adds only the reactor-specific routing state below.
53
54			/// File descriptor this operation targets.
55		134216x	int fd = -1;
56
57			/// Owning socket impl (for stop_token cancellation routing).
58		134216x	Socket* socket_impl_ = nullptr;
59
60			/// Owning acceptor impl (for stop_token cancellation routing).
61		134216x	Acceptor* acceptor_impl_ = nullptr;
62
63		268432x	reactor_op() = default;
64
65			/// Reset operation state for reuse.
66		1240873x	void reset() noexcept
67			{
68		1240873x	fd = -1;
69		1240873x	errn = 0;
70		1240873x	bytes_transferred = 0;
71		1240873x	cancelled.store(false, std::memory_order_relaxed);
72		1240873x	impl_ptr.reset();
73		1240873x	socket_impl_ = nullptr;
74		1240873x	acceptor_impl_ = nullptr;
75		1240873x	}
76
77			/// Return true if this is a read-direction operation.
78		123250x	virtual bool is_read_operation() const noexcept
79			{
80		123250x	return false;
81			}
82
83			/// Cancel this operation via the owning impl.
84			virtual void cancel() noexcept = 0;
85
86			/// coro_op cancellation hook (fired by the shared canceller when the
87			/// stop_token requests cancellation): route to the impl-specific cancel().
88		774x	void on_cancel() noexcept override
89			{
90		774x	cancel();
91		774x	}
92
93			/// Destroy without invoking.
94		12x	void destroy() override
95			{
96		12x	stop_cb.reset();
97		12x	reactor_op_base::destroy();
98		12x	}
99
100			/// Arm the stop-token callback for a socket operation.
101		303929x	void start(std::stop_token const& token, Socket* impl)
102			{
103		303929x	socket_impl_ = impl;
104		303929x	acceptor_impl_ = nullptr;
105		303929x	coro_op::start(token);
106		303929x	}
107
108			/// Arm the stop-token callback for an acceptor operation.
109		6925x	void start(std::stop_token const& token, Acceptor* impl)
110			{
111		6925x	socket_impl_ = nullptr;
112		6925x	acceptor_impl_ = impl;
113		6925x	coro_op::start(token);
114		6925x	}
115			};
116
117			/** Shared connect operation.
118
119			Checks SO_ERROR for connect completion status. The operator()()
120			and cancel() are provided by the concrete backend type.
121
122			@tparam Base The backend's base op type.
123			@tparam Endpoint The endpoint type (endpoint or local_endpoint).
124			*/
125			template<class Base, class Endpoint = endpoint>
126			struct reactor_connect_op : Base
127			{
128			/// Endpoint to connect to.
129			Endpoint target_endpoint;
130
131			/// Reset operation state for reuse.
132		6928x	void reset() noexcept
133			{
134		6928x	Base::reset();
135		6928x	target_endpoint = Endpoint{};
136		6928x	}
137
138		6859x	void perform_io() noexcept override
139			{
140		6859x	int err = 0;
141		6859x	socklen_t len = sizeof(err);
142	4/32 ✓ Branch 0 taken 4862 times. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✓ Branch 3 taken 4862 times. ✓ Branch 4 taken 1997 times. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✓ Branch 7 taken 1997 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.	6859x	if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
143		✗	err = errno;
144		6859x	this->complete(err, 0);
145		6859x	}
146			};
147
148			/** Readiness-only wait operation.
149
150			Does not perform any I/O syscall. Completion is signalled by
151			the reactor delivering the requested edge event; reactor_descriptor_state
152			calls complete() directly and never invokes perform_io().
153
154			@tparam Base The backend's base op type.
155			*/
156			template<class Base>
157		68013x	struct reactor_wait_op : Base
158			{
159			/* Mirror of reactor_event_read from reactor_descriptor_state.hpp.
160			Including that header from here would create an include cycle
161			(descriptor_state -> reactor_op_base; reactor_op -> reactor_op_base),
162			so we carry the value locally. Both must stay in sync. */
163			static constexpr std::uint32_t read_event = 0x001;
164
165			/// Which event bit this wait targets (reactor_event_read/write/error).
166		68013x	std::uint32_t wait_event = 0;
167
168		72x	void reset() noexcept
169			{
170		72x	Base::reset();
171		72x	wait_event = 0;
172		72x	}
173
174		✗	bool is_read_operation() const noexcept override
175			{
176		✗	return wait_event == read_event;
177			}
178
179			/* perform_io() should never be called for a wait op — readiness
180			IS the completion. Overridden here to satisfy the virtual and
181			produce a safe result if called defensively. */
182		✗	void perform_io() noexcept override
183			{
184		✗	this->complete(0, 0);
185		✗	}
186			};
187
188			/** Shared scatter-read operation.
189
190			Uses readv() with an EINTR retry loop.
191
192			@tparam Base The backend's base op type.
193			*/
194			template<class Base>
195		20982x	struct reactor_read_op : Base
196			{
197			/// Maximum scatter-gather buffer count.
198			static constexpr std::size_t max_buffers = 16;
199
200			/// Scatter-gather I/O vectors.
201			iovec iovecs[max_buffers];
202
203			/// Number of active I/O vectors.
204		20982x	int iovec_count = 0;
205
206			/// True for zero-length reads (completed immediately).
207		20982x	bool empty_buffer_read = false;
208
209			/// Return true (this is a read-direction operation).
210		173612x	bool is_read_operation() const noexcept override
211			{
212		173612x	return !empty_buffer_read;
213			}
214
215		552098x	void reset() noexcept
216			{
217		552098x	Base::reset();
218		552098x	iovec_count = 0;
219		552098x	empty_buffer_read = false;
220		552098x	}
221
222		84279x	void perform_io() noexcept override
223			{
224			ssize_t n;
225		84279x	do
226			{
227	2/8 ✓ Branch 0 taken 47990 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 36289 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	84279x	n = ::readv(this->fd, iovecs, iovec_count);
228		168558x	}
229	6/16 ✓ Branch 0 taken 501 times. ✓ Branch 1 taken 47489 times. ✓ Branch 2 taken 501 times. ✗ Branch 3 not taken. ✓ Branch 4 taken 1309 times. ✓ Branch 5 taken 34980 times. ✓ Branch 6 taken 1309 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.	84279x	while (n < 0 && errno == EINTR);
230
231	4/8 ✓ Branch 0 taken 501 times. ✓ Branch 1 taken 47489 times. ✓ Branch 2 taken 1309 times. ✓ Branch 3 taken 34980 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	84279x	if (n >= 0)
232		82469x	this->complete(0, static_cast<std::size_t>(n));
233			else
234	2/8 ✓ Branch 0 taken 501 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 1309 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	1810x	this->complete(errno, 0);
235		84279x	}
236			};
237
238			/** Shared gather-write operation.
239
240			Delegates the actual syscall to WritePolicy::write(fd, iovecs, count),
241			which returns ssize_t (bytes written or -1 with errno set).
242
243			@tparam Base The backend's base op type.
244			@tparam WritePolicy Provides `static ssize_t write(int, iovec*, int)`.
245			*/
246			template<class Base, class WritePolicy>
247		20982x	struct reactor_write_op : Base
248			{
249			/// The write syscall policy type.
250			using write_policy = WritePolicy;
251
252			/// Maximum scatter-gather buffer count.
253			static constexpr std::size_t max_buffers = 16;
254
255			/// Scatter-gather I/O vectors.
256			iovec iovecs[max_buffers];
257
258			/// Number of active I/O vectors.
259		20982x	int iovec_count = 0;
260
261		674650x	void reset() noexcept
262			{
263		674650x	Base::reset();
264		674650x	iovec_count = 0;
265		674650x	}
266
267		998x	void perform_io() noexcept override
268			{
269		998x	ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
270	3/8 ✓ Branch 0 taken 493 times. ✓ Branch 1 taken 3 times. ✓ Branch 2 taken 502 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	998x	if (n >= 0)
271		995x	this->complete(0, static_cast<std::size_t>(n));
272			else
273	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);
274		998x	}
275			};
276
277			/** Shared accept operation.
278
279			Delegates the actual syscall to AcceptPolicy::do_accept(fd, peer_storage),
280			which returns the accepted fd or -1 with errno set.
281
282			@tparam Base The backend's base op type.
283			@tparam AcceptPolicy Provides `static int do_accept(int, sockaddr_storage&)`.
284			*/
285			template<class Base, class AcceptPolicy>
286		1297x	struct reactor_accept_op : Base
287			{
288			/// File descriptor of the accepted connection.
289		1297x	int accepted_fd = -1;
290
291			/// Pointer to the peer socket implementation.
292		1297x	io_object::implementation* peer_impl = nullptr;
293
294			/// Output pointer for the accepted implementation.
295		1297x	io_object::implementation** impl_out = nullptr;
296
297			/// Peer address storage filled by accept.
298		1297x	sockaddr_storage peer_storage{};
299
300			/// Peer address length returned by accept.
301		1297x	socklen_t peer_addrlen = 0;
302
303		6913x	void reset() noexcept
304			{
305		6913x	Base::reset();
306		6913x	accepted_fd = -1;
307		6913x	peer_impl = nullptr;
308		6913x	impl_out = nullptr;
309		6913x	peer_storage = {};
310		6913x	peer_addrlen = 0;
311		6913x	}
312
313		6860x	void perform_io() noexcept override
314			{
315		6860x	int new_fd = AcceptPolicy::do_accept(
316		6860x	this->fd, peer_storage, peer_addrlen);
317	2/8 ✓ Branch 0 taken 4867 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 1993 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	6860x	if (new_fd >= 0)
318			{
319		6860x	accepted_fd = new_fd;
320		6860x	this->complete(0, 0);
321		6860x	}
322			else
323			{
324		✗	this->complete(errno, 0);
325			}
326		6860x	}
327			};
328
329			/** Shared connected send operation for datagram sockets.
330
331			Uses sendmsg() with msg_name=nullptr (connected mode).
332
333			@tparam Base The backend's base op type.
334			*/
335			template<class Base>
336		392x	struct reactor_send_op : Base
337			{
338			/// Maximum scatter-gather buffer count.
339			static constexpr std::size_t max_buffers = 16;
340
341			/// Scatter-gather I/O vectors.
342			iovec iovecs[max_buffers];
343
344			/// Number of active I/O vectors.
345		392x	int iovec_count = 0;
346
347			/// User-supplied message flags.
348		392x	int msg_flags = 0;
349
350		44x	void reset() noexcept
351			{
352		44x	Base::reset();
353		44x	iovec_count = 0;
354		44x	msg_flags = 0;
355		44x	}
356
357		✗	void perform_io() noexcept override
358			{
359		✗	msghdr msg{};
360		✗	msg.msg_iov = iovecs;
361		✗	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
362
363			#ifdef MSG_NOSIGNAL
364		✗	int send_flags = msg_flags \| MSG_NOSIGNAL;
365			#else
366			int send_flags = msg_flags;
367			#endif
368
369			ssize_t n;
370		✗	do
371			{
372		✗	n = ::sendmsg(this->fd, &msg, send_flags);
373		✗	}
374		✗	while (n < 0 && errno == EINTR);
375
376		✗	if (n >= 0)
377		✗	this->complete(0, static_cast<std::size_t>(n));
378			else
379		✗	this->complete(errno, 0);
380		✗	}
381			};
382
383			/** Shared connected recv operation for datagram sockets.
384
385			Uses recvmsg() with msg_name=nullptr (connected mode).
386			Unlike reactor_read_op, does not map n==0 to EOF
387			(zero-length datagrams are valid).
388
389			@tparam Base The backend's base op type.
390			*/
391			template<class Base>
392		392x	struct reactor_recv_op : Base
393			{
394			/// Maximum scatter-gather buffer count.
395			static constexpr std::size_t max_buffers = 16;
396
397			/// Scatter-gather I/O vectors.
398			iovec iovecs[max_buffers];
399
400			/// Number of active I/O vectors.
401		392x	int iovec_count = 0;
402
403			/// User-supplied message flags.
404		392x	int msg_flags = 0;
405
406			/// Return true (this is a read-direction operation).
407		✗	bool is_read_operation() const noexcept override
408			{
409		✗	return true;
410			}
411
412		42x	void reset() noexcept
413			{
414		42x	Base::reset();
415		42x	iovec_count = 0;
416		42x	msg_flags = 0;
417		42x	}
418
419		7x	void perform_io() noexcept override
420			{
421		7x	msghdr msg{};
422		7x	msg.msg_iov = iovecs;
423		7x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
424
425			ssize_t n;
426		7x	do
427			{
428	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);
429		14x	}
430	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);
431
432	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)
433		7x	this->complete(0, static_cast<std::size_t>(n));
434			else
435		✗	this->complete(errno, 0);
436		7x	}
437			};
438
439			/** Shared send_to operation for datagram sockets.
440
441			Uses sendmsg() with the destination endpoint in msg_name.
442
443			@tparam Base The backend's base op type.
444			*/
445			template<class Base>
446		392x	struct reactor_send_to_op : Base
447			{
448			/// Maximum scatter-gather buffer count.
449			static constexpr std::size_t max_buffers = 16;
450
451			/// Scatter-gather I/O vectors.
452			iovec iovecs[max_buffers];
453
454			/// Number of active I/O vectors.
455		392x	int iovec_count = 0;
456
457			/// Destination address storage.
458		392x	sockaddr_storage dest_storage{};
459
460			/// Destination address length.
461		392x	socklen_t dest_len = 0;
462
463			/// User-supplied message flags.
464		392x	int msg_flags = 0;
465
466		58x	void reset() noexcept
467			{
468		58x	Base::reset();
469		58x	iovec_count = 0;
470		58x	dest_storage = {};
471		58x	dest_len = 0;
472		58x	msg_flags = 0;
473		58x	}
474
475		✗	void perform_io() noexcept override
476			{
477		✗	msghdr msg{};
478		✗	msg.msg_name = &dest_storage;
479		✗	msg.msg_namelen = dest_len;
480		✗	msg.msg_iov = iovecs;
481		✗	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
482
483			#ifdef MSG_NOSIGNAL
484		✗	int send_flags = msg_flags \| MSG_NOSIGNAL;
485			#else
486			int send_flags = msg_flags;
487			#endif
488
489			ssize_t n;
490		✗	do
491			{
492		✗	n = ::sendmsg(this->fd, &msg, send_flags);
493		✗	}
494		✗	while (n < 0 && errno == EINTR);
495
496		✗	if (n >= 0)
497		✗	this->complete(0, static_cast<std::size_t>(n));
498			else
499		✗	this->complete(errno, 0);
500		✗	}
501			};
502
503			/** Shared recv_from operation for datagram sockets.
504
505			Uses recvmsg() with msg_name to capture the source endpoint.
506
507			@tparam Base The backend's base op type.
508			@tparam Endpoint The endpoint type (endpoint or local_endpoint).
509			*/
510			template<class Base, class Endpoint = endpoint>
511		392x	struct reactor_recv_from_op : Base
512			{
513			/// Maximum scatter-gather buffer count.
514			static constexpr std::size_t max_buffers = 16;
515
516			/// Scatter-gather I/O vectors.
517			iovec iovecs[max_buffers];
518
519			/// Number of active I/O vectors.
520		392x	int iovec_count = 0;
521
522			/// Source address storage filled by recvmsg.
523		392x	sockaddr_storage source_storage{};
524
525			/// Actual source address length returned by recvmsg.
526		392x	socklen_t source_addrlen = 0;
527
528			/// Output pointer for the source endpoint (set by do_recv_from).
529		392x	Endpoint* source_out = nullptr;
530
531			/// User-supplied message flags.
532		392x	int msg_flags = 0;
533
534			/// Return true (this is a read-direction operation).
535		✗	bool is_read_operation() const noexcept override
536			{
537		✗	return true;
538			}
539
540		68x	void reset() noexcept
541			{
542		68x	Base::reset();
543		68x	iovec_count = 0;
544		68x	source_storage = {};
545		68x	source_addrlen = 0;
546		68x	source_out = nullptr;
547		68x	msg_flags = 0;
548		68x	}
549
550		16x	void perform_io() noexcept override
551			{
552		16x	msghdr msg{};
553		16x	msg.msg_name = &source_storage;
554		16x	msg.msg_namelen = sizeof(source_storage);
555		16x	msg.msg_iov = iovecs;
556		16x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
557
558			ssize_t n;
559		16x	do
560			{
561	2/8 ✓ Branch 0 taken 4 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 12 times. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	16x	n = ::recvmsg(this->fd, &msg, msg_flags);
562		32x	}
563	2/16 ✗ Branch 0 not taken. ✓ Branch 1 taken 4 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken. ✗ Branch 4 not taken. ✓ Branch 5 taken 12 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.	16x	while (n < 0 && errno == EINTR);
564
565	2/8 ✗ Branch 0 not taken. ✓ Branch 1 taken 4 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 12 times. ✗ Branch 4 not taken. ✗ Branch 5 not taken. ✗ Branch 6 not taken. ✗ Branch 7 not taken.	16x	if (n >= 0)
566			{
567		16x	source_addrlen = msg.msg_namelen;
568		16x	this->complete(0, static_cast<std::size_t>(n));
569		16x	}
570			else
571		✗	this->complete(errno, 0);
572		16x	}
573			};
574
575			} // namespace boost::corosio::detail
576
577			#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
578