include/boost/corosio/local_datagram_socket.hpp

92.2% Lines (47/51) 100.0% List of functions (19/19)

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

local_datagram_socket.hpp

f(x) Functions (19)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Michael Vandeberg
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_LOCAL_DATAGRAM_SOCKET_HPP
11		#define BOOST_COROSIO_LOCAL_DATAGRAM_SOCKET_HPP
12
13		#include <boost/corosio/detail/config.hpp>
14		#include <boost/corosio/detail/platform.hpp>
15		#include <boost/corosio/detail/except.hpp>
16		#include <boost/corosio/detail/native_handle.hpp>
17		#include <boost/corosio/detail/op_base.hpp>
18		#include <boost/corosio/io/io_object.hpp>
19		#include <boost/capy/io_result.hpp>
20		#include <boost/corosio/detail/buffer_param.hpp>
21		#include <boost/corosio/local_endpoint.hpp>
22		#include <boost/corosio/local_datagram.hpp>
23		#include <boost/corosio/message_flags.hpp>
24		#include <boost/corosio/shutdown_type.hpp>
25		#include <boost/capy/ex/executor_ref.hpp>
26		#include <boost/capy/ex/execution_context.hpp>
27		#include <boost/capy/ex/io_env.hpp>
28		#include <boost/capy/concept/executor.hpp>
29
30		#include <system_error>
31
32		#include <concepts>
33		#include <coroutine>
34		#include <cstddef>
35		#include <stop_token>
36		#include <type_traits>
37
38		namespace boost::corosio {
39
40		/** An asynchronous Unix datagram socket for coroutine I/O.
41
42		This class provides asynchronous Unix domain datagram socket
43		operations that return awaitable types. Each operation
44		participates in the affine awaitable protocol, ensuring
45		coroutines resume on the correct executor.
46
47		Supports two modes of operation:
48
49		Connectionless mode: each send_to specifies a destination
50		endpoint, and each recv_from captures the source endpoint.
51		The socket must be opened (and optionally bound) before I/O.
52
53		Connected mode: call connect() to set a default peer,
54		then use send()/recv() without endpoint arguments. The kernel
55		filters incoming datagrams to those from the connected peer.
56
57		@par Thread Safety
58		Distinct objects: Safe.@n
59		Shared objects: Unsafe. A socket must not have concurrent
60		operations of the same type (e.g., two simultaneous recv_from).
61		One send_to and one recv_from may be in flight simultaneously.
62		Note that recv and recv_from share the same internal read slot,
63		so they must not overlap; likewise send and send_to share the
64		write slot.
65		*/
66		class BOOST_COROSIO_DECL local_datagram_socket : public io_object
67		{
68		public:
69		using shutdown_type = corosio::shutdown_type;
70		using enum corosio::shutdown_type;
71
72		/** Define backend hooks for local datagram socket operations.
73
74		Platform backends (epoll, kqueue, select) derive from this
75		to implement datagram I/O, connection, and option management.
76		*/
77		struct implementation : io_object::implementation
78		{
79		/** Initiate an asynchronous send_to operation.
80
81		@param h Coroutine handle to resume on completion.
82		@param ex Executor for dispatching the completion.
83		@param buf The buffer data to send.
84		@param dest The destination endpoint.
85		@param flags Message flags (e.g. MSG_DONTROUTE).
86		@param token Stop token for cancellation.
87		@param ec Output error code.
88		@param bytes_out Output bytes transferred.
89
90		@return Coroutine handle to resume immediately.
91		*/
92		virtual std::coroutine_handle<> send_to(
93		std::coroutine_handle<> h,
94		capy::executor_ref ex,
95		buffer_param buf,
96		corosio::local_endpoint dest,
97		int flags,
98		std::stop_token token,
99		std::error_code* ec,
100		std::size_t* bytes_out) = 0;
101
102		/** Initiate an asynchronous recv_from operation.
103
104		@param h Coroutine handle to resume on completion.
105		@param ex Executor for dispatching the completion.
106		@param buf The buffer to receive into.
107		@param source Output endpoint for the sender's address.
108		@param flags Message flags (e.g. MSG_PEEK).
109		@param token Stop token for cancellation.
110		@param ec Output error code.
111		@param bytes_out Output bytes transferred.
112
113		@return Coroutine handle to resume immediately.
114		*/
115		virtual std::coroutine_handle<> recv_from(
116		std::coroutine_handle<> h,
117		capy::executor_ref ex,
118		buffer_param buf,
119		corosio::local_endpoint* source,
120		int flags,
121		std::stop_token token,
122		std::error_code* ec,
123		std::size_t* bytes_out) = 0;
124
125		/** Initiate an asynchronous connect to set the default peer.
126
127		@param h Coroutine handle to resume on completion.
128		@param ex Executor for dispatching the completion.
129		@param ep The remote endpoint to connect to.
130		@param token Stop token for cancellation.
131		@param ec Output error code.
132
133		@return Coroutine handle to resume immediately.
134		*/
135		virtual std::coroutine_handle<> connect(
136		std::coroutine_handle<> h,
137		capy::executor_ref ex,
138		corosio::local_endpoint ep,
139		std::stop_token token,
140		std::error_code* ec) = 0;
141
142		/** Initiate an asynchronous connected send operation.
143
144		@param h Coroutine handle to resume on completion.
145		@param ex Executor for dispatching the completion.
146		@param buf The buffer data to send.
147		@param flags Message flags (e.g. MSG_DONTROUTE).
148		@param token Stop token for cancellation.
149		@param ec Output error code.
150		@param bytes_out Output bytes transferred.
151
152		@return Coroutine handle to resume immediately.
153		*/
154		virtual std::coroutine_handle<> send(
155		std::coroutine_handle<> h,
156		capy::executor_ref ex,
157		buffer_param buf,
158		int flags,
159		std::stop_token token,
160		std::error_code* ec,
161		std::size_t* bytes_out) = 0;
162
163		/** Initiate an asynchronous connected recv operation.
164
165		@param h Coroutine handle to resume on completion.
166		@param ex Executor for dispatching the completion.
167		@param buf The buffer to receive into.
168		@param flags Message flags (e.g. MSG_PEEK).
169		@param token Stop token for cancellation.
170		@param ec Output error code.
171		@param bytes_out Output bytes transferred.
172
173		@return Coroutine handle to resume immediately.
174		*/
175		virtual std::coroutine_handle<> recv(
176		std::coroutine_handle<> h,
177		capy::executor_ref ex,
178		buffer_param buf,
179		int flags,
180		std::stop_token token,
181		std::error_code* ec,
182		std::size_t* bytes_out) = 0;
183
184		/// Shut down part or all of the socket.
185		virtual std::error_code shutdown(shutdown_type what) noexcept = 0;
186
187		/// Return the platform socket descriptor.
188		virtual native_handle_type native_handle() const noexcept = 0;
189
190		virtual native_handle_type release_socket() noexcept = 0;
191
192		/** Request cancellation of pending asynchronous operations.
193
194		All outstanding operations complete with operation_canceled
195		error. Check ec == cond::canceled for portable comparison.
196		*/
197		virtual void cancel() noexcept = 0;
198
199		/** Set a socket option.
200
201		@param level The protocol level (e.g. SOL_SOCKET).
202		@param optname The option name.
203		@param data Pointer to the option value.
204		@param size Size of the option value in bytes.
205		@return Error code on failure, empty on success.
206		*/
207		virtual std::error_code set_option(
208		int level,
209		int optname,
210		void const* data,
211		std::size_t size) noexcept = 0;
212
213		/** Get a socket option.
214
215		@param level The protocol level (e.g. SOL_SOCKET).
216		@param optname The option name.
217		@param data Pointer to receive the option value.
218		@param size On entry, the size of the buffer. On exit,
219		the size of the option value.
220		@return Error code on failure, empty on success.
221		*/
222		virtual std::error_code
223		get_option(int level, int optname, void* data, std::size_t* size)
224		const noexcept = 0;
225
226		/// Return the cached local endpoint.
227		virtual corosio::local_endpoint local_endpoint() const noexcept = 0;
228
229		/// Return the cached remote endpoint (connected mode).
230		virtual corosio::local_endpoint remote_endpoint() const noexcept = 0;
231
232		/** Bind the socket to a local endpoint.
233
234		@param ep The local endpoint to bind to.
235		@return Error code on failure, empty on success.
236		*/
237		virtual std::error_code
238		bind(corosio::local_endpoint ep) noexcept = 0;
239		};
240
241		/** Represent the awaitable returned by @ref send_to.
242
243		Captures the destination endpoint and buffer, then dispatches
244		to the backend implementation on suspension.
245		*/
246		struct send_to_awaitable
247		: detail::bytes_op_base<send_to_awaitable>
248		{
249		local_datagram_socket& s_;
250		buffer_param buf_;
251		corosio::local_endpoint dest_;
252		int flags_;
253
254	6x	send_to_awaitable(
255		local_datagram_socket& s, buffer_param buf,
256		corosio::local_endpoint dest, int flags = 0) noexcept
257	6x	: s_(s), buf_(buf), dest_(dest), flags_(flags) {}
258
259	6x	std::coroutine_handle<> dispatch(
260		std::coroutine_handle<> h, capy::executor_ref ex) const
261		{
262	12x	return s_.get().send_to(
263	12x	h, ex, buf_, dest_, flags_, token_, &ec_, &bytes_);
264		}
265		};
266
267		struct recv_from_awaitable
268		: detail::bytes_op_base<recv_from_awaitable>
269		{
270		local_datagram_socket& s_;
271		buffer_param buf_;
272		corosio::local_endpoint& source_;
273		int flags_;
274
275	8x	recv_from_awaitable(
276		local_datagram_socket& s, buffer_param buf,
277		corosio::local_endpoint& source, int flags = 0) noexcept
278	8x	: s_(s), buf_(buf), source_(source), flags_(flags) {}
279
280	8x	std::coroutine_handle<> dispatch(
281		std::coroutine_handle<> h, capy::executor_ref ex) const
282		{
283	16x	return s_.get().recv_from(
284	16x	h, ex, buf_, &source_, flags_, token_, &ec_, &bytes_);
285		}
286		};
287
288		struct connect_awaitable
289		: detail::void_op_base<connect_awaitable>
290		{
291		local_datagram_socket& s_;
292		corosio::local_endpoint endpoint_;
293
294		connect_awaitable(
295		local_datagram_socket& s,
296		corosio::local_endpoint ep) noexcept
297		: s_(s), endpoint_(ep) {}
298
299		std::coroutine_handle<> dispatch(
300		std::coroutine_handle<> h, capy::executor_ref ex) const
301		{
302		return s_.get().connect(
303		h, ex, endpoint_, token_, &ec_);
304		}
305		};
306
307		struct send_awaitable
308		: detail::bytes_op_base<send_awaitable>
309		{
310		local_datagram_socket& s_;
311		buffer_param buf_;
312		int flags_;
313
314	10x	send_awaitable(
315		local_datagram_socket& s, buffer_param buf,
316		int flags = 0) noexcept
317	10x	: s_(s), buf_(buf), flags_(flags) {}
318
319	10x	std::coroutine_handle<> dispatch(
320		std::coroutine_handle<> h, capy::executor_ref ex) const
321		{
322	20x	return s_.get().send(
323	20x	h, ex, buf_, flags_, token_, &ec_, &bytes_);
324		}
325		};
326
327		struct recv_awaitable
328		: detail::bytes_op_base<recv_awaitable>
329		{
330		local_datagram_socket& s_;
331		buffer_param buf_;
332		int flags_;
333
334	12x	recv_awaitable(
335		local_datagram_socket& s, buffer_param buf,
336		int flags = 0) noexcept
337	12x	: s_(s), buf_(buf), flags_(flags) {}
338
339	12x	std::coroutine_handle<> dispatch(
340		std::coroutine_handle<> h, capy::executor_ref ex) const
341		{
342	24x	return s_.get().recv(
343	24x	h, ex, buf_, flags_, token_, &ec_, &bytes_);
344		}
345		};
346
347		public:
348		/** Destructor.
349
350		Closes the socket if open, cancelling any pending operations.
351		*/
352		~local_datagram_socket() override;
353
354		/** Construct a socket from an execution context.
355
356		@param ctx The execution context that will own this socket.
357		*/
358		explicit local_datagram_socket(capy::execution_context& ctx);
359
360		/** Construct a socket from an executor.
361
362		The socket is associated with the executor's context.
363
364		@param ex The executor whose context will own the socket.
365		*/
366		template<class Ex>
367		requires(
368		!std::same_as<std::remove_cvref_t<Ex>, local_datagram_socket>) &&
369		capy::Executor<Ex>
370		explicit local_datagram_socket(Ex const& ex)
371		: local_datagram_socket(ex.context())
372		{
373		}
374
375		/** Move constructor.
376
377		Transfers ownership of the socket resources.
378
379		@param other The socket to move from.
380		*/
381	18x	local_datagram_socket(local_datagram_socket&& other) noexcept
382	18x	: io_object(std::move(other))
383		{
384	18x	}
385
386		/** Move assignment operator.
387
388		Closes any existing socket and transfers ownership.
389
390		@param other The socket to move from.
391		@return Reference to this socket.
392		*/
393		local_datagram_socket& operator=(local_datagram_socket&& other) noexcept
394		{
395		if (this != &other)
396		{
397		close();
398		io_object::operator=(std::move(other));
399		}
400		return *this;
401		}
402
403		local_datagram_socket(local_datagram_socket const&) = delete;
404		local_datagram_socket& operator=(local_datagram_socket const&) = delete;
405
406		/** Open the socket.
407
408		Creates a Unix datagram socket and associates it with
409		the platform reactor.
410
411		@param proto The protocol. Defaults to local_datagram{}.
412
413		@throws std::system_error on failure.
414		*/
415		void open(local_datagram proto = {});
416
417		/// Close the socket.
418		void close();
419
420		/// Check if the socket is open.
421	158x	bool is_open() const noexcept
422		{
423	158x	return h_ && get().native_handle() >= 0;
424		}
425
426		/** Bind the socket to a local endpoint.
427
428		Associates the socket with a local address (filesystem path).
429		Required before calling recv_from in connectionless mode.
430
431		@param ep The local endpoint to bind to.
432
433		@return Error code on failure, empty on success.
434
435		@throws std::logic_error if the socket is not open.
436		*/
437		std::error_code bind(corosio::local_endpoint ep);
438
439		/** Initiate an asynchronous connect to set the default peer.
440
441		If the socket is not already open, it is opened automatically.
442
443		@param ep The remote endpoint to connect to.
444
445		@par Cancellation
446		Supports cancellation via the awaitable's stop_token or by
447		calling @ref cancel. On cancellation, yields
448		`errc::operation_canceled`.
449
450		@return An awaitable that completes with io_result<>.
451
452		@throws std::system_error if the socket needs to be opened
453		and the open fails.
454		*/
455		auto connect(corosio::local_endpoint ep)
456		{
457		if (!is_open())
458		open();
459		return connect_awaitable(*this, ep);
460		}
461
462		/** Send a datagram to the specified destination.
463
464		@param buf The buffer containing data to send.
465		@param dest The destination endpoint.
466
467		@par Cancellation
468		Supports cancellation via the awaitable's stop_token or by
469		calling @ref cancel. On cancellation, yields
470		`errc::operation_canceled`.
471
472		@return An awaitable that completes with
473		io_result<std::size_t>.
474
475		@throws std::logic_error if the socket is not open.
476		*/
477		template<capy::ConstBufferSequence Buffers>
478	6x	auto send_to(
479		Buffers const& buf,
480		corosio::local_endpoint dest,
481		corosio::message_flags flags)
482		{
483	6x	if (!is_open())
484	✗	detail::throw_logic_error("send_to: socket not open");
485		return send_to_awaitable(
486	6x	*this, buf, dest, static_cast<int>(flags));
487		}
488
489		/// @overload
490		template<capy::ConstBufferSequence Buffers>
491	6x	auto send_to(Buffers const& buf, corosio::local_endpoint dest)
492		{
493	6x	return send_to(buf, dest, corosio::message_flags::none);
494		}
495
496		/** Receive a datagram and capture the sender's endpoint.
497
498		@param buf The buffer to receive data into.
499		@param source Reference to an endpoint that will be set to
500		the sender's address on successful completion.
501		@param flags Message flags (e.g. message_flags::peek).
502
503		@par Cancellation
504		Supports cancellation via the awaitable's stop_token or by
505		calling @ref cancel. On cancellation, yields
506		`errc::operation_canceled`.
507
508		@return An awaitable that completes with
509		io_result<std::size_t>.
510
511		@throws std::logic_error if the socket is not open.
512		*/
513		template<capy::MutableBufferSequence Buffers>
514	8x	auto recv_from(
515		Buffers const& buf,
516		corosio::local_endpoint& source,
517		corosio::message_flags flags)
518		{
519	8x	if (!is_open())
520	✗	detail::throw_logic_error("recv_from: socket not open");
521		return recv_from_awaitable(
522	8x	*this, buf, source, static_cast<int>(flags));
523		}
524
525		/// @overload
526		template<capy::MutableBufferSequence Buffers>
527	6x	auto recv_from(Buffers const& buf, corosio::local_endpoint& source)
528		{
529	6x	return recv_from(buf, source, corosio::message_flags::none);
530		}
531
532		/** Send a datagram to the connected peer.
533
534		@param buf The buffer containing data to send.
535		@param flags Message flags.
536
537		@par Cancellation
538		Supports cancellation via the awaitable's stop_token or by
539		calling @ref cancel. On cancellation, yields
540		`errc::operation_canceled`.
541
542		@return An awaitable that completes with
543		io_result<std::size_t>.
544
545		@throws std::logic_error if the socket is not open.
546		*/
547		template<capy::ConstBufferSequence Buffers>
548	10x	auto send(Buffers const& buf, corosio::message_flags flags)
549		{
550	10x	if (!is_open())
551	✗	detail::throw_logic_error("send: socket not open");
552		return send_awaitable(
553	10x	*this, buf, static_cast<int>(flags));
554		}
555
556		/// @overload
557		template<capy::ConstBufferSequence Buffers>
558	10x	auto send(Buffers const& buf)
559		{
560	10x	return send(buf, corosio::message_flags::none);
561		}
562
563		/** Receive a datagram from the connected peer.
564
565		@param buf The buffer to receive data into.
566		@param flags Message flags (e.g. message_flags::peek).
567
568		@par Cancellation
569		Supports cancellation via the awaitable's stop_token or by
570		calling @ref cancel. On cancellation, yields
571		`errc::operation_canceled`.
572
573		@return An awaitable that completes with
574		io_result<std::size_t>.
575
576		@throws std::logic_error if the socket is not open.
577		*/
578		template<capy::MutableBufferSequence Buffers>
579	12x	auto recv(Buffers const& buf, corosio::message_flags flags)
580		{
581	12x	if (!is_open())
582	✗	detail::throw_logic_error("recv: socket not open");
583		return recv_awaitable(
584	12x	*this, buf, static_cast<int>(flags));
585		}
586
587		/// @overload
588		template<capy::MutableBufferSequence Buffers>
589	10x	auto recv(Buffers const& buf)
590		{
591	10x	return recv(buf, corosio::message_flags::none);
592		}
593
594		/** Cancel any pending asynchronous operations.
595
596		All outstanding operations complete with
597		errc::operation_canceled. Check ec == cond::canceled
598		for portable comparison.
599		*/
600		void cancel();
601
602		/** Get the native socket handle.
603
604		@return The native socket handle, or -1 if not open.
605		*/
606		native_handle_type native_handle() const noexcept;
607
608		/** Release ownership of the native socket handle.
609
610		Deregisters the socket from the reactor and cancels pending
611		operations without closing the fd. The caller takes ownership
612		of the returned descriptor.
613
614		@return The native handle.
615
616		@throws std::logic_error if the socket is not open.
617		*/
618		native_handle_type release();
619
620		/** Query the number of bytes available for reading.
621
622		@return The number of bytes that can be read without blocking.
623
624		@throws std::logic_error if the socket is not open.
625		@throws std::system_error on ioctl failure.
626		*/
627		std::size_t available() const;
628
629		/** Shut down part or all of the socket (best-effort).
630
631		Calls `::shutdown` on the underlying descriptor when open.
632		Errors from the syscall (such as `ENOTCONN` on a peer that
633		already closed) are swallowed because they are typically
634		unhelpful at this layer; if the socket is not open, the call
635		is a no-op. To observe errors, use the
636		@ref shutdown(shutdown_type,std::error_code&) overload.
637
638		@param what Which direction to shut down.
639		*/
640		void shutdown(shutdown_type what);
641
642		/** Shut down part or all of the socket (non-throwing).
643
644		@param what Which direction to shut down.
645		@param ec Set to the error code on failure.
646		*/
647		void shutdown(shutdown_type what, std::error_code& ec) noexcept;
648
649		/** Set a socket option.
650
651		@param opt The option to set.
652
653		@throws std::logic_error if the socket is not open.
654		@throws std::system_error on failure.
655		*/
656		template<class Option>
657		void set_option(Option const& opt)
658		{
659		if (!is_open())
660		detail::throw_logic_error("set_option: socket not open");
661		std::error_code ec = get().set_option(
662		Option::level(), Option::name(), opt.data(), opt.size());
663		if (ec)
664		detail::throw_system_error(
665		ec, "local_datagram_socket::set_option");
666		}
667
668		/** Get a socket option.
669
670		@return The current option value.
671
672		@throws std::logic_error if the socket is not open.
673		@throws std::system_error on failure.
674		*/
675		template<class Option>
676		Option get_option() const
677		{
678		if (!is_open())
679		detail::throw_logic_error("get_option: socket not open");
680		Option opt{};
681		std::size_t sz = opt.size();
682		std::error_code ec =
683		get().get_option(Option::level(), Option::name(), opt.data(), &sz);
684		if (ec)
685		detail::throw_system_error(
686		ec, "local_datagram_socket::get_option");
687		opt.resize(sz);
688		return opt;
689		}
690
691		/** Assign an existing file descriptor to this socket.
692
693		The fd is adopted and registered with the platform reactor.
694
695		@param fd The file descriptor to adopt.
696
697		@par Preconditions
698		The socket must not already be open.
699
700		@throws std::logic_error if the precondition is violated
701		(the socket is already open).
702		@throws std::system_error on any other failure (e.g. the
703		fd is not an AF_UNIX datagram socket, or backend
704		configuration fails).
705		*/
706		void assign(int fd);
707
708		/** Get the local endpoint of the socket.
709
710		@return The local endpoint, or a default endpoint if not bound.
711		*/
712		corosio::local_endpoint local_endpoint() const noexcept;
713
714		/** Get the remote endpoint of the socket.
715
716		Returns the address of the connected peer.
717
718		@return The remote endpoint, or a default endpoint if
719		not connected.
720		*/
721		corosio::local_endpoint remote_endpoint() const noexcept;
722
723		protected:
724		/// Default-construct (for derived types).
725		local_datagram_socket() noexcept = default;
726
727		/// Construct from a pre-built handle.
728		explicit local_datagram_socket(handle h) noexcept
729		: io_object(std::move(h))
730		{
731		}
732
733		private:
734		void open_for_family(int family, int type, int protocol);
735
736	174x	inline implementation& get() const noexcept
737		{
738	174x	return static_cast<implementation>(h_.get());
739		}
740		};
741
742		} // namespace boost::corosio
743
744		#endif // BOOST_COROSIO_LOCAL_DATAGRAM_SOCKET_HPP
745