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

97.0% Lines (193/0/199) 86.9% List of functions (146/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	102876x	reactor_op() = default;
66
67		/// Reset operation state for reuse.
68	419801x	void reset() noexcept
69		{
70	419801x	fd = -1;
71	419801x	errn = 0;
72	419801x	bytes_transferred = 0;
73	419801x	cancelled.store(false, std::memory_order_relaxed);
74	419801x	impl_ptr.reset();
75	419801x	socket_impl_ = nullptr;
76	419801x	acceptor_impl_ = nullptr;
77	419801x	}
78
79		/// Return true if this is a read-direction operation.
80	41119x	virtual bool is_read_operation() const noexcept
81		{
82	41119x	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	56x	void destroy() override
97		{
98	56x	stop_cb.reset();
99	56x	reactor_op_base::destroy();
100	56x	}
101
102		/// Arm the stop-token callback for a socket operation.
103	88136x	void start(std::stop_token const& token, Socket* impl)
104		{
105	88136x	socket_impl_ = impl;
106	88136x	acceptor_impl_ = nullptr;
107	88136x	coro_op::start(token);
108	88136x	}
109
110		/// Arm the stop-token callback for an acceptor operation.
111	5242x	void start(std::stop_token const& token, Acceptor* impl)
112		{
113	5242x	socket_impl_ = nullptr;
114	5242x	acceptor_impl_ = impl;
115	5242x	coro_op::start(token);
116	5242x	}
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	5233x	void reset() noexcept
135		{
136	5233x	Base::reset();
137	5233x	target_endpoint = Endpoint{};
138	5233x	}
139
140	5146x	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	5146x	pollfd pfd{};
149	5146x	pfd.fd = this->fd;
150	5146x	pfd.events = POLLOUT;
151		int r;
152		do
153		{
154	5146x	r = ::poll(&pfd, 1, 0);
155		}
156	5146x	while (r < 0 && errno == EINTR);
157
158	5146x	if (r == 0)
159		{
160	1x	this->complete(EAGAIN, 0);
161	3x	return;
162		}
163	5145x	if (r < 0)
164		{
165		// EAGAIN must not escape: it is the stay-parked sentinel.
166	2x	this->complete(
167	2x	(errno == EAGAIN \|\| errno == EWOULDBLOCK) ? ENOMEM
168	2x	: errno,
169		0);
170	2x	return;
171		}
172
173	5143x	int err = 0;
174	5143x	socklen_t len = sizeof(err);
175	5143x	if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
176	2x	err = errno;
177	5143x	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	144x	void reset() noexcept
200		{
201	144x	Base::reset();
202	144x	wait_event = 0;
203	144x	}
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	268x	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	268x	if (fd < 0)
230		{
231	16x	err = EBADF;
232	16x	return true;
233		}
234
235	252x	pollfd pfd{};
236	252x	pfd.fd = fd;
237	252x	if (event == reactor_event_read)
238	151x	pfd.events = POLLIN;
239	101x	else if (event == reactor_event_write)
240	41x	pfd.events = POLLOUT;
241		else
242	60x	pfd.events = POLLPRI;
243
244		int r;
245		do
246		{
247	252x	r = ::poll(&pfd, 1, 0);
248		}
249	252x	while (r < 0 && errno == EINTR);
250
251	252x	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	2x	err = (errno == EAGAIN \|\| errno == EWOULDBLOCK)
258	2x	? ENOMEM
259	2x	: errno;
260	2x	return true;
261		}
262	250x	return r != 0;
263		}
264
265	107x	void perform_io() noexcept override
266		{
267	107x	int err = 0;
268	107x	if (probe(this->fd, wait_event, err))
269	25x	this->complete(err, 0);
270		else
271	82x	this->complete(EAGAIN, 0);
272	107x	}
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	41494x	bool is_read_operation() const noexcept override
298		{
299	41494x	return !empty_buffer_read;
300		}
301
302	204665x	void reset() noexcept
303		{
304	204665x	Base::reset();
305	204665x	iovec_count = 0;
306	204665x	empty_buffer_read = false;
307	204665x	}
308
309	663x	void perform_io() noexcept override
310		{
311		ssize_t n;
312		do
313		{
314	663x	n = ::readv(this->fd, iovecs, iovec_count);
315		}
316	663x	while (n < 0 && errno == EINTR);
317
318	663x	if (n >= 0)
319	422x	this->complete(0, static_cast<std::size_t>(n));
320		else
321	241x	this->complete(errno, 0);
322	663x	}
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	203987x	void reset() noexcept
349		{
350	203987x	Base::reset();
351	203987x	iovec_count = 0;
352	203987x	}
353
354	135x	void perform_io() noexcept override
355		{
356	135x	ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
357	135x	if (n >= 0)
358	130x	this->complete(0, static_cast<std::size_t>(n));
359		else
360	5x	this->complete(errno, 0);
361	135x	}
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	5208x	void reset() noexcept
391		{
392	5208x	Base::reset();
393	5208x	accepted_fd = -1;
394	5208x	peer_impl = nullptr;
395	5208x	impl_out = nullptr;
396	5208x	peer_storage = {};
397	5208x	peer_addrlen = 0;
398	5208x	}
399
400	5109x	void perform_io() noexcept override
401		{
402	5109x	int new_fd = AcceptPolicy::do_accept(
403	5109x	this->fd, peer_storage, peer_addrlen);
404	5109x	if (new_fd >= 0)
405		{
406	5108x	accepted_fd = new_fd;
407	5108x	this->complete(0, 0);
408		}
409		else
410		{
411	1x	this->complete(errno, 0);
412		}
413	5109x	}
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	115x	void reset() noexcept
438		{
439	115x	Base::reset();
440	115x	iovec_count = 0;
441	115x	msg_flags = 0;
442	115x	}
443
444	34x	void perform_io() noexcept override
445		{
446	34x	msghdr msg{};
447	34x	msg.msg_iov = iovecs;
448	34x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
449
450		#ifdef MSG_NOSIGNAL
451	34x	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	34x	n = ::sendmsg(this->fd, &msg, send_flags);
460		}
461	34x	while (n < 0 && errno == EINTR);
462
463	34x	if (n >= 0)
464	30x	this->complete(0, static_cast<std::size_t>(n));
465		else
466	4x	this->complete(errno, 0);
467	34x	}
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	115x	void reset() noexcept
500		{
501	115x	Base::reset();
502	115x	iovec_count = 0;
503	115x	msg_flags = 0;
504	115x	}
505
506	39x	void perform_io() noexcept override
507		{
508	39x	msghdr msg{};
509	39x	msg.msg_iov = iovecs;
510	39x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
511
512		ssize_t n;
513		do
514		{
515	39x	n = ::recvmsg(this->fd, &msg, msg_flags);
516		}
517	39x	while (n < 0 && errno == EINTR);
518
519	39x	if (n >= 0)
520	34x	this->complete(0, static_cast<std::size_t>(n));
521		else
522	5x	this->complete(errno, 0);
523	39x	}
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	157x	void reset() noexcept
554		{
555	157x	Base::reset();
556	157x	iovec_count = 0;
557	157x	dest_storage = {};
558	157x	dest_len = 0;
559	157x	msg_flags = 0;
560	157x	}
561
562	34x	void perform_io() noexcept override
563		{
564	34x	msghdr msg{};
565	34x	msg.msg_name = &dest_storage;
566	34x	msg.msg_namelen = dest_len;
567	34x	msg.msg_iov = iovecs;
568	34x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
569
570		#ifdef MSG_NOSIGNAL
571	34x	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	34x	n = ::sendmsg(this->fd, &msg, send_flags);
580		}
581	34x	while (n < 0 && errno == EINTR);
582
583	34x	if (n >= 0)
584	30x	this->complete(0, static_cast<std::size_t>(n));
585		else
586	4x	this->complete(errno, 0);
587	34x	}
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	177x	void reset() noexcept
628		{
629	177x	Base::reset();
630	177x	iovec_count = 0;
631	177x	source_storage = {};
632	177x	source_addrlen = 0;
633	177x	source_out = nullptr;
634	177x	msg_flags = 0;
635	177x	}
636
637	47x	void perform_io() noexcept override
638		{
639	47x	msghdr msg{};
640	47x	msg.msg_name = &source_storage;
641	47x	msg.msg_namelen = sizeof(source_storage);
642	47x	msg.msg_iov = iovecs;
643	47x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
644
645		ssize_t n;
646		do
647		{
648	47x	n = ::recvmsg(this->fd, &msg, msg_flags);
649		}
650	47x	while (n < 0 && errno == EINTR);
651
652	47x	if (n >= 0)
653		{
654	42x	source_addrlen = msg.msg_namelen;
655	42x	this->complete(0, static_cast<std::size_t>(n));
656		}
657		else
658	5x	this->complete(errno, 0);
659	47x	}
660		};
661
662		} // namespace boost::corosio::detail
663
664		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
665