include/boost/corosio/udp_socket.hpp

89.8% Lines (106/118) 100.0% List of functions (43/43) 68.3% Branches (28/41)

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

f(x) Functions (43)

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