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

99.0% Lines (189/4/195) 91.2% List of functions (146/8/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	92752x	reactor_op() = default;
66
67		/// Reset operation state for reuse.
68	415512x	void reset() noexcept
69		{
70	415512x	fd = -1;
71	415512x	errn = 0;
72	415512x	bytes_transferred = 0;
73	415512x	cancelled.store(false, std::memory_order_relaxed);
74	415512x	impl_ptr.reset();
75	415512x	socket_impl_ = nullptr;
76	415512x	acceptor_impl_ = nullptr;
77	415512x	}
78
79		/// Return true if this is a read-direction operation.
80	40819x	virtual bool is_read_operation() const noexcept
81		{
82	40819x	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	387x	void on_cancel() noexcept override
91		{
92	387x	cancel();
93	387x	}
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	87008x	void start(std::stop_token const& token, Socket* impl)
104		{
105	87008x	socket_impl_ = impl;
106	87008x	acceptor_impl_ = nullptr;
107	87008x	coro_op::start(token);
108	87008x	}
109
110		/// Arm the stop-token callback for an acceptor operation.
111	4682x	void start(std::stop_token const& token, Acceptor* impl)
112		{
113	4682x	socket_impl_ = nullptr;
114	4682x	acceptor_impl_ = impl;
115	4682x	coro_op::start(token);
116	4682x	}
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	4615x	void reset() noexcept
135		{
136	4615x	Base::reset();
137	4615x	target_endpoint = Endpoint{};
138	4615x	}
139
140	4513x	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	4513x	pollfd pfd{};
149	4513x	pfd.fd = this->fd;
150	4513x	pfd.events = POLLOUT;
151		int r;
152		do
153		{
154	4513x	r = ::poll(&pfd, 1, 0);
155		}
156	4513x	while (r < 0 && errno == EINTR);
157
158	4513x	if (r == 0)
159		{
160	1x	this->complete(EAGAIN, 0);
161	3x	return;
162		}
163	4512x	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	4510x	int err = 0;
172	4510x	socklen_t len = sizeof(err);
173	4510x	if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
174	2x	err = errno;
175	4510x	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	180x	void reset() noexcept
198		{
199	180x	Base::reset();
200	180x	wait_event = 0;
201	180x	}
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	330x	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	330x	if (fd < 0)
228		{
229	16x	err = EBADF;
230	16x	return true;
231		}
232
233	314x	pollfd pfd{};
234	314x	pfd.fd = fd;
235	314x	if (event == reactor_event_read)
236	209x	pfd.events = POLLIN;
237	105x	else if (event == reactor_event_write)
238	45x	pfd.events = POLLOUT;
239		else
240	60x	pfd.events = POLLPRI;
241
242		int r;
243		do
244		{
245	314x	r = ::poll(&pfd, 1, 0);
246		}
247	314x	while (r < 0 && errno == EINTR);
248
249	314x	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	312x	return r != 0;
259		}
260
261	127x	void perform_io() noexcept override
262		{
263	127x	int err = 0;
264	127x	if (probe(this->fd, wait_event, err))
265	27x	this->complete(err, 0);
266		else
267	100x	this->complete(EAGAIN, 0);
268	127x	}
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	41199x	bool is_read_operation() const noexcept override
294		{
295	41199x	return !empty_buffer_read;
296		}
297
298	203065x	void reset() noexcept
299		{
300	203065x	Base::reset();
301	203065x	iovec_count = 0;
302	203065x	empty_buffer_read = false;
303	203065x	}
304
305	666x	void perform_io() noexcept override
306		{
307		ssize_t n;
308		do
309		{
310	666x	n = ::readv(this->fd, iovecs, iovec_count);
311		}
312	666x	while (n < 0 && errno == EINTR);
313
314	666x	if (n >= 0)
315	424x	this->complete(0, static_cast<std::size_t>(n));
316		else
317	242x	this->complete(errno, 0);
318	666x	}
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	202371x	void reset() noexcept
345		{
346	202371x	Base::reset();
347	202371x	iovec_count = 0;
348	202371x	}
349
350	135x	void perform_io() noexcept override
351		{
352	135x	ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
353	135x	if (n >= 0)
354	130x	this->complete(0, static_cast<std::size_t>(n));
355		else
356	5x	this->complete(errno, 0);
357	135x	}
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	4638x	void reset() noexcept
387		{
388	4638x	Base::reset();
389	4638x	accepted_fd = -1;
390	4638x	peer_impl = nullptr;
391	4638x	impl_out = nullptr;
392	4638x	peer_storage = {};
393	4638x	peer_addrlen = 0;
394	4638x	}
395
396	4478x	void perform_io() noexcept override
397		{
398		int new_fd =
399	4478x	AcceptPolicy::do_accept(this->fd, peer_storage, peer_addrlen);
400	4478x	if (new_fd >= 0)
401		{
402	4476x	accepted_fd = new_fd;
403	4476x	this->complete(0, 0);
404		}
405		else
406		{
407	2x	this->complete(errno, 0);
408		}
409	4478x	}
410		};
411
412		/** Shared connected send operation for datagram sockets.
413
414		Uses sendmsg() with msg_name=nullptr (connected mode).
415
416		@tparam Base The backend's base op type.
417		*/
418		template<class Base>
419		struct reactor_send_op : Base
420		{
421		/// Maximum scatter-gather buffer count.
422		static constexpr std::size_t max_buffers = 16;
423
424		/// Scatter-gather I/O vectors.
425		iovec iovecs[max_buffers];
426
427		/// Number of active I/O vectors.
428		int iovec_count = 0;
429
430		/// User-supplied message flags.
431		int msg_flags = 0;
432
433	125x	void reset() noexcept
434		{
435	125x	Base::reset();
436	125x	iovec_count = 0;
437	125x	msg_flags = 0;
438	125x	}
439
440	34x	void perform_io() noexcept override
441		{
442	34x	msghdr msg{};
443	34x	msg.msg_iov = iovecs;
444	34x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
445
446		#ifdef MSG_NOSIGNAL
447	34x	int send_flags = msg_flags \| MSG_NOSIGNAL;
448		#else
449		int send_flags = msg_flags;
450		#endif
451
452		ssize_t n;
453		do
454		{
455	34x	n = ::sendmsg(this->fd, &msg, send_flags);
456		}
457	34x	while (n < 0 && errno == EINTR);
458
459	34x	if (n >= 0)
460	30x	this->complete(0, static_cast<std::size_t>(n));
461		else
462	4x	this->complete(errno, 0);
463	34x	}
464		};
465
466		/** Shared connected recv operation for datagram sockets.
467
468		Uses recvmsg() with msg_name=nullptr (connected mode).
469		Unlike reactor_read_op, does not map n==0 to EOF
470		(zero-length datagrams are valid).
471
472		@tparam Base The backend's base op type.
473		*/
474		template<class Base>
475		struct reactor_recv_op : Base
476		{
477		/// Maximum scatter-gather buffer count.
478		static constexpr std::size_t max_buffers = 16;
479
480		/// Scatter-gather I/O vectors.
481		iovec iovecs[max_buffers];
482
483		/// Number of active I/O vectors.
484		int iovec_count = 0;
485
486		/// User-supplied message flags.
487		int msg_flags = 0;
488
489		/// Return true (this is a read-direction operation).
490		// LCOV_EXCL_START: devirtualized and inlined at the templated
491		// completion call site; the out-of-line body is never entered.
492	−	bool is_read_operation() const noexcept override
493		{
494	−	return true;
495		}
496		// LCOV_EXCL_STOP
497
498	162x	void reset() noexcept
499		{
500	162x	Base::reset();
501	162x	iovec_count = 0;
502	162x	msg_flags = 0;
503	162x	}
504
505	39x	void perform_io() noexcept override
506		{
507	39x	msghdr msg{};
508	39x	msg.msg_iov = iovecs;
509	39x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
510
511		ssize_t n;
512		do
513		{
514	39x	n = ::recvmsg(this->fd, &msg, msg_flags);
515		}
516	39x	while (n < 0 && errno == EINTR);
517
518	39x	if (n >= 0)
519	34x	this->complete(0, static_cast<std::size_t>(n));
520		else
521	5x	this->complete(errno, 0);
522	39x	}
523		};
524
525		/** Shared send_to operation for datagram sockets.
526
527		Uses sendmsg() with the destination endpoint in msg_name.
528
529		@tparam Base The backend's base op type.
530		*/
531		template<class Base>
532		struct reactor_send_to_op : Base
533		{
534		/// Maximum scatter-gather buffer count.
535		static constexpr std::size_t max_buffers = 16;
536
537		/// Scatter-gather I/O vectors.
538		iovec iovecs[max_buffers];
539
540		/// Number of active I/O vectors.
541		int iovec_count = 0;
542
543		/// Destination address storage.
544		sockaddr_storage dest_storage{};
545
546		/// Destination address length.
547		socklen_t dest_len = 0;
548
549		/// User-supplied message flags.
550		int msg_flags = 0;
551
552	167x	void reset() noexcept
553		{
554	167x	Base::reset();
555	167x	iovec_count = 0;
556	167x	dest_storage = {};
557	167x	dest_len = 0;
558	167x	msg_flags = 0;
559	167x	}
560
561	34x	void perform_io() noexcept override
562		{
563	34x	msghdr msg{};
564	34x	msg.msg_name = &dest_storage;
565	34x	msg.msg_namelen = dest_len;
566	34x	msg.msg_iov = iovecs;
567	34x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
568
569		#ifdef MSG_NOSIGNAL
570	34x	int send_flags = msg_flags \| MSG_NOSIGNAL;
571		#else
572		int send_flags = msg_flags;
573		#endif
574
575		ssize_t n;
576		do
577		{
578	34x	n = ::sendmsg(this->fd, &msg, send_flags);
579		}
580	34x	while (n < 0 && errno == EINTR);
581
582	34x	if (n >= 0)
583	30x	this->complete(0, static_cast<std::size_t>(n));
584		else
585	4x	this->complete(errno, 0);
586	34x	}
587		};
588
589		/** Shared recv_from operation for datagram sockets.
590
591		Uses recvmsg() with msg_name to capture the source endpoint.
592
593		@tparam Base The backend's base op type.
594		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
595		*/
596		template<class Base, class Endpoint = endpoint>
597		struct reactor_recv_from_op : Base
598		{
599		/// Maximum scatter-gather buffer count.
600		static constexpr std::size_t max_buffers = 16;
601
602		/// Scatter-gather I/O vectors.
603		iovec iovecs[max_buffers];
604
605		/// Number of active I/O vectors.
606		int iovec_count = 0;
607
608		/// Source address storage filled by recvmsg.
609		sockaddr_storage source_storage{};
610
611		/// Actual source address length returned by recvmsg.
612		socklen_t source_addrlen = 0;
613
614		/// Output pointer for the source endpoint (set by do_recv_from).
615		Endpoint* source_out = nullptr;
616
617		/// User-supplied message flags.
618		int msg_flags = 0;
619
620		/// Return true (this is a read-direction operation).
621		// LCOV_EXCL_START: devirtualized and inlined at the templated
622		// completion call site; the out-of-line body is never entered.
623	−	bool is_read_operation() const noexcept override
624		{
625	−	return true;
626		}
627		// LCOV_EXCL_STOP
628
629	189x	void reset() noexcept
630		{
631	189x	Base::reset();
632	189x	iovec_count = 0;
633	189x	source_storage = {};
634	189x	source_addrlen = 0;
635	189x	source_out = nullptr;
636	189x	msg_flags = 0;
637	189x	}
638
639	49x	void perform_io() noexcept override
640		{
641	49x	msghdr msg{};
642	49x	msg.msg_name = &source_storage;
643	49x	msg.msg_namelen = sizeof(source_storage);
644	49x	msg.msg_iov = iovecs;
645	49x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
646
647		ssize_t n;
648		do
649		{
650	49x	n = ::recvmsg(this->fd, &msg, msg_flags);
651		}
652	49x	while (n < 0 && errno == EINTR);
653
654	49x	if (n >= 0)
655		{
656	44x	source_addrlen = msg.msg_namelen;
657	44x	this->complete(0, static_cast<std::size_t>(n));
658		}
659		else
660	5x	this->complete(errno, 0);
661	49x	}
662		};
663
664		} // namespace boost::corosio::detail
665
666		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
667