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

80.4% Lines (502/624) 93.8% List of functions (30/32)

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

reactor_op_complete.hpp

f(x) Functions (32)

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_COMPLETE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
12
13		#include <boost/corosio/detail/dispatch_coro.hpp>
14		#include <boost/corosio/native/detail/endpoint_convert.hpp>
15		#include <boost/corosio/native/detail/make_err.hpp>
16		#include <boost/corosio/io/io_object.hpp>
17
18		#include <coroutine>
19		#include <mutex>
20		#include <utility>
21
22		#include <netinet/in.h>
23		#include <sys/socket.h>
24		#include <unistd.h>
25
26		namespace boost::corosio::detail {
27
28		/** Complete a base read/write operation.
29
30		Translates the recorded errno and cancellation state into
31		an error_code, stores the byte count, then resumes the
32		caller via symmetric transfer.
33
34		@tparam Op The concrete operation type.
35		@param op The operation to complete.
36		*/
37		template<typename Op>
38		void
39	79153x	complete_io_op(Op& op)
40		{
41	79153x	op.stop_cb.reset();
42	79153x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
43
44	79153x	if (op.cancelled.load(std::memory_order_acquire))
45	305x	*op.ec_out = capy::error::canceled;
46	78848x	else if (op.errn != 0)
47	✗	*op.ec_out = make_err(op.errn);
48	78848x	else if (op.is_read_operation() && op.bytes_transferred == 0)
49	✗	*op.ec_out = capy::error::eof;
50		else
51	78848x	*op.ec_out = {};
52
53	79153x	*op.bytes_out = op.bytes_transferred;
54
55	79153x	op.cont_op.cont.h = op.h;
56	79153x	capy::executor_ref saved_ex(op.ex);
57	79153x	auto prevent = std::move(op.impl_ptr);
58	79153x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
59	79153x	}
60
61		/** Complete a connect operation with endpoint caching.
62
63		On success, queries the local endpoint via getsockname and
64		caches both endpoints in the socket impl. Then resumes the
65		caller via symmetric transfer.
66
67		@tparam Op The concrete connect operation type.
68		@param op The operation to complete.
69		*/
70		template<typename Op>
71		void
72	8001x	complete_connect_op(Op& op)
73		{
74	8001x	op.stop_cb.reset();
75	8001x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
76
77	8001x	bool success =
78	8001x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
79
80	8001x	if (success && op.socket_impl_)
81		{
82		using ep_type = decltype(op.target_endpoint);
83	7997x	ep_type local_ep;
84	7997x	sockaddr_storage local_storage{};
85	7997x	socklen_t local_len = sizeof(local_storage);
86	7997x	if (::getsockname(
87		op.fd, reinterpret_cast<sockaddr*>(&local_storage),
88	7997x	&local_len) == 0)
89	7993x	local_ep =
90	7997x	from_sockaddr_as(local_storage, local_len, ep_type{});
91	7997x	op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
92		}
93
94	8001x	if (op.cancelled.load(std::memory_order_acquire))
95	✗	*op.ec_out = capy::error::canceled;
96	8001x	else if (op.errn != 0)
97	4x	*op.ec_out = make_err(op.errn);
98		else
99	7997x	*op.ec_out = {};
100
101	8001x	op.cont_op.cont.h = op.h;
102	8001x	capy::executor_ref saved_ex(op.ex);
103	8001x	auto prevent = std::move(op.impl_ptr);
104	8001x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
105	8001x	}
106
107		/** Construct and register a peer socket from an accepted fd.
108
109		Creates a new socket impl via the acceptor's associated
110		socket service, registers it with the scheduler, and caches
111		the local and remote endpoints.
112
113		@tparam SocketImpl The concrete socket implementation type.
114		@tparam AcceptorImpl The concrete acceptor implementation type.
115		@param acceptor_impl The acceptor that accepted the connection.
116		@param accepted_fd The accepted file descriptor (set to -1 on success).
117		@param peer_storage The peer address from accept().
118		@param peer_addrlen The actual peer address length from accept().
119		@param impl_out Output pointer for the new socket impl.
120		@param ec_out Output pointer for any error.
121		@return True on success, false on failure.
122		*/
123		template<typename SocketImpl, typename AcceptorImpl>
124		bool
125	7983x	setup_accepted_socket(
126		AcceptorImpl* acceptor_impl,
127		int& accepted_fd,
128		sockaddr_storage const& peer_storage,
129		socklen_t peer_addrlen,
130		io_object::implementation** impl_out,
131		std::error_code* ec_out)
132		{
133	7983x	auto* socket_svc = acceptor_impl->service().stream_service();
134	7983x	if (!socket_svc)
135		{
136	✗	*ec_out = make_err(ENOENT);
137	✗	return false;
138		}
139
140	7983x	auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
141	7983x	impl.set_socket(accepted_fd);
142
143	7983x	impl.desc_state_.fd = accepted_fd;
144		{
145	7983x	std::lock_guard lock(impl.desc_state_.mutex);
146	7983x	impl.desc_state_.read_op = nullptr;
147	7983x	impl.desc_state_.write_op = nullptr;
148	7983x	impl.desc_state_.connect_op = nullptr;
149	7983x	}
150	7983x	socket_svc->scheduler().register_descriptor(accepted_fd, &impl.desc_state_);
151
152		using ep_type = decltype(acceptor_impl->local_endpoint());
153	7983x	impl.set_endpoints(
154		acceptor_impl->local_endpoint(),
155	7983x	from_sockaddr_as(peer_storage, peer_addrlen, ep_type{}));
156
157	7983x	if (impl_out)
158	7983x	*impl_out = &impl;
159	7983x	accepted_fd = -1;
160	7983x	return true;
161		}
162
163		/** Complete an accept operation.
164
165		Sets up the peer socket on success, or closes the accepted
166		fd on failure. Then resumes the caller via symmetric transfer.
167
168		@tparam SocketImpl The concrete socket implementation type.
169		@tparam Op The concrete accept operation type.
170		@param op The operation to complete.
171		*/
172		template<typename SocketImpl, typename Op>
173		void
174	7995x	complete_accept_op(Op& op)
175		{
176	7995x	op.stop_cb.reset();
177	7995x	op.acceptor_impl_->desc_state_.scheduler_->reset_inline_budget();
178
179	7995x	bool success =
180	7995x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
181
182	7995x	if (op.cancelled.load(std::memory_order_acquire))
183	12x	*op.ec_out = capy::error::canceled;
184	7983x	else if (op.errn != 0)
185	✗	*op.ec_out = make_err(op.errn);
186		else
187	7983x	*op.ec_out = {};
188
189	7995x	if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
190		{
191	7983x	if (!setup_accepted_socket<SocketImpl>(
192	7983x	op.acceptor_impl_, op.accepted_fd, op.peer_storage,
193		op.peer_addrlen, op.impl_out, op.ec_out))
194	✗	success = false;
195		}
196
197	7995x	if (!success \|\| !op.acceptor_impl_)
198		{
199	12x	if (op.accepted_fd >= 0)
200		{
201	✗	::close(op.accepted_fd);
202	✗	op.accepted_fd = -1;
203		}
204	12x	if (op.impl_out)
205	12x	*op.impl_out = nullptr;
206		}
207
208	7995x	op.cont_op.cont.h = op.h;
209	7995x	capy::executor_ref saved_ex(op.ex);
210	7995x	auto prevent = std::move(op.impl_ptr);
211	7995x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
212	7995x	}
213
214		/** Complete a connected datagram operation (send or recv).
215
216		No source endpoint to capture. Critically, does NOT map
217		bytes_transferred == 0 to EOF — zero-length datagrams are valid
218		events on connected datagram sockets.
219
220		@tparam Op The concrete datagram operation type.
221		@param op The operation to complete.
222		*/
223		template<typename Op>
224		void
225	10x	complete_datagram_op(Op& op)
226		{
227	10x	op.stop_cb.reset();
228	10x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
229
230	10x	if (op.cancelled.load(std::memory_order_acquire))
231	2x	*op.ec_out = capy::error::canceled;
232	8x	else if (op.errn != 0)
233	✗	*op.ec_out = make_err(op.errn);
234		else
235	8x	*op.ec_out = {};
236
237	10x	*op.bytes_out = op.bytes_transferred;
238
239	10x	op.cont_op.cont.h = op.h;
240	10x	capy::executor_ref saved_ex(op.ex);
241	10x	auto prevent = std::move(op.impl_ptr);
242	10x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
243	10x	}
244
245		/** Complete a datagram operation with source endpoint capture.
246
247		For recv_from operations, writes the source endpoint from the
248		recorded sockaddr_storage into the caller's endpoint pointer.
249		Then resumes the caller via symmetric transfer.
250
251		@tparam Op The concrete datagram operation type.
252		@param op The operation to complete.
253		@param source_out Optional pointer to store source endpoint
254		(non-null for recv_from, null for send_to).
255		*/
256		template<typename Op, typename Endpoint>
257		void
258	18x	complete_datagram_op(Op& op, Endpoint* source_out)
259		{
260	18x	op.stop_cb.reset();
261	18x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
262
263	18x	if (op.cancelled.load(std::memory_order_acquire))
264	6x	*op.ec_out = capy::error::canceled;
265	12x	else if (op.errn != 0)
266	✗	*op.ec_out = make_err(op.errn);
267		else
268	12x	*op.ec_out = {};
269
270	18x	*op.bytes_out = op.bytes_transferred;
271
272	28x	if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
273	10x	op.errn == 0)
274	20x	*source_out = from_sockaddr_as(
275	10x	op.source_storage,
276		op.source_addrlen,
277		Endpoint{});
278
279	18x	op.cont_op.cont.h = op.h;
280	18x	capy::executor_ref saved_ex(op.ex);
281	18x	auto prevent = std::move(op.impl_ptr);
282	18x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
283	18x	}
284
285		} // namespace boost::corosio::detail
286
287		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
288