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

92.0% Lines (183/0/199) 84.5% List of functions (142/0/168)

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

reactor_op.hpp

f(x) Functions (168)

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_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		int fd = -1;
58
59		/// Owning socket impl (for stop_token cancellation routing).
60		Socket* socket_impl_ = nullptr;
61
62		/// Owning acceptor impl (for stop_token cancellation routing).
63		Acceptor* acceptor_impl_ = nullptr;
64
65	87208x	reactor_op() = default;
66
67		/// Reset operation state for reuse.
68	404422x	void reset() noexcept
69		{
70	404422x	fd = -1;
71	404422x	errn = 0;
72	404422x	bytes_transferred = 0;
73	404422x	cancelled.store(false, std::memory_order_relaxed);
74	404422x	impl_ptr.reset();
75	404422x	socket_impl_ = nullptr;
76	404422x	acceptor_impl_ = nullptr;
77	404422x	}
78
79		/// Return true if this is a read-direction operation.
80	39725x	virtual bool is_read_operation() const noexcept
81		{
82	39725x	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	302x	void on_cancel() noexcept override
91		{
92	302x	cancel();
93	302x	}
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	84512x	void start(std::stop_token const& token, Socket* impl)
104		{
105	84512x	socket_impl_ = impl;
106	84512x	acceptor_impl_ = nullptr;
107	84512x	coro_op::start(token);
108	84512x	}
109
110		/// Arm the stop-token callback for an acceptor operation.
111	4464x	void start(std::stop_token const& token, Acceptor* impl)
112		{
113	4464x	socket_impl_ = nullptr;
114	4464x	acceptor_impl_ = impl;
115	4464x	coro_op::start(token);
116	4464x	}
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	4448x	void reset() noexcept
135		{
136	4448x	Base::reset();
137	4448x	target_endpoint = Endpoint{};
138	4448x	}
139
140	4374x	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	4374x	pollfd pfd{};
149	4374x	pfd.fd = this->fd;
150	4374x	pfd.events = POLLOUT;
151		int r;
152		do
153		{
154	4374x	r = ::poll(&pfd, 1, 0);
155		}
156	4374x	while (r < 0 && errno == EINTR);
157
158	4374x	if (r == 0)
159		{
160	1x	this->complete(EAGAIN, 0);
161	1x	return;
162		}
163	4373x	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	4373x	int err = 0;
174	4373x	socklen_t len = sizeof(err);
175	4373x	if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
176	✗	err = errno;
177	4373x	this->complete(err, 0);
178		}
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		struct reactor_wait_op : Base
195		{
196		/// Which event bit this wait targets (reactor_event_read/write/error).
197		std::uint32_t wait_event = 0;
198
199	102x	void reset() noexcept
200		{
201	102x	Base::reset();
202	102x	wait_event = 0;
203	102x	}
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	202x	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	202x	if (fd < 0)
230		{
231	2x	err = EBADF;
232	2x	return true;
233		}
234
235	200x	pollfd pfd{};
236	200x	pfd.fd = fd;
237	200x	if (event == reactor_event_read)
238	126x	pfd.events = POLLIN;
239	74x	else if (event == reactor_event_write)
240	30x	pfd.events = POLLOUT;
241		else
242	44x	pfd.events = POLLPRI;
243
244		int r;
245		do
246		{
247	200x	r = ::poll(&pfd, 1, 0);
248		}
249	200x	while (r < 0 && errno == EINTR);
250
251	200x	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	200x	return r != 0;
263		}
264
265	87x	void perform_io() noexcept override
266		{
267	87x	int err = 0;
268	87x	if (probe(this->fd, wait_event, err))
269	22x	this->complete(err, 0);
270		else
271	65x	this->complete(EAGAIN, 0);
272	87x	}
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		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		int iovec_count = 0;
292
293		/// True for zero-length reads (completed immediately).
294		bool empty_buffer_read = false;
295
296		/// Return true (this is a read-direction operation).
297	40106x	bool is_read_operation() const noexcept override
298		{
299	40106x	return !empty_buffer_read;
300		}
301
302	197776x	void reset() noexcept
303		{
304	197776x	Base::reset();
305	197776x	iovec_count = 0;
306	197776x	empty_buffer_read = false;
307	197776x	}
308
309	655x	void perform_io() noexcept override
310		{
311		ssize_t n;
312		do
313		{
314	655x	n = ::readv(this->fd, iovecs, iovec_count);
315		}
316	655x	while (n < 0 && errno == EINTR);
317
318	655x	if (n >= 0)
319	418x	this->complete(0, static_cast<std::size_t>(n));
320		else
321	237x	this->complete(errno, 0);
322	655x	}
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		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		int iovec_count = 0;
347
348	197131x	void reset() noexcept
349		{
350	197131x	Base::reset();
351	197131x	iovec_count = 0;
352	197131x	}
353
354	132x	void perform_io() noexcept override
355		{
356	132x	ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
357	132x	if (n >= 0)
358	129x	this->complete(0, static_cast<std::size_t>(n));
359		else
360	3x	this->complete(errno, 0);
361	132x	}
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		struct reactor_accept_op : Base
374		{
375		/// File descriptor of the accepted connection.
376		int accepted_fd = -1;
377
378		/// Pointer to the peer socket implementation.
379		io_object::implementation* peer_impl = nullptr;
380
381		/// Output pointer for the accepted implementation.
382		io_object::implementation** impl_out = nullptr;
383
384		/// Peer address storage filled by accept.
385		sockaddr_storage peer_storage{};
386
387		/// Peer address length returned by accept.
388		socklen_t peer_addrlen = 0;
389
390	4441x	void reset() noexcept
391		{
392	4441x	Base::reset();
393	4441x	accepted_fd = -1;
394	4441x	peer_impl = nullptr;
395	4441x	impl_out = nullptr;
396	4441x	peer_storage = {};
397	4441x	peer_addrlen = 0;
398	4441x	}
399
400	4363x	void perform_io() noexcept override
401		{
402	4363x	int new_fd = AcceptPolicy::do_accept(
403	4363x	this->fd, peer_storage, peer_addrlen);
404	4363x	if (new_fd >= 0)
405		{
406	4363x	accepted_fd = new_fd;
407	4363x	this->complete(0, 0);
408		}
409		else
410		{
411	✗	this->complete(errno, 0);
412		}
413	4363x	}
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		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		int iovec_count = 0;
433
434		/// User-supplied message flags.
435		int msg_flags = 0;
436
437	107x	void reset() noexcept
438		{
439	107x	Base::reset();
440	107x	iovec_count = 0;
441	107x	msg_flags = 0;
442	107x	}
443
444	32x	void perform_io() noexcept override
445		{
446	32x	msghdr msg{};
447	32x	msg.msg_iov = iovecs;
448	32x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
449
450		#ifdef MSG_NOSIGNAL
451	32x	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	32x	n = ::sendmsg(this->fd, &msg, send_flags);
460		}
461	32x	while (n < 0 && errno == EINTR);
462
463	32x	if (n >= 0)
464	30x	this->complete(0, static_cast<std::size_t>(n));
465		else
466	2x	this->complete(errno, 0);
467	32x	}
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		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		int iovec_count = 0;
489
490		/// User-supplied message flags.
491		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	109x	void reset() noexcept
500		{
501	109x	Base::reset();
502	109x	iovec_count = 0;
503	109x	msg_flags = 0;
504	109x	}
505
506	37x	void perform_io() noexcept override
507		{
508	37x	msghdr msg{};
509	37x	msg.msg_iov = iovecs;
510	37x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
511
512		ssize_t n;
513		do
514		{
515	37x	n = ::recvmsg(this->fd, &msg, msg_flags);
516		}
517	37x	while (n < 0 && errno == EINTR);
518
519	37x	if (n >= 0)
520	34x	this->complete(0, static_cast<std::size_t>(n));
521		else
522	3x	this->complete(errno, 0);
523	37x	}
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		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		int iovec_count = 0;
543
544		/// Destination address storage.
545		sockaddr_storage dest_storage{};
546
547		/// Destination address length.
548		socklen_t dest_len = 0;
549
550		/// User-supplied message flags.
551		int msg_flags = 0;
552
553	145x	void reset() noexcept
554		{
555	145x	Base::reset();
556	145x	iovec_count = 0;
557	145x	dest_storage = {};
558	145x	dest_len = 0;
559	145x	msg_flags = 0;
560	145x	}
561
562	32x	void perform_io() noexcept override
563		{
564	32x	msghdr msg{};
565	32x	msg.msg_name = &dest_storage;
566	32x	msg.msg_namelen = dest_len;
567	32x	msg.msg_iov = iovecs;
568	32x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
569
570		#ifdef MSG_NOSIGNAL
571	32x	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	32x	n = ::sendmsg(this->fd, &msg, send_flags);
580		}
581	32x	while (n < 0 && errno == EINTR);
582
583	32x	if (n >= 0)
584	30x	this->complete(0, static_cast<std::size_t>(n));
585		else
586	2x	this->complete(errno, 0);
587	32x	}
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		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		int iovec_count = 0;
608
609		/// Source address storage filled by recvmsg.
610		sockaddr_storage source_storage{};
611
612		/// Actual source address length returned by recvmsg.
613		socklen_t source_addrlen = 0;
614
615		/// Output pointer for the source endpoint (set by do_recv_from).
616		Endpoint* source_out = nullptr;
617
618		/// User-supplied message flags.
619		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	163x	void reset() noexcept
628		{
629	163x	Base::reset();
630	163x	iovec_count = 0;
631	163x	source_storage = {};
632	163x	source_addrlen = 0;
633	163x	source_out = nullptr;
634	163x	msg_flags = 0;
635	163x	}
636
637	44x	void perform_io() noexcept override
638		{
639	44x	msghdr msg{};
640	44x	msg.msg_name = &source_storage;
641	44x	msg.msg_namelen = sizeof(source_storage);
642	44x	msg.msg_iov = iovecs;
643	44x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
644
645		ssize_t n;
646		do
647		{
648	44x	n = ::recvmsg(this->fd, &msg, msg_flags);
649		}
650	44x	while (n < 0 && errno == EINTR);
651
652	44x	if (n >= 0)
653		{
654	41x	source_addrlen = msg.msg_namelen;
655	41x	this->complete(0, static_cast<std::size_t>(n));
656		}
657		else
658	3x	this->complete(errno, 0);
659	44x	}
660		};
661
662		} // namespace boost::corosio::detail
663
664		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
665