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

93.6% Lines (102/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	135842x	complete_io_op(Op& op)
41		{
42	135842x	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	135842x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
47	135826x	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	271661x	decode_io_result(
54		op.ec_out,
55	135842x	op.cancelled.load(std::memory_order_acquire),
56	135842x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
57	135842x	op.is_read_operation(), op.bytes_transferred, /empty_buffer=/false);
58
59	135842x	*op.bytes_out = op.bytes_transferred;
60
61	135842x	coro_resume(&op);
62	135842x	}
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	120x	complete_wait_op(Op& op)
103		{
104	120x	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	120x	if (op.socket_impl_)
109		{
110	89x	if (auto* sched = op.socket_impl_->desc_state_.scheduler_)
111	77x	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	218x	decode_io_result(
120		op.ec_out,
121	120x	op.cancelled.load(std::memory_order_acquire),
122	120x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
123		/is_read=/false, /bytes=/0, /empty_buffer=/false);
124
125	120x	coro_resume(&op);
126	120x	}
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	7082x	complete_connect_op(Op& op)
140		{
141	7082x	op.stop_cb.reset();
142	7082x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
143
144	7082x	bool success =
145	7082x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
146
147	7082x	if (success && op.socket_impl_)
148		{
149		using ep_type = decltype(op.target_endpoint);
150	7037x	ep_type local_ep;
151	7037x	sockaddr_storage local_storage{};
152	7037x	socklen_t local_len = sizeof(local_storage);
153	7037x	if (::getsockname(
154		op.fd, reinterpret_cast<sockaddr*>(&local_storage),
155	7037x	&local_len) == 0)
156	7010x	local_ep =
157	7037x	from_sockaddr_as(local_storage, local_len, ep_type{});
158	7037x	op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
159		}
160
161	14130x	decode_io_result(
162		op.ec_out,
163	7082x	op.cancelled.load(std::memory_order_acquire),
164	7082x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
165		/is_read=/false, /bytes=/0, /empty_buffer=/false);
166
167	7082x	coro_resume(&op);
168	7082x	}
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	7020x	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	7020x	auto* socket_svc = acceptor_impl->service().stream_service();
198	7020x	if (!socket_svc)
199		{
200	✗	*ec_out = make_err(ENOENT);
201	✗	return false;
202		}
203
204	7020x	auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
205	7020x	impl.set_socket(accepted_fd);
206
207	7020x	impl.desc_state_.fd = accepted_fd;
208		{
209	7020x	std::lock_guard lock(impl.desc_state_.mutex);
210	7020x	impl.desc_state_.read_op = nullptr;
211	7020x	impl.desc_state_.write_op = nullptr;
212	7020x	impl.desc_state_.connect_op = nullptr;
213	7020x	}
214	7020x	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	✗	accepted_fd = -1;
219	✗	socket_svc->destroy(&impl);
220	✗	*ec_out = ec;
221	✗	return false;
222		}
223
224		using ep_type = decltype(acceptor_impl->local_endpoint());
225	7020x	impl.set_endpoints(
226		acceptor_impl->local_endpoint(),
227	7020x	from_sockaddr_as(
228		peer_storage,
229		peer_addrlen,
230		ep_type{}));
231
232	7020x	if (impl_out)
233	7020x	*impl_out = &impl;
234	7020x	accepted_fd = -1;
235	7020x	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	7064x	complete_accept_op(Op& op)
250		{
251	7064x	op.stop_cb.reset();
252	7064x	if (auto* sched = op.acceptor_impl_->desc_state_.scheduler_)
253	7060x	sched->reset_inline_budget();
254
255	7064x	bool success =
256	7064x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
257
258	14124x	decode_io_result(
259		op.ec_out,
260	7064x	op.cancelled.load(std::memory_order_acquire),
261	7064x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
262		/is_read=/false, /bytes=/0, /empty_buffer=/false);
263
264	7064x	if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
265		{
266	7020x	if (!setup_accepted_socket<SocketImpl>(
267	7020x	op.acceptor_impl_, op.accepted_fd, op.peer_storage,
268		op.peer_addrlen, op.impl_out, op.ec_out))
269	✗	success = false;
270		}
271
272	7064x	if (!success \|\| !op.acceptor_impl_)
273		{
274	44x	if (op.accepted_fd >= 0)
275		{
276	2x	::close(op.accepted_fd);
277	2x	op.accepted_fd = -1;
278		}
279	44x	if (op.impl_out)
280	44x	*op.impl_out = nullptr;
281		}
282
283	7064x	coro_resume(&op);
284	7064x	}
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	62x	complete_datagram_op(Op& op)
298		{
299	62x	op.stop_cb.reset();
300	62x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
301
302		// No EOF: a zero-length datagram is valid (success with 0 bytes).
303	124x	decode_io_result(
304		op.ec_out,
305	62x	op.cancelled.load(std::memory_order_acquire),
306	62x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
307		/is_read=/false, /bytes=/0, /empty_buffer=/false);
308
309	62x	*op.bytes_out = op.bytes_transferred;
310
311	62x	coro_resume(&op);
312	62x	}
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	80x	complete_datagram_op(Op& op, Endpoint* source_out)
328		{
329	80x	op.stop_cb.reset();
330	80x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
331
332		// No EOF: a zero-length datagram is valid (success with 0 bytes).
333	160x	decode_io_result(
334		op.ec_out,
335	80x	op.cancelled.load(std::memory_order_acquire),
336	80x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
337		/is_read=/false, /bytes=/0, /empty_buffer=/false);
338
339	80x	*op.bytes_out = op.bytes_transferred;
340
341	132x	if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
342	52x	op.errn == 0)
343	104x	*source_out = from_sockaddr_as(
344	52x	op.source_storage,
345		op.source_addrlen,
346		Endpoint{});
347
348	80x	coro_resume(&op);
349	80x	}
350
351		} // namespace boost::corosio::detail
352
353		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
354