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

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