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

95.0% Lines (96/101) 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	75700x	complete_io_op(Op& op)
41		{
42	75700x	op.stop_cb.reset();
43	75700x	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	151400x	decode_io_result(
50		op.ec_out,
51	75700x	op.cancelled.load(std::memory_order_acquire),
52	75700x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
53	75700x	op.is_read_operation(), op.bytes_transferred, /empty_buffer=/false);
54
55	75700x	*op.bytes_out = op.bytes_transferred;
56
57	75700x	coro_resume(&op);
58	75700x	}
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	56x	complete_wait_op(Op& op)
99		{
100	56x	op.stop_cb.reset();
101	56x	if (op.socket_impl_)
102	44x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
103		else
104	12x	op.acceptor_impl_->desc_state_.scheduler_->reset_inline_budget();
105
106		// Wait reports only success/cancel/error — no bytes, no EOF.
107	112x	decode_io_result(
108		op.ec_out,
109	56x	op.cancelled.load(std::memory_order_acquire),
110	56x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
111		/is_read=/false, /bytes=/0, /empty_buffer=/false);
112
113	56x	coro_resume(&op);
114	56x	}
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	4214x	complete_connect_op(Op& op)
128		{
129	4214x	op.stop_cb.reset();
130	4214x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
131
132	4214x	bool success =
133	4214x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
134
135	4214x	if (success && op.socket_impl_)
136		{
137		using ep_type = decltype(op.target_endpoint);
138	4187x	ep_type local_ep;
139	4187x	sockaddr_storage local_storage{};
140	4187x	socklen_t local_len = sizeof(local_storage);
141	4187x	if (::getsockname(
142		op.fd, reinterpret_cast<sockaddr*>(&local_storage),
143	4187x	&local_len) == 0)
144	4167x	local_ep =
145	4187x	from_sockaddr_as(local_storage, local_len, ep_type{});
146	4187x	op.socket_impl_->set_endpoints(local_ep, op.target_endpoint);
147		}
148
149	8403x	decode_io_result(
150		op.ec_out,
151	4214x	op.cancelled.load(std::memory_order_acquire),
152	4214x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
153		/is_read=/false, /bytes=/0, /empty_buffer=/false);
154
155	4214x	coro_resume(&op);
156	4214x	}
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	4157x	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	4157x	auto* socket_svc = acceptor_impl->service().stream_service();
184	4157x	if (!socket_svc)
185		{
186	✗	*ec_out = make_err(ENOENT);
187	✗	return false;
188		}
189
190	4157x	auto& impl = static_cast<SocketImpl&>(*socket_svc->construct());
191	4157x	impl.set_socket(accepted_fd);
192
193	4157x	impl.desc_state_.fd = accepted_fd;
194		{
195	4157x	std::lock_guard lock(impl.desc_state_.mutex);
196	4157x	impl.desc_state_.read_op = nullptr;
197	4157x	impl.desc_state_.write_op = nullptr;
198	4157x	impl.desc_state_.connect_op = nullptr;
199	4157x	}
200	4157x	socket_svc->scheduler().register_descriptor(accepted_fd, &impl.desc_state_);
201
202		using ep_type = decltype(acceptor_impl->local_endpoint());
203	4157x	impl.set_endpoints(
204		acceptor_impl->local_endpoint(),
205	4157x	from_sockaddr_as(
206		peer_storage,
207		peer_addrlen,
208		ep_type{}));
209
210	4157x	if (impl_out)
211	4157x	*impl_out = &impl;
212	4157x	accepted_fd = -1;
213	4157x	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	4177x	complete_accept_op(Op& op)
228		{
229	4177x	op.stop_cb.reset();
230	4177x	op.acceptor_impl_->desc_state_.scheduler_->reset_inline_budget();
231
232	4177x	bool success =
233	4177x	(op.errn == 0 && !op.cancelled.load(std::memory_order_acquire));
234
235	8354x	decode_io_result(
236		op.ec_out,
237	4177x	op.cancelled.load(std::memory_order_acquire),
238	4177x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
239		/is_read=/false, /bytes=/0, /empty_buffer=/false);
240
241	4177x	if (success && op.accepted_fd >= 0 && op.acceptor_impl_)
242		{
243	4157x	if (!setup_accepted_socket<SocketImpl>(
244	4157x	op.acceptor_impl_, op.accepted_fd, op.peer_storage,
245		op.peer_addrlen, op.impl_out, op.ec_out))
246	✗	success = false;
247		}
248
249	4177x	if (!success \|\| !op.acceptor_impl_)
250		{
251	20x	if (op.accepted_fd >= 0)
252		{
253	✗	::close(op.accepted_fd);
254	✗	op.accepted_fd = -1;
255		}
256	20x	if (op.impl_out)
257	20x	*op.impl_out = nullptr;
258		}
259
260	4177x	coro_resume(&op);
261	4177x	}
262
263		/** Complete a datagram operation (send_to or recv_from).
264
265		For recv_from operations, writes the source endpoint from the
266		recorded sockaddr_storage into the caller's endpoint pointer.
267		Then resumes the caller via symmetric transfer.
268
269		@tparam Op The concrete datagram operation type.
270		@param op The operation to complete.
271		*/
272		template<typename Op>
273		void
274	18x	complete_datagram_op(Op& op)
275		{
276	18x	op.stop_cb.reset();
277	18x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
278
279		// No EOF: a zero-length datagram is valid (success with 0 bytes).
280	36x	decode_io_result(
281		op.ec_out,
282	18x	op.cancelled.load(std::memory_order_acquire),
283	18x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
284		/is_read=/false, /bytes=/0, /empty_buffer=/false);
285
286	18x	*op.bytes_out = op.bytes_transferred;
287
288	18x	coro_resume(&op);
289	18x	}
290
291		/** Complete a datagram operation with source endpoint capture.
292
293		For recv_from operations, writes the source endpoint from the
294		recorded sockaddr_storage into the caller's endpoint pointer.
295		Then resumes the caller via symmetric transfer.
296
297		@tparam Op The concrete datagram operation type.
298		@param op The operation to complete.
299		@param source_out Optional pointer to store source endpoint
300		(non-null for recv_from, null for send_to).
301		*/
302		template<typename Op, typename Endpoint>
303		void
304	26x	complete_datagram_op(Op& op, Endpoint* source_out)
305		{
306	26x	op.stop_cb.reset();
307	26x	op.socket_impl_->desc_state_.scheduler_->reset_inline_budget();
308
309		// No EOF: a zero-length datagram is valid (success with 0 bytes).
310	52x	decode_io_result(
311		op.ec_out,
312	26x	op.cancelled.load(std::memory_order_acquire),
313	26x	op.errn != 0 ? make_err(op.errn) : std::error_code{},
314		/is_read=/false, /bytes=/0, /empty_buffer=/false);
315
316	26x	*op.bytes_out = op.bytes_transferred;
317
318	36x	if (source_out && !op.cancelled.load(std::memory_order_acquire) &&
319	10x	op.errn == 0)
320	20x	*source_out = from_sockaddr_as(
321	10x	op.source_storage,
322		op.source_addrlen,
323		Endpoint{});
324
325	26x	coro_resume(&op);
326	26x	}
327
328		} // namespace boost::corosio::detail
329
330		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_OP_COMPLETE_HPP
331