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

92.3% Lines (108/117) 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	126807x	complete_io_op(Op& op)
40		{
41	126807x	op.stop_cb.reset();
42	126807x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
43
44	126807x	if (op.cancelled.load(std::memory_order_acquire))
45	1050x	*op.ec_out = capy::error::canceled;
46	125757x	else if (op.errn != 0)
47	4x	*op.ec_out = make_err(op.errn);
48	125753x	else if (op.is_read_operation() && op.bytes_transferred == 0)
49	✗	*op.ec_out = capy::error::eof;
50		else
51	125753x	*op.ec_out = {};
52
53	126807x	*op.bytes_out = op.bytes_transferred;
54
55	126807x	op.cont_op.cont.h = op.h;
56	126807x	capy::executor_ref saved_ex(op.ex);
57	126807x	auto prevent = std::move(op.impl_ptr);
58	126807x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
59	126807x	}
60
61		/** Complete a datagram recv operation (connected mode).
62
63		Like complete_io_op but does not translate zero bytes into
64		EOF. Zero-length datagrams are valid and should be reported
65		as success with 0 bytes transferred.
66
67		@param op The operation to complete.
68		*/
69		template<typename Op>
70		void
71		complete_dgram_recv_op(Op& op)
72		{
73		op.stop_cb.reset();
74		op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
75
76		if (op.cancelled.load(std::memory_order_acquire))
77		*op.ec_out = capy::error::canceled;
78		else if (op.errn != 0)
79		*op.ec_out = make_err(op.errn);
80		else
81		*op.ec_out = {};
82
83		*op.bytes_out = op.bytes_transferred;
84
85		op.cont_op.cont.h = op.h;
86		capy::executor_ref saved_ex(op.ex);
87		auto prevent = std::move(op.impl_ptr);
88		dispatch_coro(saved_ex, op.cont_op.cont).resume();
89		}
90
91		/** Complete a connect operation with endpoint caching.
92
93		On success, queries the local endpoint via getsockname and
94		caches both endpoints in the socket impl. Then resumes the
95		caller via symmetric transfer.
96
97		@tparam Op The concrete connect operation type.
98		@param op The operation to complete.
99		*/
100		template<typename Op>
101		void
102	11733x	complete_connect_op(Op& op)
103		{
104	11733x	op.stop_cb.reset();
105	11733x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
106
107	11733x	bool success =
108	11733x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
109
110	11733x	if (success && op.socket_impl_)
111		{
112		using ep_type = decltype(op.target_endpoint);
113	11717x	ep_type local_ep;
114	11717x	sockaddr_storage local_storage{};
115	11717x	socklen_t local_len = sizeof(local_storage);
116	11717x	if (::getsockname(
117		op.fd, reinterpret_cast<sockaddr*>(&local_storage),
118	11717x	&local_len) == 0)
119	11713x	local_ep =
120	11717x	from_sockaddr_as(local_storage, local_len, ep_type{});
121	11717x	op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
122		}
123
124	11733x	if (op.cancelled.load(std::memory_order_acquire))
125	2x	*op.ec_out = capy::error::canceled;
126	11731x	else if (op.errn != 0)
127	14x	*op.ec_out = make_err(op.errn);
128		else
129	11717x	*op.ec_out = {};
130
131	11733x	op.cont_op.cont.h = op.h;
132	11733x	capy::executor_ref saved_ex(op.ex);
133	11733x	auto prevent = std::move(op.impl_ptr);
134	11733x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
135	11733x	}
136
137		/** Construct and register a peer socket from an accepted fd.
138
139		Creates a new socket impl via the acceptor's associated
140		socket service, registers it with the scheduler, and caches
141		the local and remote endpoints.
142
143		@tparam SocketImpl The concrete socket implementation type.
144		@tparam AcceptorImpl The concrete acceptor implementation type.
145		@param acceptor_impl The acceptor that accepted the connection.
146		@param accepted_fd The accepted file descriptor (set to -1 on success).
147		@param peer_storage The peer address from accept().
148		@param impl_out Output pointer for the new socket impl.
149		@param ec_out Output pointer for any error.
150		@return True on success, false on failure.
151		*/
152		template<typename SocketImpl, typename AcceptorImpl>
153		bool
154	11707x	setup_accepted_socket(
155		AcceptorImpl* acceptor_impl,
156		int& accepted_fd,
157		sockaddr_storage const& peer_storage,
158		socklen_t peer_addrlen,
159		io_object::implementation** impl_out,
160		std::error_code* ec_out)
161		{
162	11707x	auto* socket_svc = acceptor_impl->service().stream_service();
163	11707x	if (!socket_svc)
164		{
165	✗	*ec_out = make_err(ENOENT);
166	✗	return false;
167		}
168
169	11707x	auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
170	11707x	impl.set_socket(accepted_fd);
171
172	11707x	impl.desc_state_.fd = accepted_fd;
173		{
174	11707x	std::lock_guard lock(impl.desc_state_.mutex);
175	11707x	impl.desc_state_.read_op = nullptr;
176	11707x	impl.desc_state_.write_op = nullptr;
177	11707x	impl.desc_state_.connect_op = nullptr;
178	11707x	}
179	11707x	socket_svc->scheduler().register_descriptor(accepted_fd, &impl.desc_state_);
180
181		using ep_type = decltype(acceptor_impl->local_endpoint());
182	11707x	impl.set_endpoints(
183		acceptor_impl->local_endpoint(),
184	11707x	from_sockaddr_as(
185		peer_storage,
186		peer_addrlen,
187		ep_type{}));
188
189	11707x	if (impl_out)
190	11707x	*impl_out = &impl;
191	11707x	accepted_fd = -1;
192	11707x	return true;
193		}
194
195		/** Complete an accept operation.
196
197		Sets up the peer socket on success, or closes the accepted
198		fd on failure. Then resumes the caller via symmetric transfer.
199
200		@tparam SocketImpl The concrete socket implementation type.
201		@tparam Op The concrete accept operation type.
202		@param op The operation to complete.
203		*/
204		template<typename SocketImpl, typename Op>
205		void
206	11719x	complete_accept_op(Op& op)
207		{
208	11719x	op.stop_cb.reset();
209	11719x	op.acceptor_impl_->desc_state_.scheduler_->reset_inline_budget();
210
211	11719x	bool success =
212	11719x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
213
214	11719x	if (op.cancelled.load(std::memory_order_acquire))
215	12x	*op.ec_out = capy::error::canceled;
216	11707x	else if (op.errn != 0)
217	✗	*op.ec_out = make_err(op.errn);
218		else
219	11707x	*op.ec_out = {};
220
221	11719x	if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
222		{
223	11707x	if (!setup_accepted_socket<SocketImpl>(
224	11707x	op.acceptor_impl_, op.accepted_fd, op.peer_storage,
225		op.peer_addrlen, op.impl_out, op.ec_out))
226	✗	success = false;
227		}
228
229	11719x	if (!success \|\| !op.acceptor_impl_)
230		{
231	12x	if (op.accepted_fd >= 0)
232		{
233	✗	::close(op.accepted_fd);
234	✗	op.accepted_fd = -1;
235		}
236	12x	if (op.impl_out)
237	12x	*op.impl_out = nullptr;
238		}
239
240	11719x	op.cont_op.cont.h = op.h;
241	11719x	capy::executor_ref saved_ex(op.ex);
242	11719x	auto prevent = std::move(op.impl_ptr);
243	11719x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
244	11719x	}
245
246		/** Complete a datagram operation (send_to or recv_from).
247
248		For recv_from operations, writes the source endpoint from the
249		recorded sockaddr_storage into the caller's endpoint pointer.
250		Then resumes the caller via symmetric transfer.
251
252		@tparam Op The concrete datagram operation type.
253		@param op The operation to complete.
254		*/
255		template<typename Op>
256		void
257	6x	complete_datagram_op(Op& op)
258		{
259	6x	op.stop_cb.reset();
260	6x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
261
262	6x	if (op.cancelled.load(std::memory_order_acquire))
263	2x	*op.ec_out = capy::error::canceled;
264	4x	else if (op.errn != 0)
265	✗	*op.ec_out = make_err(op.errn);
266		else
267	4x	*op.ec_out = {};
268
269	6x	*op.bytes_out = op.bytes_transferred;
270
271	6x	op.cont_op.cont.h = op.h;
272	6x	capy::executor_ref saved_ex(op.ex);
273	6x	auto prevent = std::move(op.impl_ptr);
274	6x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
275	6x	}
276
277		/** Complete a datagram operation with source endpoint capture.
278
279		For recv_from operations, writes the source endpoint from the
280		recorded sockaddr_storage into the caller's endpoint pointer.
281		Then resumes the caller via symmetric transfer.
282
283		@tparam Op The concrete datagram operation type.
284		@param op The operation to complete.
285		@param source_out Optional pointer to store source endpoint
286		(non-null for recv_from, null for send_to).
287		*/
288		template<typename Op, typename Endpoint>
289		void
290	18x	complete_datagram_op(Op& op, Endpoint* source_out)
291		{
292	18x	op.stop_cb.reset();
293	18x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
294
295	18x	if (op.cancelled.load(std::memory_order_acquire))
296	6x	*op.ec_out = capy::error::canceled;
297	12x	else if (op.errn != 0)
298	✗	*op.ec_out = make_err(op.errn);
299		else
300	12x	*op.ec_out = {};
301
302	18x	*op.bytes_out = op.bytes_transferred;
303
304	28x	if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
305	10x	op.errn == 0)
306	20x	*source_out = from_sockaddr_as(
307	10x	op.source_storage,
308		op.source_addrlen,
309		Endpoint{});
310
311	18x	op.cont_op.cont.h = op.h;
312	18x	capy::executor_ref saved_ex(op.ex);
313	18x	auto prevent = std::move(op.impl_ptr);
314	18x	dispatch_coro(saved_ex, op.cont_op.cont).resume();
315	18x	}
316
317		} // namespace boost::corosio::detail
318
319		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
320