include/boost/corosio/udp_socket.hpp

90.6% Lines (77/85) 100.0% List of functions (39/39)

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

udp_socket.hpp

f(x) Functions (39)

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