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

59.9% Lines (479/800) 66.2% List of functions (90/136)

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

reactor_op.hpp

f(x) Functions (136)

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/corosio/detail/continuation_op.hpp>
17		#include <boost/capy/ex/executor_ref.hpp>
18
19		#include <atomic>
20		#include <coroutine>
21		#include <cstddef>
22		#include <memory>
23		#include <optional>
24		#include <stop_token>
25		#include <system_error>
26
27		#include <errno.h>
28
29		#include <netinet/in.h>
30		#include <sys/socket.h>
31		#include <sys/uio.h>
32
33		namespace boost::corosio::detail {
34
35		/** Base operation for reactor-based backends.
36
37		Holds per-operation state that depends on the concrete backend
38		socket/acceptor types: coroutine handle, executor, output
39		pointers, file descriptor, stop_callback, and type-specific
40		impl pointers.
41
42		Fields shared across all backends (errn, bytes_transferred,
43		cancelled, impl_ptr, perform_io, complete) live in
44		reactor_op_base so the scheduler and descriptor_state can
45		access them without template instantiation.
46
47		@tparam Socket The backend socket impl type (forward-declared).
48		@tparam Acceptor The backend acceptor impl type (forward-declared).
49		*/
50		template<class Socket, class Acceptor>
51		struct reactor_op : reactor_op_base
52		{
53		/// Stop-token callback that invokes cancel() on the target op.
54		struct canceller
55		{
56		reactor_op* op;
57	207x	void operator()() const noexcept
58		{
59	207x	op->cancel();
60	207x	}
61		};
62
63		/// Caller's coroutine handle to resume on completion.
64		std::coroutine_handle<> h;
65
66		/// Scheduler-ready continuation for executor dispatch/post (wraps h).
67		detail::continuation_op cont_op;
68
69		/// Executor for dispatching the completion.
70		capy::executor_ref ex;
71
72		/// Output pointer for the error code.
73		std::error_code* ec_out = nullptr;
74
75		/// Output pointer for bytes transferred.
76		std::size_t* bytes_out = nullptr;
77
78		/// File descriptor this operation targets.
79		int fd = -1;
80
81		/// Stop-token callback registration.
82		std::optional<std::stop_callback<canceller>> stop_cb;
83
84		/// Owning socket impl (for stop_token cancellation).
85		Socket* socket_impl_ = nullptr;
86
87		/// Owning acceptor impl (for stop_token cancellation).
88		Acceptor* acceptor_impl_ = nullptr;
89
90	66182x	reactor_op() = default;
91
92		/// Reset operation state for reuse.
93	717632x	void reset() noexcept
94		{
95	717632x	fd = -1;
96	717632x	errn = 0;
97	717632x	bytes_transferred = 0;
98	717632x	cancelled.store(false, std::memory_order_relaxed);
99	717632x	impl_ptr.reset();
100	717632x	socket_impl_ = nullptr;
101	717632x	acceptor_impl_ = nullptr;
102	717632x	}
103
104		/// Return true if this is a read-direction operation.
105	70280x	virtual bool is_read_operation() const noexcept
106		{
107	70280x	return false;
108		}
109
110		/// Cancel this operation via the owning impl.
111		virtual void cancel() noexcept = 0;
112
113		/// Destroy without invoking.
114	✗	void destroy() override
115		{
116	✗	stop_cb.reset();
117	✗	reactor_op_base::destroy();
118	✗	}
119
120		/// Arm the stop-token callback for a socket operation.
121	148180x	void start(std::stop_token const& token, Socket* impl)
122		{
123	148180x	cancelled.store(false, std::memory_order_release);
124	148180x	stop_cb.reset();
125	148180x	socket_impl_ = impl;
126	148180x	acceptor_impl_ = nullptr;
127
128	148180x	if (token.stop_possible())
129	205x	stop_cb.emplace(token, canceller{this});
130	148180x	}
131
132		/// Arm the stop-token callback for an acceptor operation.
133	7240x	void start(std::stop_token const& token, Acceptor* impl)
134		{
135	7240x	cancelled.store(false, std::memory_order_release);
136	7240x	stop_cb.reset();
137	7240x	socket_impl_ = nullptr;
138	7240x	acceptor_impl_ = impl;
139
140	7240x	if (token.stop_possible())
141	9x	stop_cb.emplace(token, canceller{this});
142	7240x	}
143		};
144
145		/** Shared connect operation.
146
147		Checks SO_ERROR for connect completion status. The operator()()
148		and cancel() are provided by the concrete backend type.
149
150		@tparam Base The backend's base op type.
151		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
152		*/
153		template<class Base, class Endpoint = endpoint>
154		struct reactor_connect_op : Base
155		{
156		/// Endpoint to connect to.
157		Endpoint target_endpoint;
158
159		/// Reset operation state for reuse.
160	7242x	void reset() noexcept
161		{
162	7242x	Base::reset();
163	7242x	target_endpoint = Endpoint{};
164	7242x	}
165
166	7227x	void perform_io() noexcept override
167		{
168	7227x	int err = 0;
169	7227x	socklen_t len = sizeof(err);
170	7227x	if (::getsockopt(this->fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
171	✗	err = errno;
172	7227x	this->complete(err, 0);
173	7227x	}
174		};
175
176		/** Shared scatter-read operation.
177
178		Uses readv() with an EINTR retry loop.
179
180		@tparam Base The backend's base op type.
181		*/
182		template<class Base>
183		struct reactor_read_op : Base
184		{
185		/// Maximum scatter-gather buffer count.
186		static constexpr std::size_t max_buffers = 16;
187
188		/// Scatter-gather I/O vectors.
189		iovec iovecs[max_buffers];
190
191		/// Number of active I/O vectors.
192		int iovec_count = 0;
193
194		/// True for zero-length reads (completed immediately).
195		bool empty_buffer_read = false;
196
197		/// Return true (this is a read-direction operation).
198	70315x	bool is_read_operation() const noexcept override
199		{
200	70315x	return !empty_buffer_read;
201		}
202
203	351680x	void reset() noexcept
204		{
205	351680x	Base::reset();
206	351680x	iovec_count = 0;
207	351680x	empty_buffer_read = false;
208	351680x	}
209
210	341x	void perform_io() noexcept override
211		{
212		ssize_t n;
213		do
214		{
215	341x	n = ::readv(this->fd, iovecs, iovec_count);
216		}
217	341x	while (n < 0 && errno == EINTR);
218
219	341x	if (n >= 0)
220	101x	this->complete(0, static_cast<std::size_t>(n));
221		else
222	240x	this->complete(errno, 0);
223	341x	}
224		};
225
226		/** Shared gather-write operation.
227
228		Delegates the actual syscall to WritePolicy::write(fd, iovecs, count),
229		which returns ssize_t (bytes written or -1 with errno set).
230
231		@tparam Base The backend's base op type.
232		@tparam WritePolicy Provides `static ssize_t write(int, iovec*, int)`.
233		*/
234		template<class Base, class WritePolicy>
235		struct reactor_write_op : Base
236		{
237		/// The write syscall policy type.
238		using write_policy = WritePolicy;
239
240		/// Maximum scatter-gather buffer count.
241		static constexpr std::size_t max_buffers = 16;
242
243		/// Scatter-gather I/O vectors.
244		iovec iovecs[max_buffers];
245
246		/// Number of active I/O vectors.
247		int iovec_count = 0;
248
249	351374x	void reset() noexcept
250		{
251	351374x	Base::reset();
252	351374x	iovec_count = 0;
253	351374x	}
254
255	✗	void perform_io() noexcept override
256		{
257	✗	ssize_t n = WritePolicy::write(this->fd, iovecs, iovec_count);
258	✗	if (n >= 0)
259	✗	this->complete(0, static_cast<std::size_t>(n));
260		else
261	✗	this->complete(errno, 0);
262	✗	}
263		};
264
265		/** Shared accept operation.
266
267		Delegates the actual syscall to AcceptPolicy::do_accept(fd, peer_storage),
268		which returns the accepted fd or -1 with errno set.
269
270		@tparam Base The backend's base op type.
271		@tparam AcceptPolicy Provides `static int do_accept(int, sockaddr_storage&)`.
272		*/
273		template<class Base, class AcceptPolicy>
274		struct reactor_accept_op : Base
275		{
276		/// File descriptor of the accepted connection.
277		int accepted_fd = -1;
278
279		/// Pointer to the peer socket implementation.
280		io_object::implementation* peer_impl = nullptr;
281
282		/// Output pointer for the accepted implementation.
283		io_object::implementation** impl_out = nullptr;
284
285		/// Peer address storage filled by accept.
286		sockaddr_storage peer_storage{};
287
288		/// Peer address length returned by accept.
289		socklen_t peer_addrlen = 0;
290
291	7240x	void reset() noexcept
292		{
293	7240x	Base::reset();
294	7240x	accepted_fd = -1;
295	7240x	peer_impl = nullptr;
296	7240x	impl_out = nullptr;
297	7240x	peer_storage = {};
298	7240x	peer_addrlen = 0;
299	7240x	}
300
301	7222x	void perform_io() noexcept override
302		{
303	7222x	int new_fd = AcceptPolicy::do_accept(
304	7222x	this->fd, peer_storage, peer_addrlen);
305	7222x	if (new_fd >= 0)
306		{
307	7222x	accepted_fd = new_fd;
308	7222x	this->complete(0, 0);
309		}
310		else
311		{
312	✗	this->complete(errno, 0);
313		}
314	7222x	}
315		};
316
317		/** Shared connected send operation for datagram sockets.
318
319		Uses sendmsg() with msg_name=nullptr (connected mode).
320
321		@tparam Base The backend's base op type.
322		*/
323		template<class Base>
324		struct reactor_send_op : Base
325		{
326		/// Maximum scatter-gather buffer count.
327		static constexpr std::size_t max_buffers = 16;
328
329		/// Scatter-gather I/O vectors.
330		iovec iovecs[max_buffers];
331
332		/// Number of active I/O vectors.
333		int iovec_count = 0;
334
335		/// User-supplied message flags.
336		int msg_flags = 0;
337
338	14x	void reset() noexcept
339		{
340	14x	Base::reset();
341	14x	iovec_count = 0;
342	14x	msg_flags = 0;
343	14x	}
344
345	✗	void perform_io() noexcept override
346		{
347	✗	msghdr msg{};
348	✗	msg.msg_iov = iovecs;
349	✗	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
350
351		#ifdef MSG_NOSIGNAL
352	✗	int send_flags = msg_flags \| MSG_NOSIGNAL;
353		#else
354		int send_flags = msg_flags;
355		#endif
356
357		ssize_t n;
358		do
359		{
360	✗	n = ::sendmsg(this->fd, &msg, send_flags);
361		}
362	✗	while (n < 0 && errno == EINTR);
363
364	✗	if (n >= 0)
365	✗	this->complete(0, static_cast<std::size_t>(n));
366		else
367	✗	this->complete(errno, 0);
368	✗	}
369		};
370
371		/** Shared connected recv operation for datagram sockets.
372
373		Uses recvmsg() with msg_name=nullptr (connected mode).
374		Unlike reactor_read_op, does not map n==0 to EOF
375		(zero-length datagrams are valid).
376
377		@tparam Base The backend's base op type.
378		*/
379		template<class Base>
380		struct reactor_recv_op : Base
381		{
382		/// Maximum scatter-gather buffer count.
383		static constexpr std::size_t max_buffers = 16;
384
385		/// Scatter-gather I/O vectors.
386		iovec iovecs[max_buffers];
387
388		/// Number of active I/O vectors.
389		int iovec_count = 0;
390
391		/// User-supplied message flags.
392		int msg_flags = 0;
393
394		/// Return true (this is a read-direction operation).
395	2x	bool is_read_operation() const noexcept override
396		{
397	2x	return true;
398		}
399
400	14x	void reset() noexcept
401		{
402	14x	Base::reset();
403	14x	iovec_count = 0;
404	14x	msg_flags = 0;
405	14x	}
406
407	✗	void perform_io() noexcept override
408		{
409	✗	msghdr msg{};
410	✗	msg.msg_iov = iovecs;
411	✗	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
412
413		ssize_t n;
414		do
415		{
416	✗	n = ::recvmsg(this->fd, &msg, msg_flags);
417		}
418	✗	while (n < 0 && errno == EINTR);
419
420	✗	if (n >= 0)
421	✗	this->complete(0, static_cast<std::size_t>(n));
422		else
423	✗	this->complete(errno, 0);
424	✗	}
425		};
426
427		/** Shared send_to operation for datagram sockets.
428
429		Uses sendmsg() with the destination endpoint in msg_name.
430
431		@tparam Base The backend's base op type.
432		*/
433		template<class Base>
434		struct reactor_send_to_op : Base
435		{
436		/// Maximum scatter-gather buffer count.
437		static constexpr std::size_t max_buffers = 16;
438
439		/// Scatter-gather I/O vectors.
440		iovec iovecs[max_buffers];
441
442		/// Number of active I/O vectors.
443		int iovec_count = 0;
444
445		/// Destination address storage.
446		sockaddr_storage dest_storage{};
447
448		/// Destination address length.
449		socklen_t dest_len = 0;
450
451		/// User-supplied message flags.
452		int msg_flags = 0;
453
454	28x	void reset() noexcept
455		{
456	28x	Base::reset();
457	28x	iovec_count = 0;
458	28x	dest_storage = {};
459	28x	dest_len = 0;
460	28x	msg_flags = 0;
461	28x	}
462
463	✗	void perform_io() noexcept override
464		{
465	✗	msghdr msg{};
466	✗	msg.msg_name = &dest_storage;
467	✗	msg.msg_namelen = dest_len;
468	✗	msg.msg_iov = iovecs;
469	✗	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
470
471		#ifdef MSG_NOSIGNAL
472	✗	int send_flags = msg_flags \| MSG_NOSIGNAL;
473		#else
474		int send_flags = msg_flags;
475		#endif
476
477		ssize_t n;
478		do
479		{
480	✗	n = ::sendmsg(this->fd, &msg, send_flags);
481		}
482	✗	while (n < 0 && errno == EINTR);
483
484	✗	if (n >= 0)
485	✗	this->complete(0, static_cast<std::size_t>(n));
486		else
487	✗	this->complete(errno, 0);
488	✗	}
489		};
490
491		/** Shared recv_from operation for datagram sockets.
492
493		Uses recvmsg() with msg_name to capture the source endpoint.
494
495		@tparam Base The backend's base op type.
496		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
497		*/
498		template<class Base, class Endpoint = endpoint>
499		struct reactor_recv_from_op : Base
500		{
501		/// Maximum scatter-gather buffer count.
502		static constexpr std::size_t max_buffers = 16;
503
504		/// Scatter-gather I/O vectors.
505		iovec iovecs[max_buffers];
506
507		/// Number of active I/O vectors.
508		int iovec_count = 0;
509
510		/// Source address storage filled by recvmsg.
511		sockaddr_storage source_storage{};
512
513		/// Actual source address length returned by recvmsg.
514		socklen_t source_addrlen = 0;
515
516		/// Output pointer for the source endpoint (set by do_recv_from).
517		Endpoint* source_out = nullptr;
518
519		/// User-supplied message flags.
520		int msg_flags = 0;
521
522		/// Return true (this is a read-direction operation).
523	✗	bool is_read_operation() const noexcept override
524		{
525	✗	return true;
526		}
527
528	40x	void reset() noexcept
529		{
530	40x	Base::reset();
531	40x	iovec_count = 0;
532	40x	source_storage = {};
533	40x	source_addrlen = 0;
534	40x	source_out = nullptr;
535	40x	msg_flags = 0;
536	40x	}
537
538	2x	void perform_io() noexcept override
539		{
540	2x	msghdr msg{};
541	2x	msg.msg_name = &source_storage;
542	2x	msg.msg_namelen = sizeof(source_storage);
543	2x	msg.msg_iov = iovecs;
544	2x	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
545
546		ssize_t n;
547		do
548		{
549	2x	n = ::recvmsg(this->fd, &msg, msg_flags);
550		}
551	2x	while (n < 0 && errno == EINTR);
552
553	2x	if (n >= 0)
554		{
555	2x	source_addrlen = msg.msg_namelen;
556	2x	this->complete(0, static_cast<std::size_t>(n));
557		}
558		else
559	✗	this->complete(errno, 0);
560	2x	}
561		};
562
563		} // namespace boost::corosio::detail
564
565		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_HPP
566