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

79.9% Lines (454/568) 92.9% List of functions (26/28)

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

reactor_op_complete.hpp

f(x) Functions (28)

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	80217x	complete_io_op(Op& op)
40		{
41	80217x	op.stop_cb.reset();
42	80217x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
43
44	80217x	if (op.cancelled.load(std::memory_order_acquire))
45	306x	*op.ec_out = capy::error::canceled;
46	79911x	else if (op.errn != 0)
47	✗	*op.ec_out = make_err(op.errn);
48	79911x	else if (op.is_read_operation() && op.bytes_transferred == 0)
49	✗	*op.ec_out = capy::error::eof;
50		else
51	79911x	*op.ec_out = {};
52
53	80217x	*op.bytes_out = op.bytes_transferred;
54
55	80217x	op.cont_op.cont.h = op.h;
56	80217x	capy::executor_ref saved_ex(op.ex);
57	80217x	auto prevent = std::move(op.impl_ptr);
58	80217x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
59	80217x	}
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	5056x	complete_connect_op(Op& op)
73		{
74	5056x	op.stop_cb.reset();
75	5056x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
76
77	5056x	bool success =
78	5056x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
79
80	5056x	if (success && op.socket_impl_)
81		{
82		using ep_type = decltype(op.target_endpoint);
83	5051x	ep_type local_ep;
84	5051x	sockaddr_storage local_storage{};
85	5051x	socklen_t local_len = sizeof(local_storage);
86	5051x	if (::getsockname(
87		op.fd, reinterpret_cast<sockaddr*>(&local_storage),
88	5051x	&local_len) == 0)
89	5047x	local_ep =
90	5051x	from_sockaddr_as(local_storage, local_len, ep_type{});
91	5051x	op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
92		}
93
94	5056x	if (op.cancelled.load(std::memory_order_acquire))
95	✗	*op.ec_out = capy::error::canceled;
96	5056x	else if (op.errn != 0)
97	5x	*op.ec_out = make_err(op.errn);
98		else
99	5051x	*op.ec_out = {};
100
101	5056x	op.cont_op.cont.h = op.h;
102	5056x	capy::executor_ref saved_ex(op.ex);
103	5056x	auto prevent = std::move(op.impl_ptr);
104	5056x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
105	5056x	}
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	5041x	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	5041x	auto* socket_svc = acceptor_impl->service().stream_service();
134	5041x	if (!socket_svc)
135		{
136	✗	*ec_out = make_err(ENOENT);
137	✗	return false;
138		}
139
140	5041x	auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
141	5041x	impl.set_socket(accepted_fd);
142
143	5041x	impl.desc_state_.fd = accepted_fd;
144		{
145	5041x	std::lock_guard lock(impl.desc_state_.mutex);
146	5041x	impl.desc_state_.read_op = nullptr;
147	5041x	impl.desc_state_.write_op = nullptr;
148	5041x	impl.desc_state_.connect_op = nullptr;
149	5041x	}
150	5041x	socket_svc->scheduler().register_descriptor(accepted_fd, &impl.desc_state_);
151
152		using ep_type = decltype(acceptor_impl->local_endpoint());
153	5041x	impl.set_endpoints(
154		acceptor_impl->local_endpoint(),
155	5041x	from_sockaddr_as(peer_storage, peer_addrlen, ep_type{}));
156
157	5041x	if (impl_out)
158	5041x	*impl_out = &impl;
159	5041x	accepted_fd = -1;
160	5041x	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	5053x	complete_accept_op(Op& op)
175		{
176	5053x	op.stop_cb.reset();
177	5053x	op.acceptor_impl_->desc_state_.scheduler_->reset_inline_budget();
178
179	5053x	bool success =
180	5053x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
181
182	5053x	if (op.cancelled.load(std::memory_order_acquire))
183	12x	*op.ec_out = capy::error::canceled;
184	5041x	else if (op.errn != 0)
185	✗	*op.ec_out = make_err(op.errn);
186		else
187	5041x	*op.ec_out = {};
188
189	5053x	if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
190		{
191	5041x	if (!setup_accepted_socket<SocketImpl>(
192	5041x	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	5053x	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	5053x	op.cont_op.cont.h = op.h;
209	5053x	capy::executor_ref saved_ex(op.ex);
210	5053x	auto prevent = std::move(op.impl_ptr);
211	5053x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
212	5053x	}
213
214		/** Complete a datagram operation (send_to or recv_from).
215
216		For recv_from operations, writes the source endpoint from the
217		recorded sockaddr_storage into the caller's endpoint pointer.
218		Then resumes the caller via symmetric transfer.
219
220		@tparam Op The concrete datagram operation type.
221		@param op The operation to complete.
222		@param source_out Optional pointer to store source endpoint
223		(non-null for recv_from, null for send_to).
224		*/
225		template<typename Op, typename Endpoint>
226		void
227	18x	complete_datagram_op(Op& op, Endpoint* source_out)
228		{
229	18x	op.stop_cb.reset();
230	18x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
231
232	18x	if (op.cancelled.load(std::memory_order_acquire))
233	6x	*op.ec_out = capy::error::canceled;
234	12x	else if (op.errn != 0)
235	✗	*op.ec_out = make_err(op.errn);
236		else
237	12x	*op.ec_out = {};
238
239	18x	*op.bytes_out = op.bytes_transferred;
240
241	28x	if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
242	10x	op.errn == 0)
243	20x	*source_out = from_sockaddr_as(
244	10x	op.source_storage,
245		op.source_addrlen,
246		Endpoint{});
247
248	18x	op.cont_op.cont.h = op.h;
249	18x	capy::executor_ref saved_ex(op.ex);
250	18x	auto prevent = std::move(op.impl_ptr);
251	18x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
252	18x	}
253
254		} // namespace boost::corosio::detail
255
256		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
257