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

98.2% Lines (107/0/109) 100.0% List of functions (40/0/40)

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

reactor_op_complete.hpp

f(x) Functions (40)

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