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	82613x	complete_io_op(Op& op)
41		{
42	82613x	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	82613x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
47	82597x	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	165187x	decode_io_result(
54		op.ec_out,
55	82613x	op.cancelled.load(std::memory_order_acquire),
56	82613x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
57	82613x	op.is_read_operation(), op.bytes_transferred, /empty_buffer=/false);
58
59	82613x	*op.bytes_out = op.bytes_transferred;
60
61	82613x	coro_resume(&op);
62	82613x	}
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	135x	complete_wait_op(Op& op)
103		{
104	135x	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	135x	if (op.socket_impl_)
109		{
110	104x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
111	92x	sched->reset_inline_budget();
112		}
113	31x	else if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
114		{
115	27x	sched->reset_inline_budget();
116		}
117
118		// Wait reports only success/cancel/error — no bytes, no EOF.
119	239x	decode_io_result(
120		op.ec_out,
121	135x	op.cancelled.load(std::memory_order_acquire),
122	135x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
123		/is_read=/false, /bytes=/0, /empty_buffer=/false);
124
125	135x	coro_resume(&op);
126	135x	}
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	5229x	complete_connect_op(Op& op)
140		{
141	5229x	op.stop_cb.reset();
142	5229x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
143
144	5229x	bool success =
145	5229x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
146
147	5229x	if (success && op.socket_impl_)
148		{
149		using ep_type = decltype(op.target_endpoint);
150	5173x	ep_type local_ep;
151	5173x	sockaddr_storage local_storage{};
152	5173x	socklen_t local_len = sizeof(local_storage);
153	5173x	if (::getsockname(
154		op.fd, reinterpret_cast<sockaddr*>(&local_storage),
155	5173x	&local_len) == 0)
156	5144x	local_ep =
157	5173x	from_sockaddr_as(local_storage, local_len, ep_type{});
158	5173x	op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
159		}
160
161	10413x	decode_io_result(
162		op.ec_out,
163	5229x	op.cancelled.load(std::memory_order_acquire),
164	5229x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
165		/is_read=/false, /bytes=/0, /empty_buffer=/false);
166
167	5229x	coro_resume(&op);
168	5229x	}
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	5132x	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	5132x	auto* socket_svc = acceptor_impl->service().stream_service();
198	5132x	if (!socket_svc)
199		{
200	✗	*ec_out = make_err(ENOENT);
201	✗	return false;
202		}
203
204	5132x	auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
205	5132x	impl.set_socket(accepted_fd);
206
207	5132x	impl.desc_state_.fd = accepted_fd;
208		{
209	5132x	std::lock_guard lock(impl.desc_state_.mutex);
210	5132x	impl.desc_state_.read_op = nullptr;
211	5132x	impl.desc_state_.write_op = nullptr;
212	5132x	impl.desc_state_.connect_op = nullptr;
213	5132x	}
214	5132x	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	5131x	impl.set_endpoints(
226		acceptor_impl->local_endpoint(),
227	5131x	from_sockaddr_as(
228		peer_storage,
229		peer_addrlen,
230		ep_type{}));
231
232	5131x	if (impl_out)
233	5131x	*impl_out = &impl;
234	5131x	accepted_fd = -1;
235	5131x	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	5183x	complete_accept_op(Op& op)
250		{
251	5183x	op.stop_cb.reset();
252	5183x	if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
253	5179x	sched->reset_inline_budget();
254
255	5183x	bool success =
256	5183x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
257
258	10355x	decode_io_result(
259		op.ec_out,
260	5183x	op.cancelled.load(std::memory_order_acquire),
261	5183x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
262		/is_read=/false, /bytes=/0, /empty_buffer=/false);
263
264	5183x	if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
265		{
266	5132x	if (!setup_accepted_socket<SocketImpl>(
267	5132x	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	5183x	if (!success \|\| !op.acceptor_impl_)
273		{
274	52x	if (op.accepted_fd >= 0)
275		{
276	2x	::close(op.accepted_fd);
277	2x	op.accepted_fd = -1;
278		}
279	52x	if (op.impl_out)
280	52x	*op.impl_out = nullptr;
281		}
282
283	5183x	coro_resume(&op);
284	5183x	}
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	66x	complete_datagram_op(Op& op)
298		{
299	66x	op.stop_cb.reset();
300	66x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
301
302		// No EOF: a zero-length datagram is valid (success with 0 bytes).
303	130x	decode_io_result(
304		op.ec_out,
305	66x	op.cancelled.load(std::memory_order_acquire),
306	66x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
307		/is_read=/false, /bytes=/0, /empty_buffer=/false);
308
309	66x	*op.bytes_out = op.bytes_transferred;
310
311	66x	coro_resume(&op);
312	66x	}
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	83x	complete_datagram_op(Op& op, Endpoint* source_out)
328		{
329	83x	op.stop_cb.reset();
330	83x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
331
332		// No EOF: a zero-length datagram is valid (success with 0 bytes).
333	164x	decode_io_result(
334		op.ec_out,
335	83x	op.cancelled.load(std::memory_order_acquire),
336	83x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
337		/is_read=/false, /bytes=/0, /empty_buffer=/false);
338
339	83x	*op.bytes_out = op.bytes_transferred;
340
341	138x	if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
342	55x	op.errn == 0)
343	106x	*source_out = from_sockaddr_as(
344	53x	op.source_storage,
345		op.source_addrlen,
346		Endpoint{});
347
348	83x	coro_resume(&op);
349	83x	}
350
351		} // namespace boost::corosio::detail
352
353		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
354