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

96.2% Lines (102 / 106) 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	82991x	complete_io_op(Op& op)
41		{
42	82991x	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	82991x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
47	82973x	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	165935x	decode_io_result(
54	82991x	op.ec_out, op.bytes_out, op.cancelled.load(std::memory_order_acquire),
55	82991x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
56	82991x	op.is_read_operation(), op.bytes_transferred, /empty_buffer=/false);
57
58	82991x	coro_resume(&op);
59	82991x	}
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	167x	complete_wait_op(Op& op)
73		{
74	167x	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	167x	if (op.socket_impl_)
79		{
80	132x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
81	118x	sched->reset_inline_budget();
82		}
83	35x	else if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
84		{
85	31x	sched->reset_inline_budget();
86		}
87
88		// Wait reports only success/cancel/error — no bytes, no EOF.
89	300x	decode_io_result(
90		op.ec_out, /bytes_out=/nullptr,
91	167x	op.cancelled.load(std::memory_order_acquire),
92	167x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
93		/is_read=/false, /bytes=/0, /empty_buffer=/false);
94
95	167x	coro_resume(&op);
96	167x	}
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	4587x	complete_connect_op(Op& op)
110		{
111	4587x	op.stop_cb.reset();
112	4587x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
113
114	4587x	bool success =
115	4587x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
116
117	4587x	if (success && op.socket_impl_)
118		{
119		using ep_type = decltype(op.target_endpoint);
120	4539x	ep_type local_ep;
121	4539x	sockaddr_storage local_storage{};
122	4539x	socklen_t local_len = sizeof(local_storage);
123	4539x	if (::getsockname(
124		op.fd, reinterpret_cast<sockaddr*>(&local_storage),
125	4539x	&local_len) == 0)
126	4539x	local_ep = from_sockaddr_as(local_storage, local_len, ep_type{});
127	4539x	op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
128		}
129
130	9130x	decode_io_result(
131		op.ec_out, /bytes_out=/nullptr,
132	4587x	op.cancelled.load(std::memory_order_acquire),
133	4587x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
134		/is_read=/false, /bytes=/0, /empty_buffer=/false);
135
136	4587x	coro_resume(&op);
137	4587x	}
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	4483x	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	4483x	auto* socket_svc = acceptor_impl->service().stream_service();
167	4483x	if (!socket_svc)
168		{
169	✗	*ec_out = make_err(ENOENT);
170	✗	return false;
171		}
172
173	4483x	auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
174	4483x	impl.set_socket(accepted_fd);
175
176	4483x	impl.desc_state_.fd = accepted_fd;
177		{
178	4483x	std::lock_guard lock(impl.desc_state_.mutex);
179	4483x	impl.desc_state_.read_op = nullptr;
180	4483x	impl.desc_state_.write_op = nullptr;
181	4483x	impl.desc_state_.connect_op = nullptr;
182	4483x	}
183	4483x	if (auto ec = socket_svc->scheduler().register_descriptor(
184		accepted_fd, &impl.desc_state_))
185		{
186		// destroy() closes the fd the impl already owns.
187	1x	accepted_fd = -1;
188	1x	socket_svc->destroy(&impl);
189	1x	*ec_out = ec;
190	1x	return false;
191		}
192
193		using ep_type = decltype(acceptor_impl->local_endpoint());
194	4482x	impl.set_endpoints(
195		acceptor_impl->local_endpoint(),
196	4482x	from_sockaddr_as(peer_storage, peer_addrlen, ep_type{}));
197
198	4482x	if (impl_out)
199	4482x	*impl_out = &impl;
200	4482x	accepted_fd = -1;
201	4482x	return true;
202		}
203
204		/** Complete an accept operation.
205
206		Sets up the peer socket on success, or closes the accepted
207		fd on failure. Then resumes the caller via symmetric transfer.
208
209		@tparam SocketImpl The concrete socket implementation type.
210		@tparam Op The concrete accept operation type.
211		@param op The operation to complete.
212		*/
213		template<typename SocketImpl, typename Op>
214		void
215	4582x	complete_accept_op(Op& op)
216		{
217	4582x	op.stop_cb.reset();
218	4582x	if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
219	4578x	sched->reset_inline_budget();
220
221	4582x	bool success =
222	4582x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
223
224	9153x	decode_io_result(
225		op.ec_out, /bytes_out=/nullptr,
226	4582x	op.cancelled.load(std::memory_order_acquire),
227	4582x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
228		/is_read=/false, /bytes=/0, /empty_buffer=/false);
229
230	4582x	if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
231		{
232	4483x	if (!setup_accepted_socket<SocketImpl>(
233	4483x	op.acceptor_impl_, op.accepted_fd, op.peer_storage,
234		op.peer_addrlen, op.impl_out, op.ec_out))
235	1x	success = false;
236		}
237
238	4582x	if (!success \|\| !op.acceptor_impl_)
239		{
240	100x	if (op.accepted_fd >= 0)
241		{
242	✗	::close(op.accepted_fd);
243	✗	op.accepted_fd = -1;
244		}
245	100x	if (op.impl_out)
246	100x	*op.impl_out = nullptr;
247		}
248
249	4582x	coro_resume(&op);
250	4582x	}
251
252		/** Complete a datagram operation (send_to or recv_from).
253
254		For recv_from operations, writes the source endpoint from the
255		recorded sockaddr_storage into the caller's endpoint pointer.
256		Then resumes the caller via symmetric transfer.
257
258		@tparam Op The concrete datagram operation type.
259		@param op The operation to complete.
260		*/
261		template<typename Op>
262		void
263	101x	complete_datagram_op(Op& op)
264		{
265	101x	op.stop_cb.reset();
266	101x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
267
268		// No EOF: a zero-length datagram is valid (success with 0 bytes).
269	198x	decode_io_result(
270	101x	op.ec_out, op.bytes_out, op.cancelled.load(std::memory_order_acquire),
271	101x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
272		/is_read=/false, op.bytes_transferred, /empty_buffer=/false);
273
274	101x	coro_resume(&op);
275	101x	}
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	85x	complete_datagram_op(Op& op, Endpoint* source_out)
291		{
292	85x	op.stop_cb.reset();
293	85x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
294
295		// No EOF: a zero-length datagram is valid (success with 0 bytes).
296	168x	decode_io_result(
297	85x	op.ec_out, op.bytes_out, op.cancelled.load(std::memory_order_acquire),
298	85x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
299		/is_read=/false, op.bytes_transferred, /empty_buffer=/false);
300
301		// Write the source exactly when the decode reported success — a
302		// transfer outranks a raced cancellation flag there too.
303	164x	if (source_out && op.errn == 0 &&
304	79x	(op.bytes_transferred > 0 \|\|
305	20x	!op.cancelled.load(std::memory_order_acquire)))
306	20x	*source_out =
307	59x	from_sockaddr_as(op.source_storage, op.source_addrlen, Endpoint{});
308
309	85x	coro_resume(&op);
310	85x	}
311
312		} // namespace boost::corosio::detail
313
314		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
315