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