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

93.3% Lines (154/165) 78.8% List of functions (126/160)

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

reactor_op.hpp

f(x) Functions (160)

Function Calls Lines Blocks

Line	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		int fd = -1;
56
57		/// Owning socket impl (for stop_token cancellation routing).
58		Socket* socket_impl_ = nullptr;
59
60		/// Owning acceptor impl (for stop_token cancellation routing).
61		Acceptor* acceptor_impl_ = nullptr;
62
63	147408x	reactor_op() = default;
64
65		/// Reset operation state for reuse.
66	578204x	void reset() noexcept
67		{
68	578204x	fd = -1;
69	578204x	errn = 0;
70	578204x	bytes_transferred = 0;
71	578204x	cancelled.store(false, std::memory_order_relaxed);
72	578204x	impl_ptr.reset();
73	578204x	socket_impl_ = nullptr;
74	578204x	acceptor_impl_ = nullptr;
75	578204x	}
76
77		/// Return true if this is a read-direction operation.
78	64414x	virtual bool is_read_operation() const noexcept
79		{
80	64414x	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	931x	void on_cancel() noexcept override
89		{
90	931x	cancel();
91	931x	}
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	124289x	void start(std::stop_token const& token, Socket* impl)
102		{
103	124289x	socket_impl_ = impl;
104	124289x	acceptor_impl_ = nullptr;
105	124289x	coro_op::start(token);
106	124289x	}
107
108		/// Arm the stop-token callback for an acceptor operation.
109	7641x	void start(std::stop_token const& token, Acceptor* impl)
110		{
111	7641x	socket_impl_ = nullptr;
112	7641x	acceptor_impl_ = impl;
113	7641x	coro_op::start(token);
114	7641x	}
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	7644x	void reset() noexcept
133		{
134	7644x	Base::reset();
135	7644x	target_endpoint = Endpoint{};
136	7644x	}
137
138	7580x	void perform_io() noexcept override
139		{
140	7580x	int err = 0;
141	7580x	socklen_t len = sizeof(err);
142	7580x	if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
143	✗	err = errno;
144	7580x	this->complete(err, 0);
145	7580x	}
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		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		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		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		int iovec_count = 0;
205
206		/// True for zero-length reads (completed immediately).
207		bool empty_buffer_read = false;
208
209		/// Return true (this is a read-direction operation).
210	52039x	bool is_read_operation() const noexcept override
211		{
212	52039x	return !empty_buffer_read;
213		}
214
215	245833x	void reset() noexcept
216		{
217	245833x	Base::reset();
218	245833x	iovec_count = 0;
219	245833x	empty_buffer_read = false;
220	245833x	}
221
222	2294x	void perform_io() noexcept override
223		{
224		ssize_t n;
225		do
226		{
227	2294x	n = ::readv(this->fd, iovecs, iovec_count);
228		}
229	2294x	while (n < 0 && errno == EINTR);
230
231	2294x	if (n >= 0)
232	2022x	this->complete(0, static_cast<std::size_t>(n));
233		else
234	272x	this->complete(errno, 0);
235	2294x	}
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		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		int iovec_count = 0;
260
261	316560x	void reset() noexcept
262		{
263	316560x	Base::reset();
264	316560x	iovec_count = 0;
265	316560x	}
266
267	132x	void perform_io() noexcept override
268		{
269	132x	ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
270	132x	if (n >= 0)
271	129x	this->complete(0, static_cast<std::size_t>(n));
272		else
273	3x	this->complete(errno, 0);
274	132x	}
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		struct reactor_accept_op : Base
287		{
288		/// File descriptor of the accepted connection.
289		int accepted_fd = -1;
290
291		/// Pointer to the peer socket implementation.
292		io_object::implementation* peer_impl = nullptr;
293
294		/// Output pointer for the accepted implementation.
295		io_object::implementation** impl_out = nullptr;
296
297		/// Peer address storage filled by accept.
298		sockaddr_storage peer_storage{};
299
300		/// Peer address length returned by accept.
301		socklen_t peer_addrlen = 0;
302
303	7629x	void reset() noexcept
304		{
305	7629x	Base::reset();
306	7629x	accepted_fd = -1;
307	7629x	peer_impl = nullptr;
308	7629x	impl_out = nullptr;
309	7629x	peer_storage = {};
310	7629x	peer_addrlen = 0;
311	7629x	}
312
313	7568x	void perform_io() noexcept override
314		{
315	7568x	int new_fd = AcceptPolicy::do_accept(
316	7568x	this->fd, peer_storage, peer_addrlen);
317	7568x	if (new_fd >= 0)
318		{
319	7568x	accepted_fd = new_fd;
320	7568x	this->complete(0, 0);
321		}
322		else
323		{
324	✗	this->complete(errno, 0);
325		}
326	7568x	}
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		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		int iovec_count = 0;
346
347		/// User-supplied message flags.
348		int msg_flags = 0;
349
350	106x	void reset() noexcept
351		{
352	106x	Base::reset();
353	106x	iovec_count = 0;
354	106x	msg_flags = 0;
355	106x	}
356
357	32x	void perform_io() noexcept override
358		{
359	32x	msghdr msg{};
360	32x	msg.msg_iov = iovecs;
361	32x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
362
363		#ifdef MSG_NOSIGNAL
364	32x	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	32x	n = ::sendmsg(this->fd, &msg, send_flags);
373		}
374	32x	while (n < 0 && errno == EINTR);
375
376	32x	if (n >= 0)
377	30x	this->complete(0, static_cast<std::size_t>(n));
378		else
379	2x	this->complete(errno, 0);
380	32x	}
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		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		int iovec_count = 0;
402
403		/// User-supplied message flags.
404		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	104x	void reset() noexcept
413		{
414	104x	Base::reset();
415	104x	iovec_count = 0;
416	104x	msg_flags = 0;
417	104x	}
418
419	37x	void perform_io() noexcept override
420		{
421	37x	msghdr msg{};
422	37x	msg.msg_iov = iovecs;
423	37x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
424
425		ssize_t n;
426		do
427		{
428	37x	n = ::recvmsg(this->fd, &msg, msg_flags);
429		}
430	37x	while (n < 0 && errno == EINTR);
431
432	37x	if (n >= 0)
433	34x	this->complete(0, static_cast<std::size_t>(n));
434		else
435	3x	this->complete(errno, 0);
436	37x	}
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		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		int iovec_count = 0;
456
457		/// Destination address storage.
458		sockaddr_storage dest_storage{};
459
460		/// Destination address length.
461		socklen_t dest_len = 0;
462
463		/// User-supplied message flags.
464		int msg_flags = 0;
465
466	122x	void reset() noexcept
467		{
468	122x	Base::reset();
469	122x	iovec_count = 0;
470	122x	dest_storage = {};
471	122x	dest_len = 0;
472	122x	msg_flags = 0;
473	122x	}
474
475	32x	void perform_io() noexcept override
476		{
477	32x	msghdr msg{};
478	32x	msg.msg_name = &dest_storage;
479	32x	msg.msg_namelen = dest_len;
480	32x	msg.msg_iov = iovecs;
481	32x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
482
483		#ifdef MSG_NOSIGNAL
484	32x	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	32x	n = ::sendmsg(this->fd, &msg, send_flags);
493		}
494	32x	while (n < 0 && errno == EINTR);
495
496	32x	if (n >= 0)
497	30x	this->complete(0, static_cast<std::size_t>(n));
498		else
499	2x	this->complete(errno, 0);
500	32x	}
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		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		int iovec_count = 0;
521
522		/// Source address storage filled by recvmsg.
523		sockaddr_storage source_storage{};
524
525		/// Actual source address length returned by recvmsg.
526		socklen_t source_addrlen = 0;
527
528		/// Output pointer for the source endpoint (set by do_recv_from).
529		Endpoint* source_out = nullptr;
530
531		/// User-supplied message flags.
532		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	134x	void reset() noexcept
541		{
542	134x	Base::reset();
543	134x	iovec_count = 0;
544	134x	source_storage = {};
545	134x	source_addrlen = 0;
546	134x	source_out = nullptr;
547	134x	msg_flags = 0;
548	134x	}
549
550	39x	void perform_io() noexcept override
551		{
552	39x	msghdr msg{};
553	39x	msg.msg_name = &source_storage;
554	39x	msg.msg_namelen = sizeof(source_storage);
555	39x	msg.msg_iov = iovecs;
556	39x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
557
558		ssize_t n;
559		do
560		{
561	39x	n = ::recvmsg(this->fd, &msg, msg_flags);
562		}
563	39x	while (n < 0 && errno == EINTR);
564
565	39x	if (n >= 0)
566		{
567	36x	source_addrlen = msg.msg_namelen;
568	36x	this->complete(0, static_cast<std::size_t>(n));
569		}
570		else
571	3x	this->complete(errno, 0);
572	39x	}
573		};
574
575		} // namespace boost::corosio::detail
576
577		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
578