include/boost/corosio/udp_socket.hpp

90.6% Lines (77/85) 100.0% List of functions (37/37) 71.0% Branches (22/31)

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

udp_socket.hpp

f(x) Functions (37)

Function Calls Lines Branches Blocks

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