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

78.3% Lines (162/207) 78.2% List of functions (169/216) 15.1% Branches (46/304)

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

reactor_op.hpp

f(x) Functions (216)

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