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

99.0% Lines (194 / 196, 4 excl) 90.7% Functions (176 / 194, 8 excl)

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

reactor_op.hpp

f(x) Functions (202)

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		#include <unistd.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, object_ref_, 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		// The op envelope — coroutine handle h, cont, executor ex, ec_out,
53		// bytes_out, cancelled, stop_cb (+ its canceller), object_ref_ — lives in
54		// coro_op (via reactor_op_base) and is shared with io_uring/IOCP.
55		// reactor_op adds only the reactor-specific routing state below.
56
57		/// File descriptor this operation targets.
58		int fd = -1;
59
60		/// Owning socket impl (for stop_token cancellation routing).
61		Socket* socket_impl_ = nullptr;
62
63		/// Owning acceptor impl (for stop_token cancellation routing).
64		Acceptor* acceptor_impl_ = nullptr;
65
66	43012x	reactor_op() = default;
67
68		/// Reset operation state for reuse.
69	490564x	void reset() noexcept
70		{
71	490564x	fd = -1;
72	490564x	errn = 0;
73	490564x	bytes_transferred = 0;
74	490564x	cancelled.store(false, std::memory_order_relaxed);
75	490564x	object_ref_.reset();
76	490564x	socket_impl_ = nullptr;
77	490564x	acceptor_impl_ = nullptr;
78	490564x	}
79
80		/// Return true if this is a read-direction operation.
81	48016x	virtual bool is_read_operation() const noexcept
82		{
83	48016x	return false;
84		}
85
86		/// Cancel this operation via the owning impl.
87		virtual void cancel() noexcept = 0;
88
89		/// coro_op cancellation hook (fired by the shared canceller when the
90		/// stop_token requests cancellation): route to the impl-specific cancel().
91	676x	void on_cancel() noexcept override
92		{
93	676x	cancel();
94	676x	}
95
96		/// Destroy without invoking.
97	1290x	void destroy() override
98		{
99	1290x	reactor_op_base::destroy();
100	1290x	}
101
102		/// Arm the stop-token callback for a socket operation.
103	105715x	void start(std::stop_token const& token, Socket* impl)
104		{
105	105715x	socket_impl_ = impl;
106	105715x	acceptor_impl_ = nullptr;
107	105715x	coro_op::start(token);
108	105715x	}
109
110		/// Arm the stop-token callback for an acceptor operation.
111	7531x	void start(std::stop_token const& token, Acceptor* impl)
112		{
113	7531x	socket_impl_ = nullptr;
114	7531x	acceptor_impl_ = impl;
115	7531x	coro_op::start(token);
116	7531x	}
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	8229x	void reset() noexcept
135		{
136	8229x	Base::reset();
137	8229x	target_endpoint = Endpoint{};
138	8229x	}
139
140	8101x	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	8101x	pollfd pfd{};
149	8101x	pfd.fd = this->fd;
150	8101x	pfd.events = POLLOUT;
151		int r;
152		do
153		{
154	8101x	r = ::poll(&pfd, 1, 0);
155		}
156	8101x	while (r < 0 && errno == EINTR);
157
158	8101x	if (r == 0)
159		{
160	1x	this->complete(EAGAIN, 0);
161	3x	return;
162		}
163	8100x	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 : errno, 0);
168	2x	return;
169		}
170
171	8098x	int err = 0;
172	8098x	socklen_t len = sizeof(err);
173	8098x	if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
174	2x	err = errno;
175	8098x	this->complete(err, 0);
176		}
177		};
178
179		/** Readiness-only wait operation.
180
181		Completion is decided by probing the descriptor with a
182		zero-timeout `poll()`, never by the reactor's cached edge
183		events: a speculative read can drain the socket without
184		touching the reactor (stale edge), and a short read can
185		consume the edge while data remains buffered (missing edge).
186		`perform_io()` runs the probe and reports `EAGAIN` when the
187		condition does not currently hold, which keeps the op parked.
188
189		@tparam Base The backend's base op type.
190		*/
191		template<class Base>
192		struct reactor_wait_op : Base
193		{
194		/// Which event bit this wait targets (reactor_event_read/write/error).
195		std::uint32_t wait_event = 0;
196
197	206x	void reset() noexcept
198		{
199	206x	Base::reset();
200	206x	wait_event = 0;
201	206x	}
202
203	✗	bool is_read_operation() const noexcept override
204		{
205	✗	return wait_event == reactor_event_read;
206		}
207
208		/** Check whether the waited-for condition currently holds.
209
210		Zero-timeout `poll()` probe. `POLLERR`/`POLLHUP` count as
211		ready for every wait type: the wait must not park on a
212		socket whose next I/O would fail immediately. The probe is
213		side-effect free — in particular it never reads `SO_ERROR`,
214		which is consume-on-read and belongs to whichever operation
215		observes the failure next.
216
217		@param fd The descriptor to probe.
218		@param event The event bit to probe for (read/write/error).
219		@param err Receives the probe failure, if any.
220
221		@return `true` if the condition holds or the probe failed.
222		*/
223	371x	static bool probe(int fd, std::uint32_t event, int& err) noexcept
224		{
225		// poll() silently ignores negative fds; without this guard a
226		// wait on a never-opened or closed socket parks forever.
227	371x	if (fd < 0)
228		{
229	18x	err = EBADF;
230	18x	return true;
231		}
232
233	353x	pollfd pfd{};
234	353x	pfd.fd = fd;
235	353x	if (event == reactor_event_read)
236	228x	pfd.events = POLLIN;
237	125x	else if (event == reactor_event_write)
238	65x	pfd.events = POLLOUT;
239		else
240	60x	pfd.events = POLLPRI;
241
242		int r;
243		do
244		{
245	353x	r = ::poll(&pfd, 1, 0);
246		}
247	353x	while (r < 0 && errno == EINTR);
248
249	353x	if (r < 0)
250		{
251		// Complete with the probe failure rather than park forever.
252		// EAGAIN must not escape here: callers treat it as the
253		// stay-parked sentinel, and poll() can fail with it on
254		// BSD/macOS under transient resource pressure.
255	2x	err = (errno == EAGAIN \|\| errno == EWOULDBLOCK) ? ENOMEM : errno;
256	2x	return true;
257		}
258	351x	return r != 0;
259		}
260
261	123x	void perform_io() noexcept override
262		{
263	123x	int err = 0;
264	123x	if (probe(this->fd, wait_event, err))
265	37x	this->complete(err, 0);
266		else
267	86x	this->complete(EAGAIN, 0);
268	123x	}
269		};
270
271		/** Shared scatter-read operation.
272
273		Uses readv() with an EINTR retry loop.
274
275		@tparam Base The backend's base op type.
276		*/
277		template<class Base>
278		struct reactor_read_op : Base
279		{
280		/// Maximum scatter-gather buffer count.
281		static constexpr std::size_t max_buffers = 16;
282
283		/// Scatter-gather I/O vectors.
284		iovec iovecs[max_buffers];
285
286		/// Number of active I/O vectors.
287		int iovec_count = 0;
288
289		/// True for zero-length reads (completed immediately).
290		bool empty_buffer_read = false;
291
292		/// Return true (this is a read-direction operation).
293	48651x	bool is_read_operation() const noexcept override
294		{
295	48651x	return !empty_buffer_read;
296		}
297
298	237619x	void reset() noexcept
299		{
300	237619x	Base::reset();
301	237619x	iovec_count = 0;
302	237619x	empty_buffer_read = false;
303	237619x	}
304
305	1003x	void perform_io() noexcept override
306		{
307		ssize_t n;
308		do
309		{
310	1003x	n = ::readv(this->fd, iovecs, iovec_count);
311		}
312	1003x	while (n < 0 && errno == EINTR);
313
314	1003x	if (n >= 0)
315	893x	this->complete(0, static_cast<std::size_t>(n));
316		else
317	110x	this->complete(errno, 0);
318	1003x	}
319		};
320
321		/** Shared gather-write operation.
322
323		Delegates the actual syscall to WritePolicy::write(fd, iovecs, count),
324		which returns ssize_t (bytes written or -1 with errno set).
325
326		@tparam Base The backend's base op type.
327		@tparam WritePolicy Provides `static ssize_t write(int, iovec*, int)`.
328		*/
329		template<class Base, class WritePolicy>
330		struct reactor_write_op : Base
331		{
332		/// The write syscall policy type.
333		using write_policy = WritePolicy;
334
335		/// Maximum scatter-gather buffer count.
336		static constexpr std::size_t max_buffers = 16;
337
338		/// Scatter-gather I/O vectors.
339		iovec iovecs[max_buffers];
340
341		/// Number of active I/O vectors.
342		int iovec_count = 0;
343
344	236378x	void reset() noexcept
345		{
346	236378x	Base::reset();
347	236378x	iovec_count = 0;
348	236378x	}
349
350	140x	void perform_io() noexcept override
351		{
352	140x	ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
353	140x	if (n >= 0)
354	132x	this->complete(0, static_cast<std::size_t>(n));
355		else
356	8x	this->complete(errno, 0);
357	140x	}
358		};
359
360		/** Shared accept operation.
361
362		Delegates the actual syscall to AcceptPolicy::do_accept(fd, peer_storage),
363		which returns the accepted fd or -1 with errno set.
364
365		@tparam Base The backend's base op type.
366		@tparam AcceptPolicy Provides `static int do_accept(int, sockaddr_storage&)`.
367		*/
368		template<class Base, class AcceptPolicy>
369		struct reactor_accept_op : Base
370		{
371		/// File descriptor of the accepted connection.
372		int accepted_fd = -1;
373
374		/// Pointer to the peer socket implementation.
375		io_object::implementation* peer_impl = nullptr;
376
377		/// Output pointer for the accepted implementation.
378		io_object::implementation** impl_out = nullptr;
379
380		/// Peer address storage filled by accept.
381		sockaddr_storage peer_storage{};
382
383		/// Peer address length returned by accept.
384		socklen_t peer_addrlen = 0;
385
386	7490x	void reset() noexcept
387		{
388	7490x	Base::reset();
389	7490x	accepted_fd = -1;
390	7490x	peer_impl = nullptr;
391	7490x	impl_out = nullptr;
392	7490x	peer_storage = {};
393	7490x	peer_addrlen = 0;
394	7490x	}
395
396		/// Destroy without invoking, closing a connection already taken.
397	816x	void destroy() override
398		{
399	816x	if (accepted_fd >= 0)
400		{
401	5x	::close(accepted_fd);
402	5x	accepted_fd = -1;
403		}
404	816x	Base::destroy();
405	816x	}
406
407	5768x	void perform_io() noexcept override
408		{
409		int new_fd =
410	5768x	AcceptPolicy::do_accept(this->fd, peer_storage, peer_addrlen);
411	5768x	if (new_fd >= 0)
412		{
413	5765x	accepted_fd = new_fd;
414	5765x	this->complete(0, 0);
415		}
416		else
417		{
418	3x	this->complete(errno, 0);
419		}
420	5768x	}
421		};
422
423		/** Shared connected send operation for datagram sockets.
424
425		Uses sendmsg() with msg_name=nullptr (connected mode).
426
427		@tparam Base The backend's base op type.
428		*/
429		template<class Base>
430		struct reactor_send_op : Base
431		{
432		/// Maximum scatter-gather buffer count.
433		static constexpr std::size_t max_buffers = 16;
434
435		/// Scatter-gather I/O vectors.
436		iovec iovecs[max_buffers];
437
438		/// Number of active I/O vectors.
439		int iovec_count = 0;
440
441		/// User-supplied message flags.
442		int msg_flags = 0;
443
444	123x	void reset() noexcept
445		{
446	123x	Base::reset();
447	123x	iovec_count = 0;
448	123x	msg_flags = 0;
449	123x	}
450
451	34x	void perform_io() noexcept override
452		{
453	34x	msghdr msg{};
454	34x	msg.msg_iov = iovecs;
455	34x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
456
457		#ifdef MSG_NOSIGNAL
458	34x	int send_flags = msg_flags \| MSG_NOSIGNAL;
459		#else
460		int send_flags = msg_flags;
461		#endif
462
463		ssize_t n;
464		do
465		{
466	34x	n = ::sendmsg(this->fd, &msg, send_flags);
467		}
468	34x	while (n < 0 && errno == EINTR);
469
470	34x	if (n >= 0)
471	30x	this->complete(0, static_cast<std::size_t>(n));
472		else
473	4x	this->complete(errno, 0);
474	34x	}
475		};
476
477		/** Shared connected recv operation for datagram sockets.
478
479		Uses recvmsg() with msg_name=nullptr (connected mode).
480		Unlike reactor_read_op, does not map n==0 to EOF
481		(zero-length datagrams are valid).
482
483		@tparam Base The backend's base op type.
484		*/
485		template<class Base>
486		struct reactor_recv_op : Base
487		{
488		/// Maximum scatter-gather buffer count.
489		static constexpr std::size_t max_buffers = 16;
490
491		/// Scatter-gather I/O vectors.
492		iovec iovecs[max_buffers];
493
494		/// Number of active I/O vectors.
495		int iovec_count = 0;
496
497		/// User-supplied message flags.
498		int msg_flags = 0;
499
500		/// Return true (this is a read-direction operation).
501		// LCOV_EXCL_START: devirtualized and inlined at the templated
502		// completion call site; the out-of-line body is never entered.
503	−	bool is_read_operation() const noexcept override
504		{
505	−	return true;
506		}
507		// LCOV_EXCL_STOP
508
509	165x	void reset() noexcept
510		{
511	165x	Base::reset();
512	165x	iovec_count = 0;
513	165x	msg_flags = 0;
514	165x	}
515
516	39x	void perform_io() noexcept override
517		{
518	39x	msghdr msg{};
519	39x	msg.msg_iov = iovecs;
520	39x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
521
522		ssize_t n;
523		do
524		{
525	39x	n = ::recvmsg(this->fd, &msg, msg_flags);
526		}
527	39x	while (n < 0 && errno == EINTR);
528
529	39x	if (n >= 0)
530	36x	this->complete(0, static_cast<std::size_t>(n));
531		else
532	3x	this->complete(errno, 0);
533	39x	}
534		};
535
536		/** Shared send_to operation for datagram sockets.
537
538		Uses sendmsg() with the destination endpoint in msg_name.
539
540		@tparam Base The backend's base op type.
541		*/
542		template<class Base>
543		struct reactor_send_to_op : Base
544		{
545		/// Maximum scatter-gather buffer count.
546		static constexpr std::size_t max_buffers = 16;
547
548		/// Scatter-gather I/O vectors.
549		iovec iovecs[max_buffers];
550
551		/// Number of active I/O vectors.
552		int iovec_count = 0;
553
554		/// Destination address storage.
555		sockaddr_storage dest_storage{};
556
557		/// Destination address length.
558		socklen_t dest_len = 0;
559
560		/// User-supplied message flags.
561		int msg_flags = 0;
562
563	167x	void reset() noexcept
564		{
565	167x	Base::reset();
566	167x	iovec_count = 0;
567	167x	dest_storage = {};
568	167x	dest_len = 0;
569	167x	msg_flags = 0;
570	167x	}
571
572	34x	void perform_io() noexcept override
573		{
574	34x	msghdr msg{};
575	34x	msg.msg_name = &dest_storage;
576	34x	msg.msg_namelen = dest_len;
577	34x	msg.msg_iov = iovecs;
578	34x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
579
580		#ifdef MSG_NOSIGNAL
581	34x	int send_flags = msg_flags \| MSG_NOSIGNAL;
582		#else
583		int send_flags = msg_flags;
584		#endif
585
586		ssize_t n;
587		do
588		{
589	34x	n = ::sendmsg(this->fd, &msg, send_flags);
590		}
591	34x	while (n < 0 && errno == EINTR);
592
593	34x	if (n >= 0)
594	30x	this->complete(0, static_cast<std::size_t>(n));
595		else
596	4x	this->complete(errno, 0);
597	34x	}
598		};
599
600		/** Shared recv_from operation for datagram sockets.
601
602		Uses recvmsg() with msg_name to capture the source endpoint.
603
604		@tparam Base The backend's base op type.
605		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
606		*/
607		template<class Base, class Endpoint = endpoint>
608		struct reactor_recv_from_op : Base
609		{
610		/// Maximum scatter-gather buffer count.
611		static constexpr std::size_t max_buffers = 16;
612
613		/// Scatter-gather I/O vectors.
614		iovec iovecs[max_buffers];
615
616		/// Number of active I/O vectors.
617		int iovec_count = 0;
618
619		/// Source address storage filled by recvmsg.
620		sockaddr_storage source_storage{};
621
622		/// Actual source address length returned by recvmsg.
623		socklen_t source_addrlen = 0;
624
625		/// Output pointer for the source endpoint (set by do_recv_from).
626		Endpoint* source_out = nullptr;
627
628		/// User-supplied message flags.
629		int msg_flags = 0;
630
631		/// Return true (this is a read-direction operation).
632		// LCOV_EXCL_START: devirtualized and inlined at the templated
633		// completion call site; the out-of-line body is never entered.
634	−	bool is_read_operation() const noexcept override
635		{
636	−	return true;
637		}
638		// LCOV_EXCL_STOP
639
640	187x	void reset() noexcept
641		{
642	187x	Base::reset();
643	187x	iovec_count = 0;
644	187x	source_storage = {};
645	187x	source_addrlen = 0;
646	187x	source_out = nullptr;
647	187x	msg_flags = 0;
648	187x	}
649
650	51x	void perform_io() noexcept override
651		{
652	51x	msghdr msg{};
653	51x	msg.msg_name = &source_storage;
654	51x	msg.msg_namelen = sizeof(source_storage);
655	51x	msg.msg_iov = iovecs;
656	51x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
657
658		ssize_t n;
659		do
660		{
661	51x	n = ::recvmsg(this->fd, &msg, msg_flags);
662		}
663	51x	while (n < 0 && errno == EINTR);
664
665	51x	if (n >= 0)
666		{
667	48x	source_addrlen = msg.msg_namelen;
668	48x	this->complete(0, static_cast<std::size_t>(n));
669		}
670		else
671	3x	this->complete(errno, 0);
672	51x	}
673		};
674
675		} // namespace boost::corosio::detail
676
677		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
678