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

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