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

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