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

98.1% Lines (105 / 107) 100.0% Functions (44 / 44)

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

reactor_op_complete.hpp

f(x) Functions (44)

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/coro_op_complete.hpp>
15		#include <boost/corosio/native/detail/endpoint_convert.hpp>
16		#include <boost/corosio/native/detail/make_err.hpp>
17		#include <boost/corosio/io/io_object.hpp>
18
19		#include <coroutine>
20		#include <mutex>
21		#include <utility>
22
23		#include <netinet/in.h>
24		#include <sys/socket.h>
25		#include <unistd.h>
26
27		namespace boost::corosio::detail {
28
29		/** Complete a base read/write operation.
30
31		Translates the recorded errno and cancellation state into
32		an error_code, stores the byte count, then resumes the
33		caller via symmetric transfer.
34
35		@tparam Op The concrete operation type.
36		@param op The operation to complete.
37		*/
38		template<typename Op>
39		void
40	83816x	complete_io_op(Op& op)
41		{
42	83816x	op.stop_cb.reset();
43		// scheduler_ is null until the descriptor is registered; an op
44		// completed by the closed-object entry check never registered and
45		// has no budget to reset.
46	83816x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
47	83798x	sched->reset_inline_budget();
48
49		// is_read_operation() already folds in the empty-buffer case (it
50		// returns false for a zero-length read), so empty_buffer stays false
51		// here and the shared EOF test reduces to the reactor's original
52		// `is_read && bytes == 0`.
53	167218x	decode_results(
54	83816x	op.ec_out, op.bytes_out, op.cancelled.load(std::memory_order_acquire),
55	83816x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
56	83816x	op.is_read_operation(), op.bytes_transferred, /empty_buffer=/false);
57
58	83816x	coro_resume(&op);
59	83816x	}
60
61		/** Complete a wait operation.
62
63		Wait operations report only an error_code — no bytes_transferred,
64		no EOF translation. Used for socket and acceptor wait() awaitables;
65		picks the impl pointer set by start() to reach the scheduler.
66
67		@tparam Op The concrete wait operation type.
68		@param op The operation to complete.
69		*/
70		template<typename Op>
71		void
72	199x	complete_wait_op(Op& op)
73		{
74	199x	op.stop_cb.reset();
75		// scheduler_ is null until the descriptor is registered; a wait
76		// completed by the initiation probe (e.g. EBADF on a never-opened
77		// socket) has no registration to reset a budget for.
78	199x	if (op.socket_impl_)
79		{
80	161x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
81	147x	sched->reset_inline_budget();
82		}
83	38x	else if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
84		{
85	34x	sched->reset_inline_budget();
86		}
87
88		// Wait reports only success/cancel/error — no bytes, no EOF.
89	362x	decode_results(
90		op.ec_out, /bytes_out=/nullptr,
91	199x	op.cancelled.load(std::memory_order_acquire),
92	199x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
93		/is_read=/false, /bytes=/0, /empty_buffer=/false);
94
95	199x	coro_resume(&op);
96	199x	}
97
98		/** Complete a connect operation with endpoint caching.
99
100		On success, queries the local endpoint via getsockname and
101		caches both endpoints in the socket impl. Then resumes the
102		caller via symmetric transfer.
103
104		@tparam Op The concrete connect operation type.
105		@param op The operation to complete.
106		*/
107		template<typename Op>
108		void
109	11549x	complete_connect_op(Op& op)
110		{
111	11549x	op.stop_cb.reset();
112	11549x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
113
114	11549x	bool success =
115	11549x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
116
117	11549x	if (success && op.socket_impl_)
118		{
119		using ep_type = decltype(op.target_endpoint);
120	11400x	ep_type local_ep;
121	11400x	sockaddr_storage local_storage{};
122	11400x	socklen_t local_len = sizeof(local_storage);
123	11400x	if (::getsockname(
124		op.fd, reinterpret_cast<sockaddr*>(&local_storage),
125	11400x	&local_len) == 0)
126	11400x	local_ep = from_sockaddr_as(local_storage, local_len, ep_type{});
127	11400x	op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
128		}
129
130	22969x	decode_results(
131		op.ec_out, /bytes_out=/nullptr,
132	11549x	op.cancelled.load(std::memory_order_acquire),
133	11549x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
134		/is_read=/false, /bytes=/0, /empty_buffer=/false);
135
136	11549x	coro_resume(&op);
137	11549x	}
138
139		/** Construct and register a peer socket from an accepted fd.
140
141		Creates a new socket impl via the acceptor's associated
142		socket service, registers it with the scheduler, and caches
143		the local and remote endpoints.
144
145		@tparam SocketImpl The concrete socket implementation type.
146		@tparam AcceptorImpl The concrete acceptor implementation type.
147		@param acceptor_impl The acceptor that accepted the connection.
148		@param accepted_fd The accepted file descriptor. Cleared to -1
149		once the socket impl owns it, which includes the registration
150		failure that destroys the impl and closes the fd with it.
151		@param peer_storage The peer address from accept().
152		@param impl_out Output pointer for the new socket impl.
153		@param ec_out Output pointer for any error.
154		@return True on success, false on failure.
155		*/
156		template<typename SocketImpl, typename AcceptorImpl>
157		bool
158	8751x	setup_accepted_socket(
159		AcceptorImpl* acceptor_impl,
160		int& accepted_fd,
161		sockaddr_storage const& peer_storage,
162		socklen_t peer_addrlen,
163		io_object::implementation** impl_out,
164		std::error_code* ec_out)
165		{
166	8751x	auto* socket_svc = acceptor_impl->service().stream_service();
167	8751x	if (!socket_svc)
168		{
169	✗	*ec_out = make_err(ENOENT);
170	✗	return false;
171		}
172
173	8751x	auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
174	17502x	impl.set_socket(
175	8751x	accepted_fd, peer_addrlen > 0 ? peer_storage.ss_family : AF_UNSPEC);
176
177	8751x	impl.desc_state_.fd = accepted_fd;
178		{
179	8751x	std::lock_guard lock(impl.desc_state_.mutex);
180	8751x	impl.desc_state_.read_op = nullptr;
181	8751x	impl.desc_state_.write_op = nullptr;
182	8751x	impl.desc_state_.connect_op = nullptr;
183	8751x	}
184	8751x	if (auto ec = socket_svc->scheduler().register_descriptor(
185		accepted_fd, &impl.desc_state_))
186		{
187		// destroy() closes the fd the impl already owns.
188	1x	accepted_fd = -1;
189	1x	socket_svc->destroy(&impl);
190	1x	*ec_out = ec;
191	1x	return false;
192		}
193
194		using ep_type = decltype(acceptor_impl->local_endpoint());
195	8750x	impl.set_endpoints(
196		acceptor_impl->local_endpoint(),
197	8750x	from_sockaddr_as(peer_storage, peer_addrlen, ep_type{}));
198
199	8750x	if (impl_out)
200	8750x	*impl_out = &impl;
201	8750x	accepted_fd = -1;
202	8750x	return true;
203		}
204
205		/** Complete an accept operation.
206
207		Sets up the peer socket on success, or closes the accepted
208		fd on failure. Then resumes the caller via symmetric transfer.
209
210		@tparam SocketImpl The concrete socket implementation type.
211		@tparam Op The concrete accept operation type.
212		@param op The operation to complete.
213		*/
214		template<typename SocketImpl, typename Op>
215		void
216	8926x	complete_accept_op(Op& op)
217		{
218	8926x	op.stop_cb.reset();
219	8926x	if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
220	8922x	sched->reset_inline_budget();
221
222	8926x	bool success =
223	8926x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
224
225	17841x	decode_results(
226		op.ec_out, /bytes_out=/nullptr,
227	8926x	op.cancelled.load(std::memory_order_acquire),
228	8926x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
229		/is_read=/false, /bytes=/0, /empty_buffer=/false);
230
231	8926x	if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
232		{
233	8751x	if (!setup_accepted_socket<SocketImpl>(
234	8751x	op.acceptor_impl_, op.accepted_fd, op.peer_storage,
235		op.peer_addrlen, op.impl_out, op.ec_out))
236	1x	success = false;
237		}
238
239	8926x	if (!success \|\| !op.acceptor_impl_)
240		{
241	176x	if (op.accepted_fd >= 0)
242		{
243	5x	::close(op.accepted_fd);
244	5x	op.accepted_fd = -1;
245		}
246	176x	if (op.impl_out)
247	176x	*op.impl_out = nullptr;
248		}
249
250	8926x	coro_resume(&op);
251	8926x	}
252
253		/** Complete a datagram operation (send_to or recv_from).
254
255		For recv_from operations, writes the source endpoint from the
256		recorded sockaddr_storage into the caller's endpoint pointer.
257		Then resumes the caller via symmetric transfer.
258
259		@tparam Op The concrete datagram operation type.
260		@param op The operation to complete.
261		*/
262		template<typename Op>
263		void
264	112x	complete_datagram_op(Op& op)
265		{
266	112x	op.stop_cb.reset();
267	112x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
268
269		// No EOF: a zero-length datagram is valid (success with 0 bytes).
270	220x	decode_results(
271	112x	op.ec_out, op.bytes_out, op.cancelled.load(std::memory_order_acquire),
272	112x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
273		/is_read=/false, op.bytes_transferred, /empty_buffer=/false);
274
275	112x	coro_resume(&op);
276	112x	}
277
278		/** Complete a datagram operation with source endpoint capture.
279
280		For recv_from operations, writes the source endpoint from the
281		recorded sockaddr_storage into the caller's endpoint pointer.
282		Then resumes the caller via symmetric transfer.
283
284		@tparam Op The concrete datagram operation type.
285		@param op The operation to complete.
286		@param source_out Optional pointer to store source endpoint
287		(non-null for recv_from, null for send_to).
288		*/
289		template<typename Op, typename Endpoint>
290		void
291	89x	complete_datagram_op(Op& op, Endpoint* source_out)
292		{
293	89x	op.stop_cb.reset();
294	89x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
295
296		// No EOF: a zero-length datagram is valid (success with 0 bytes).
297	176x	decode_results(
298	89x	op.ec_out, op.bytes_out, op.cancelled.load(std::memory_order_acquire),
299	89x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
300		/is_read=/false, op.bytes_transferred, /empty_buffer=/false);
301
302		// Write the source exactly when the decode reported success — a
303		// transfer outranks a raced cancellation flag there too.
304	172x	if (source_out && op.errn == 0 &&
305	83x	(op.bytes_transferred > 0 \|\|
306	18x	!op.cancelled.load(std::memory_order_acquire)))
307	26x	*source_out =
308	65x	from_sockaddr_as(op.source_storage, op.source_addrlen, Endpoint{});
309
310	89x	coro_resume(&op);
311	89x	}
312
313		} // namespace boost::corosio::detail
314
315		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
316