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

98.1% Lines (106/0/108) 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	109725x	complete_io_op(Op& op)
41		{
42	109725x	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	109725x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
47	109709x	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	219310x	decode_io_result(
54	109725x	op.ec_out, op.cancelled.load(std::memory_order_acquire),
55	109725x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
56	109725x	op.is_read_operation(), op.bytes_transferred, /empty_buffer=/false);
57
58	109725x	*op.bytes_out = op.bytes_transferred;
59
60	109725x	coro_resume(&op);
61	109725x	}
62
63		/** Complete a wait operation.
64
65		Wait operations report only an error_code — no bytes_transferred,
66		no EOF translation. Used for socket and acceptor wait() awaitables;
67		picks the impl pointer set by start() to reach the scheduler.
68
69		@tparam Op The concrete wait operation type.
70		@param op The operation to complete.
71		*/
72		template<typename Op>
73		void
74	172x	complete_wait_op(Op& op)
75		{
76	172x	op.stop_cb.reset();
77		// scheduler_ is null until the descriptor is registered; a wait
78		// completed by the initiation probe (e.g. EBADF on a never-opened
79		// socket) has no registration to reset a budget for.
80	172x	if (op.socket_impl_)
81		{
82	131x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
83	119x	sched->reset_inline_budget();
84		}
85	41x	else if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
86		{
87	37x	sched->reset_inline_budget();
88		}
89
90		// Wait reports only success/cancel/error — no bytes, no EOF.
91	312x	decode_io_result(
92	172x	op.ec_out, op.cancelled.load(std::memory_order_acquire),
93	172x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
94		/is_read=/false, /bytes=/0, /empty_buffer=/false);
95
96	172x	coro_resume(&op);
97	172x	}
98
99		/** Complete a connect operation with endpoint caching.
100
101		On success, queries the local endpoint via getsockname and
102		caches both endpoints in the socket impl. Then resumes the
103		caller via symmetric transfer.
104
105		@tparam Op The concrete connect operation type.
106		@param op The operation to complete.
107		*/
108		template<typename Op>
109		void
110	9151x	complete_connect_op(Op& op)
111		{
112	9151x	op.stop_cb.reset();
113	9151x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
114
115	9151x	bool success =
116	9151x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
117
118	9151x	if (success && op.socket_impl_)
119		{
120		using ep_type = decltype(op.target_endpoint);
121	9095x	ep_type local_ep;
122	9095x	sockaddr_storage local_storage{};
123	9095x	socklen_t local_len = sizeof(local_storage);
124	9095x	if (::getsockname(
125		op.fd, reinterpret_cast<sockaddr*>(&local_storage),
126	9095x	&local_len) == 0)
127	9095x	local_ep = from_sockaddr_as(local_storage, local_len, ep_type{});
128	9095x	op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
129		}
130
131	18257x	decode_io_result(
132	9151x	op.ec_out, op.cancelled.load(std::memory_order_acquire),
133	9151x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
134		/is_read=/false, /bytes=/0, /empty_buffer=/false);
135
136	9151x	coro_resume(&op);
137	9151x	}
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	9041x	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	9041x	auto* socket_svc = acceptor_impl->service().stream_service();
167	9041x	if (!socket_svc)
168		{
169	✗	*ec_out = make_err(ENOENT);
170	✗	return false;
171		}
172
173	9041x	auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
174	9041x	impl.set_socket(accepted_fd);
175
176	9041x	impl.desc_state_.fd = accepted_fd;
177		{
178	9041x	std::lock_guard lock(impl.desc_state_.mutex);
179	9041x	impl.desc_state_.read_op = nullptr;
180	9041x	impl.desc_state_.write_op = nullptr;
181	9041x	impl.desc_state_.connect_op = nullptr;
182	9041x	}
183	9041x	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	9040x	impl.set_endpoints(
195		acceptor_impl->local_endpoint(),
196	9040x	from_sockaddr_as(peer_storage, peer_addrlen, ep_type{}));
197
198	9040x	if (impl_out)
199	9040x	*impl_out = &impl;
200	9040x	accepted_fd = -1;
201	9040x	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	9152x	complete_accept_op(Op& op)
216		{
217	9152x	op.stop_cb.reset();
218	9152x	if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
219	9148x	sched->reset_inline_budget();
220
221	9152x	bool success =
222	9152x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
223
224	18293x	decode_io_result(
225	9152x	op.ec_out, op.cancelled.load(std::memory_order_acquire),
226	9152x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
227		/is_read=/false, /bytes=/0, /empty_buffer=/false);
228
229	9152x	if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
230		{
231	9041x	if (!setup_accepted_socket<SocketImpl>(
232	9041x	op.acceptor_impl_, op.accepted_fd, op.peer_storage,
233		op.peer_addrlen, op.impl_out, op.ec_out))
234	1x	success = false;
235		}
236
237	9152x	if (!success \|\| !op.acceptor_impl_)
238		{
239	112x	if (op.accepted_fd >= 0)
240		{
241	2x	::close(op.accepted_fd);
242	2x	op.accepted_fd = -1;
243		}
244	112x	if (op.impl_out)
245	112x	*op.impl_out = nullptr;
246		}
247
248	9152x	coro_resume(&op);
249	9152x	}
250
251		/** Complete a datagram operation (send_to or recv_from).
252
253		For recv_from operations, writes the source endpoint from the
254		recorded sockaddr_storage into the caller's endpoint pointer.
255		Then resumes the caller via symmetric transfer.
256
257		@tparam Op The concrete datagram operation type.
258		@param op The operation to complete.
259		*/
260		template<typename Op>
261		void
262	113x	complete_datagram_op(Op& op)
263		{
264	113x	op.stop_cb.reset();
265	113x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
266
267		// No EOF: a zero-length datagram is valid (success with 0 bytes).
268	222x	decode_io_result(
269	113x	op.ec_out, op.cancelled.load(std::memory_order_acquire),
270	113x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
271		/is_read=/false, /bytes=/0, /empty_buffer=/false);
272
273	113x	*op.bytes_out = op.bytes_transferred;
274
275	113x	coro_resume(&op);
276	113x	}
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	95x	complete_datagram_op(Op& op, Endpoint* source_out)
292		{
293	95x	op.stop_cb.reset();
294	95x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
295
296		// No EOF: a zero-length datagram is valid (success with 0 bytes).
297	188x	decode_io_result(
298	95x	op.ec_out, op.cancelled.load(std::memory_order_acquire),
299	95x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
300		/is_read=/false, /bytes=/0, /empty_buffer=/false);
301
302	95x	*op.bytes_out = op.bytes_transferred;
303
304	156x	if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
305	61x	op.errn == 0)
306	20x	*source_out =
307	59x	from_sockaddr_as(op.source_storage, op.source_addrlen, Endpoint{});
308
309	95x	coro_resume(&op);
310	95x	}
311
312		} // namespace boost::corosio::detail
313
314		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
315