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

77.1% Lines (148/192) 87.0% List of functions (94/108) 25.7% Branches (39/152)

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

f(x) Functions (108)

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