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

64.1% Lines (491/766) 79.1% List of functions (68/86)

[ ] prev/next file n p prev/next uncovered line

reactor_stream_socket.hpp

f(x) Functions (86)

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_STREAM_SOCKET_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
12
13		#include <boost/corosio/tcp_socket.hpp>
14		#include <boost/corosio/native/detail/reactor/reactor_basic_socket.hpp>
15		#include <boost/corosio/detail/dispatch_coro.hpp>
16		#include <boost/capy/buffers.hpp>
17
18		#include <coroutine>
19
20		#include <errno.h>
21		#include <sys/socket.h>
22		#include <sys/uio.h>
23
24		namespace boost::corosio::detail {
25
26		/** CRTP base for reactor-backed stream socket implementations.
27
28		Inherits shared data members and cancel/close/register logic
29		from reactor_basic_socket. Adds the stream-specific remote
30		endpoint, shutdown, and I/O dispatch (connect, read, write).
31
32		@tparam Derived The concrete socket type (CRTP).
33		@tparam Service The backend's socket service type.
34		@tparam ConnOp The backend's connect op type.
35		@tparam ReadOp The backend's read op type.
36		@tparam WriteOp The backend's write op type.
37		@tparam DescState The backend's descriptor_state type.
38		@tparam ImplBase The public vtable base
39		(tcp_socket::implementation or
40		local_stream_socket::implementation).
41		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
42		*/
43		template<
44		class Derived,
45		class Service,
46		class ConnOp,
47		class ReadOp,
48		class WriteOp,
49		class DescState,
50		class ImplBase = tcp_socket::implementation,
51		class Endpoint = endpoint>
52		class reactor_stream_socket
53		: public reactor_basic_socket<
54		Derived,
55		ImplBase,
56		Service,
57		DescState,
58		Endpoint>
59		{
60		using base_type = reactor_basic_socket<
61		Derived,
62		ImplBase,
63		Service,
64		DescState,
65		Endpoint>;
66		friend base_type;
67		friend Derived;
68
69	15247x	explicit reactor_stream_socket(Service& svc) noexcept : base_type(svc) {}
70
71		protected:
72		Endpoint remote_endpoint_;
73
74		public:
75		/// Pending connect operation slot.
76		ConnOp conn_;
77
78		/// Pending read operation slot.
79		ReadOp rd_;
80
81		/// Pending write operation slot.
82		WriteOp wr_;
83
84	15247x	~reactor_stream_socket() override = default;
85
86		/// Return the cached remote endpoint.
87	44x	Endpoint remote_endpoint() const noexcept override
88		{
89	44x	return remote_endpoint_;
90		}
91
92		// --- Virtual method overrides (satisfy ImplBase pure virtuals) ---
93
94	5046x	std::coroutine_handle<> connect(
95		std::coroutine_handle<> h,
96		capy::executor_ref ex,
97		Endpoint ep,
98		std::stop_token token,
99		std::error_code* ec) override
100		{
101	5046x	return do_connect(h, ex, ep, token, ec);
102		}
103
104	199943x	std::coroutine_handle<> read_some(
105		std::coroutine_handle<> h,
106		capy::executor_ref ex,
107		buffer_param param,
108		std::stop_token token,
109		std::error_code* ec,
110		std::size_t* bytes_out) override
111		{
112	199943x	return do_read_some(h, ex, param, token, ec, bytes_out);
113		}
114
115	199651x	std::coroutine_handle<> write_some(
116		std::coroutine_handle<> h,
117		capy::executor_ref ex,
118		buffer_param param,
119		std::stop_token token,
120		std::error_code* ec,
121		std::size_t* bytes_out) override
122		{
123	199651x	return do_write_some(h, ex, param, token, ec, bytes_out);
124		}
125
126		std::error_code
127	6x	shutdown(corosio::shutdown_type what) noexcept override
128		{
129	6x	return do_shutdown(static_cast<int>(what));
130		}
131
132	184x	void cancel() noexcept override
133		{
134	184x	this->do_cancel();
135	184x	}
136
137		// --- End virtual overrides ---
138
139		/// Close the socket (non-virtual, called by the service).
140	45745x	void close_socket() noexcept
141		{
142	45745x	this->do_close_socket();
143	45745x	}
144
145		/** Shut down part or all of the full-duplex connection.
146
147		@param what 0 = receive, 1 = send, 2 = both.
148		*/
149	6x	std::error_code do_shutdown(int what) noexcept
150		{
151		int how;
152	6x	switch (what)
153		{
154	2x	case 0: // shutdown_receive
155	2x	how = SHUT_RD;
156	2x	break;
157	2x	case 1: // shutdown_send
158	2x	how = SHUT_WR;
159	2x	break;
160	2x	case 2: // shutdown_both
161	2x	how = SHUT_RDWR;
162	2x	break;
163	✗	default:
164	✗	return make_err(EINVAL);
165		}
166	6x	if (::shutdown(this->fd_, how) != 0)
167	✗	return make_err(errno);
168	6x	return {};
169		}
170
171		/// Cache local and remote endpoints.
172	10098x	void set_endpoints(Endpoint local, Endpoint remote) noexcept
173		{
174	10098x	this->local_endpoint_ = std::move(local);
175	10098x	remote_endpoint_ = std::move(remote);
176	10098x	}
177
178		/** Shared connect dispatch.
179
180		Tries the connect syscall speculatively. On synchronous
181		completion, returns via inline budget or posts through queue.
182		On EINPROGRESS, registers with the reactor.
183		*/
184		std::coroutine_handle<> do_connect(
185		std::coroutine_handle<>,
186		capy::executor_ref,
187		Endpoint const&,
188		std::stop_token const&,
189		std::error_code*);
190
191		/** Shared scatter-read dispatch.
192
193		Tries readv() speculatively. On success or hard error,
194		returns via inline budget or posts through queue.
195		On EAGAIN, registers with the reactor.
196		*/
197		std::coroutine_handle<> do_read_some(
198		std::coroutine_handle<>,
199		capy::executor_ref,
200		buffer_param,
201		std::stop_token const&,
202		std::error_code*,
203		std::size_t*);
204
205		/** Shared gather-write dispatch.
206
207		Tries the write via WriteOp::write_policy speculatively.
208		On success or hard error, returns via inline budget or
209		posts through queue. On EAGAIN, registers with the reactor.
210		*/
211		std::coroutine_handle<> do_write_some(
212		std::coroutine_handle<>,
213		capy::executor_ref,
214		buffer_param,
215		std::stop_token const&,
216		std::error_code*,
217		std::size_t*);
218
219		/** Close the socket and cancel pending operations.
220
221		Extends the base do_close_socket() to also reset
222		the remote endpoint.
223		*/
224	45745x	void do_close_socket() noexcept
225		{
226	45745x	base_type::do_close_socket();
227	45745x	remote_endpoint_ = Endpoint{};
228	45745x	}
229
230		/** Release the socket without closing the fd.
231
232		Extends the base do_release_socket() to also reset
233		the remote endpoint.
234		*/
235	2x	native_handle_type do_release_socket() noexcept
236		{
237	2x	auto fd = base_type::do_release_socket();
238	2x	remote_endpoint_ = Endpoint{};
239	2x	return fd;
240		}
241
242		private:
243		// CRTP callbacks for reactor_basic_socket cancel/close
244
245		template<class Op>
246	190x	reactor_op_base** op_to_desc_slot(Op& op) noexcept
247		{
248	190x	if (&op == static_cast<void*>(&conn_))
249	✗	return &this->desc_state_.connect_op;
250	190x	if (&op == static_cast<void*>(&rd_))
251	190x	return &this->desc_state_.read_op;
252	✗	if (&op == static_cast<void*>(&wr_))
253	✗	return &this->desc_state_.write_op;
254	✗	return nullptr;
255		}
256
257		template<class Op>
258	✗	bool* op_to_cancel_flag(Op& op) noexcept
259		{
260	✗	if (&op == static_cast<void*>(&conn_))
261	✗	return &this->desc_state_.connect_cancel_pending;
262	✗	if (&op == static_cast<void*>(&rd_))
263	✗	return &this->desc_state_.read_cancel_pending;
264	✗	if (&op == static_cast<void*>(&wr_))
265	✗	return &this->desc_state_.write_cancel_pending;
266	✗	return nullptr;
267		}
268
269		template<class Fn>
270	45931x	void for_each_op(Fn fn) noexcept
271		{
272	45931x	fn(conn_);
273	45931x	fn(rd_);
274	45931x	fn(wr_);
275	45931x	}
276
277		template<class Fn>
278	45931x	void for_each_desc_entry(Fn fn) noexcept
279		{
280	45931x	fn(conn_, this->desc_state_.connect_op);
281	45931x	fn(rd_, this->desc_state_.read_op);
282	45931x	fn(wr_, this->desc_state_.write_op);
283	45931x	}
284		};
285
286		template<
287		class Derived,
288		class Service,
289		class ConnOp,
290		class ReadOp,
291		class WriteOp,
292		class DescState,
293		class ImplBase,
294		class Endpoint>
295		std::coroutine_handle<>
296	5046x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState, ImplBase, Endpoint>::
297		do_connect(
298		std::coroutine_handle<> h,
299		capy::executor_ref ex,
300		Endpoint const& ep,
301		std::stop_token const& token,
302		std::error_code* ec)
303		{
304	5046x	auto& op = conn_;
305
306	5046x	sockaddr_storage storage{};
307	5046x	socklen_t addrlen = to_sockaddr(ep, socket_family(this->fd_), storage);
308		int result =
309	5046x	::connect(this->fd_, reinterpret_cast<sockaddr*>(&storage), addrlen);
310
311	5046x	if (result == 0)
312		{
313	4x	sockaddr_storage local_storage{};
314	4x	socklen_t local_len = sizeof(local_storage);
315	4x	if (::getsockname(
316		this->fd_, reinterpret_cast<sockaddr*>(&local_storage),
317	4x	&local_len) == 0)
318	✗	this->local_endpoint_ =
319	4x	from_sockaddr_as(local_storage, local_len, Endpoint{});
320	4x	remote_endpoint_ = ep;
321		}
322
323	5046x	if (result == 0 \|\| errno != EINPROGRESS)
324		{
325	4x	int err = (result < 0) ? errno : 0;
326	4x	if (this->svc_.scheduler().try_consume_inline_budget())
327		{
328	✗	*ec = err ? make_err(err) : std::error_code{};
329	✗	op.cont_op.cont.h = h;
330	✗	return dispatch_coro(ex, op.cont_op.cont);
331		}
332	4x	op.reset();
333	4x	op.h = h;
334	4x	op.ex = ex;
335	4x	op.ec_out = ec;
336	4x	op.fd = this->fd_;
337	4x	op.target_endpoint = ep;
338	4x	op.start(token, static_cast<Derived*>(this));
339	4x	op.impl_ptr = this->shared_from_this();
340	4x	op.complete(err, 0);
341	4x	this->svc_.post(&op);
342	4x	return std::noop_coroutine();
343		}
344
345		// EINPROGRESS — register with reactor
346	5042x	op.reset();
347	5042x	op.h = h;
348	5042x	op.ex = ex;
349	5042x	op.ec_out = ec;
350	5042x	op.fd = this->fd_;
351	5042x	op.target_endpoint = ep;
352	5042x	op.start(token, static_cast<Derived*>(this));
353	5042x	op.impl_ptr = this->shared_from_this();
354
355	5042x	this->register_op(
356	5042x	op, this->desc_state_.connect_op, this->desc_state_.write_ready,
357	5042x	this->desc_state_.connect_cancel_pending, true);
358	5042x	return std::noop_coroutine();
359		}
360
361		template<
362		class Derived,
363		class Service,
364		class ConnOp,
365		class ReadOp,
366		class WriteOp,
367		class DescState,
368		class ImplBase,
369		class Endpoint>
370		std::coroutine_handle<>
371	199943x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState, ImplBase, Endpoint>::
372		do_read_some(
373		std::coroutine_handle<> h,
374		capy::executor_ref ex,
375		buffer_param param,
376		std::stop_token const& token,
377		std::error_code* ec,
378		std::size_t* bytes_out)
379		{
380	199943x	auto& op = rd_;
381	199943x	op.reset();
382
383	199943x	capy::mutable_buffer bufs[ReadOp::max_buffers];
384	199943x	op.iovec_count = static_cast<int>(param.copy_to(bufs, ReadOp::max_buffers));
385
386	199943x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
387		{
388	2x	op.empty_buffer_read = true;
389	2x	op.h = h;
390	2x	op.ex = ex;
391	2x	op.ec_out = ec;
392	2x	op.bytes_out = bytes_out;
393	2x	op.start(token, static_cast<Derived*>(this));
394	2x	op.impl_ptr = this->shared_from_this();
395	2x	op.complete(0, 0);
396	2x	this->svc_.post(&op);
397	2x	return std::noop_coroutine();
398		}
399
400	399882x	for (int i = 0; i < op.iovec_count; ++i)
401		{
402	199941x	op.iovecs[i].iov_base = bufs[i].data();
403	199941x	op.iovecs[i].iov_len = bufs[i].size();
404		}
405
406		// Speculative read
407		ssize_t n;
408		do
409		{
410	199941x	n = ::readv(this->fd_, op.iovecs, op.iovec_count);
411		}
412	199941x	while (n < 0 && errno == EINTR);
413
414	199941x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
415		{
416	199557x	int err = (n < 0) ? errno : 0;
417	199557x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
418
419	199557x	if (this->svc_.scheduler().try_consume_inline_budget())
420		{
421	159678x	if (err)
422	✗	*ec = make_err(err);
423	159678x	else if (n == 0)
424	10x	*ec = capy::error::eof;
425		else
426	159668x	*ec = {};
427	159678x	*bytes_out = bytes;
428	159678x	op.cont_op.cont.h = h;
429	159678x	return dispatch_coro(ex, op.cont_op.cont);
430		}
431	39879x	op.h = h;
432	39879x	op.ex = ex;
433	39879x	op.ec_out = ec;
434	39879x	op.bytes_out = bytes_out;
435	39879x	op.start(token, static_cast<Derived*>(this));
436	39879x	op.impl_ptr = this->shared_from_this();
437	39879x	op.complete(err, bytes);
438	39879x	this->svc_.post(&op);
439	39879x	return std::noop_coroutine();
440		}
441
442		// EAGAIN — register with reactor
443	384x	op.h = h;
444	384x	op.ex = ex;
445	384x	op.ec_out = ec;
446	384x	op.bytes_out = bytes_out;
447	384x	op.fd = this->fd_;
448	384x	op.start(token, static_cast<Derived*>(this));
449	384x	op.impl_ptr = this->shared_from_this();
450
451	384x	this->register_op(
452	384x	op, this->desc_state_.read_op, this->desc_state_.read_ready,
453	384x	this->desc_state_.read_cancel_pending);
454	384x	return std::noop_coroutine();
455		}
456
457		template<
458		class Derived,
459		class Service,
460		class ConnOp,
461		class ReadOp,
462		class WriteOp,
463		class DescState,
464		class ImplBase,
465		class Endpoint>
466		std::coroutine_handle<>
467	199651x	reactor_stream_socket<Derived, Service, ConnOp, ReadOp, WriteOp, DescState, ImplBase, Endpoint>::
468		do_write_some(
469		std::coroutine_handle<> h,
470		capy::executor_ref ex,
471		buffer_param param,
472		std::stop_token const& token,
473		std::error_code* ec,
474		std::size_t* bytes_out)
475		{
476	199651x	auto& op = wr_;
477	199651x	op.reset();
478
479	199651x	capy::mutable_buffer bufs[WriteOp::max_buffers];
480	199651x	op.iovec_count =
481	199651x	static_cast<int>(param.copy_to(bufs, WriteOp::max_buffers));
482
483	199651x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
484		{
485	2x	op.h = h;
486	2x	op.ex = ex;
487	2x	op.ec_out = ec;
488	2x	op.bytes_out = bytes_out;
489	2x	op.start(token, static_cast<Derived*>(this));
490	2x	op.impl_ptr = this->shared_from_this();
491	2x	op.complete(0, 0);
492	2x	this->svc_.post(&op);
493	2x	return std::noop_coroutine();
494		}
495
496	399298x	for (int i = 0; i < op.iovec_count; ++i)
497		{
498	199649x	op.iovecs[i].iov_base = bufs[i].data();
499	199649x	op.iovecs[i].iov_len = bufs[i].size();
500		}
501
502		// Speculative write via backend-specific write policy
503		ssize_t n =
504	199649x	WriteOp::write_policy::write(this->fd_, op.iovecs, op.iovec_count);
505
506	199649x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
507		{
508	199649x	int err = (n < 0) ? errno : 0;
509	199649x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
510
511	199649x	if (this->svc_.scheduler().try_consume_inline_budget())
512		{
513	159733x	*ec = err ? make_err(err) : std::error_code{};
514	159733x	*bytes_out = bytes;
515	159733x	op.cont_op.cont.h = h;
516	159733x	return dispatch_coro(ex, op.cont_op.cont);
517		}
518	39916x	op.h = h;
519	39916x	op.ex = ex;
520	39916x	op.ec_out = ec;
521	39916x	op.bytes_out = bytes_out;
522	39916x	op.start(token, static_cast<Derived*>(this));
523	39916x	op.impl_ptr = this->shared_from_this();
524	39916x	op.complete(err, bytes);
525	39916x	this->svc_.post(&op);
526	39916x	return std::noop_coroutine();
527		}
528
529		// EAGAIN — register with reactor
530	✗	op.h = h;
531	✗	op.ex = ex;
532	✗	op.ec_out = ec;
533	✗	op.bytes_out = bytes_out;
534	✗	op.fd = this->fd_;
535	✗	op.start(token, static_cast<Derived*>(this));
536	✗	op.impl_ptr = this->shared_from_this();
537
538	✗	this->register_op(
539	✗	op, this->desc_state_.write_op, this->desc_state_.write_ready,
540	✗	this->desc_state_.write_cancel_pending, true);
541	✗	return std::noop_coroutine();
542		}
543
544		} // namespace boost::corosio::detail
545
546		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_STREAM_SOCKET_HPP
547