noq_udp/
unix.rs

1#[cfg(not(any(apple, target_os = "openbsd", solarish)))]
2use std::ptr;
3use std::{
4    io::{self, IoSliceMut},
5    mem::{self, MaybeUninit},
6    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6},
7    num::NonZeroUsize,
8    os::fd::AsRawFd,
9    sync::{
10        Mutex,
11        atomic::{AtomicBool, AtomicUsize, Ordering},
12    },
13    time::{Duration, Instant},
14};
15
16use socket2::SockRef;
17
18use super::{
19    EcnCodepoint, IO_ERROR_LOG_INTERVAL, RecvMeta, Transmit, UdpSockRef, cmsg, log_sendmsg_error,
20};
21
22#[cfg(apple_fast)]
23use super::apple_fast::{msghdr_x, recv_via_recvmsg_x, send};
24#[cfg(any(target_os = "linux", target_os = "android"))]
25use super::linux::gso;
26
27/// Tokio-compatible UDP socket with some useful specializations.
28///
29/// Unlike a standard tokio UDP socket, this allows ECN bits to be read and written on some
30/// platforms.
31#[derive(Debug)]
32pub struct UdpSocketState {
33    last_send_error: Mutex<Instant>,
34    max_gso_segments: AtomicUsize,
35    gro_segments: NonZeroUsize,
36    may_fragment: bool,
37
38    /// True if we have received EINVAL error from `sendmsg` system call at least once.
39    ///
40    /// If enabled, we assume that old kernel is used and switch to fallback mode.
41    /// In particular, we do not use IP_TOS cmsg_type in this case,
42    /// which is not supported on Linux <3.13 and results in not sending the UDP packet at all.
43    sendmsg_einval: AtomicBool,
44
45    /// Whether to use Apple's fast `sendmsg_x`/`recvmsg_x` APIs.
46    ///
47    /// These private APIs provide better performance but may not be available on all
48    /// Apple OS versions. Callers must verify availability before enabling.
49    #[cfg(apple_fast)]
50    apple_fast_path: AtomicBool,
51}
52
53impl UdpSocketState {
54    pub fn new(sock: UdpSockRef<'_>) -> io::Result<Self> {
55        let io = sock.0;
56        let mut cmsg_platform_space = 0;
57        #[cfg(not(target_os = "redox"))]
58        if cfg!(target_os = "linux")
59            || cfg!(bsd)
60            || cfg!(apple)
61            || cfg!(target_os = "android")
62            || cfg!(solarish)
63        {
64            cmsg_platform_space +=
65                unsafe { libc::CMSG_SPACE(size_of::<libc::in6_pktinfo>() as _) as usize };
66        }
67
68        assert!(
69            cmsg::LEN
70                >= unsafe { libc::CMSG_SPACE(size_of::<libc::c_int>() as _) as usize }
71                    + cmsg_platform_space
72        );
73        assert!(
74            align_of::<libc::cmsghdr>() <= align_of::<cmsg::Aligned<[u8; 0]>>(),
75            "control message buffers will be misaligned"
76        );
77
78        io.set_nonblocking(true)?;
79
80        let addr = io.local_addr()?;
81        let is_ipv4 = addr.family() == libc::AF_INET as libc::sa_family_t;
82
83        // mac and ios do not support IP_RECVTOS on dual-stack sockets :(
84        // older macos versions also don't have the flag and will error out if we don't ignore it
85        #[cfg(not(any(
86            target_os = "openbsd",
87            target_os = "netbsd",
88            target_os = "dragonfly",
89            solarish
90        )))]
91        if (is_ipv4 || !io.only_v6()?)
92            && let Err(_err) =
93                set_socket_option(&*io, libc::IPPROTO_IP, libc::IP_RECVTOS, OPTION_ON)
94        {
95            crate::log::debug!("Ignoring error setting IP_RECVTOS on socket: {_err:?}");
96        }
97
98        let mut may_fragment = false;
99        #[cfg_attr(
100            not(any(target_os = "linux", target_os = "android")),
101            expect(unused_mut)
102        )]
103        let mut gro_segments = NonZeroUsize::MIN;
104
105        #[cfg(any(target_os = "linux", target_os = "android"))]
106        {
107            // Forbid IPv4 fragmentation. Set even for IPv6 to account for IPv6 mapped IPv4 addresses.
108            // Set `may_fragment` to `true` if this option is not supported on the platform.
109            may_fragment |= !set_socket_option_supported(
110                &*io,
111                libc::IPPROTO_IP,
112                libc::IP_MTU_DISCOVER,
113                libc::IP_PMTUDISC_PROBE,
114            )?;
115
116            if is_ipv4 {
117                set_socket_option(&*io, libc::IPPROTO_IP, libc::IP_PKTINFO, OPTION_ON)?;
118            } else {
119                // Set `may_fragment` to `true` if this option is not supported on the platform.
120                may_fragment |= !set_socket_option_supported(
121                    &*io,
122                    libc::IPPROTO_IPV6,
123                    libc::IPV6_MTU_DISCOVER,
124                    libc::IPV6_PMTUDISC_PROBE,
125                )?;
126            }
127
128            if set_socket_option(&*io, libc::SOL_UDP, libc::UDP_GRO, OPTION_ON).is_ok() {
129                // As defined in net/ipv4/udp_offload.c
130                // #define UDP_GRO_CNT_MAX 64
131                //
132                // NOTE: this MUST be set to UDP_GRO_CNT_MAX to ensure that the receive buffer size
133                // (get_max_udp_payload_size() * gro_segments()) is large enough to hold the largest GRO
134                // list the kernel might potentially produce. See
135                // https://github.com/quinn-rs/quinn/pull/1354.
136                gro_segments = NonZeroUsize::new(64).expect("known");
137            }
138
139            if let Err(_err) =
140                set_socket_option(&*io, libc::SOL_SOCKET, libc::SO_TIMESTAMPNS, OPTION_ON)
141            {
142                crate::log::debug!("Ignoring error setting SO_TIMESTAMPNS on socket: {_err:?}");
143            }
144        }
145        #[cfg(any(target_os = "freebsd", apple))]
146        {
147            if is_ipv4 {
148                // Set `may_fragment` to `true` if this option is not supported on the platform.
149                may_fragment |= !set_socket_option_supported(
150                    &*io,
151                    libc::IPPROTO_IP,
152                    libc::IP_DONTFRAG,
153                    OPTION_ON,
154                )?;
155            }
156        }
157        #[cfg(any(bsd, apple, solarish))]
158        // IP_RECVDSTADDR == IP_SENDSRCADDR on FreeBSD
159        // macOS uses only IP_RECVDSTADDR, no IP_SENDSRCADDR on macOS (the same on Solaris)
160        // macOS also supports IP_PKTINFO
161        {
162            if is_ipv4 {
163                set_socket_option(&*io, libc::IPPROTO_IP, libc::IP_RECVDSTADDR, OPTION_ON)?;
164            }
165        }
166
167        // Options standardized in RFC 3542
168        #[cfg(not(target_os = "redox"))]
169        if !is_ipv4 {
170            set_socket_option(&*io, libc::IPPROTO_IPV6, libc::IPV6_RECVPKTINFO, OPTION_ON)?;
171            set_socket_option(&*io, libc::IPPROTO_IPV6, libc::IPV6_RECVTCLASS, OPTION_ON)?;
172            // Linux's IP_PMTUDISC_PROBE allows us to operate under interface MTU rather than the
173            // kernel's path MTU guess, but actually disabling fragmentation requires this too. See
174            // __ip6_append_data in ip6_output.c.
175            // Set `may_fragment` to `true` if this option is not supported on the platform.
176            may_fragment |= !set_socket_option_supported(
177                &*io,
178                libc::IPPROTO_IPV6,
179                libc::IPV6_DONTFRAG,
180                OPTION_ON,
181            )?;
182        }
183
184        let now = Instant::now();
185        Ok(Self {
186            last_send_error: Mutex::new(now.checked_sub(2 * IO_ERROR_LOG_INTERVAL).unwrap_or(now)),
187            max_gso_segments: AtomicUsize::new(gso::max_gso_segments(&*io)),
188            gro_segments,
189            may_fragment,
190            sendmsg_einval: AtomicBool::new(false),
191            #[cfg(apple_fast)]
192            apple_fast_path: AtomicBool::new(false),
193        })
194    }
195
196    /// Sends a [`Transmit`] on the given socket.
197    ///
198    /// This function will only ever return errors of kind [`io::ErrorKind::WouldBlock`].
199    /// All other errors will be logged and converted to `Ok`.
200    ///
201    /// UDP transmission errors are considered non-fatal because higher-level protocols must
202    /// employ retransmits and timeouts anyway in order to deal with UDP's unreliable nature.
203    /// Thus, logging is most likely the only thing you can do with these errors.
204    ///
205    /// If you would like to handle these errors yourself, use [`UdpSocketState::try_send`]
206    /// instead.
207    pub fn send(&self, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
208        match send(self, socket.0, transmit) {
209            Ok(()) => Ok(()),
210            Err(e) if e.kind() == io::ErrorKind::WouldBlock => Err(e),
211            // - EMSGSIZE is expected for MTU probes. Future work might be able to avoid
212            //   these by automatically clamping the MTUD upper bound to the interface MTU.
213            Err(e) if e.raw_os_error() == Some(libc::EMSGSIZE) => Ok(()),
214            Err(e) => {
215                log_sendmsg_error(&self.last_send_error, e, transmit);
216
217                Ok(())
218            }
219        }
220    }
221
222    /// Sends a [`Transmit`] on the given socket without any additional error handling.
223    pub fn try_send(&self, socket: UdpSockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
224        send(self, socket.0, transmit)
225    }
226
227    #[cfg(not(any(
228        apple,
229        target_os = "openbsd",
230        target_os = "netbsd",
231        target_os = "dragonfly",
232        target_os = "redox",
233        solarish
234    )))]
235    pub fn recv(
236        &self,
237        socket: UdpSockRef<'_>,
238        bufs: &mut [IoSliceMut<'_>],
239        meta: &mut [RecvMeta],
240    ) -> io::Result<usize> {
241        recv_via_recvmmsg(socket.0, bufs, meta)
242    }
243
244    #[cfg(apple_fast)]
245    pub fn recv(
246        &self,
247        socket: UdpSockRef<'_>,
248        bufs: &mut [IoSliceMut<'_>],
249        meta: &mut [RecvMeta],
250    ) -> io::Result<usize> {
251        if self.is_apple_fast_path_enabled() {
252            recv_via_recvmsg_x(self, socket.0, bufs, meta)
253        } else {
254            recv_single(socket.0, bufs, meta)
255        }
256    }
257
258    #[cfg(any(
259        target_os = "openbsd",
260        target_os = "netbsd",
261        target_os = "dragonfly",
262        target_os = "redox",
263        solarish,
264        apple_slow
265    ))]
266    pub fn recv(
267        &self,
268        socket: UdpSockRef<'_>,
269        bufs: &mut [IoSliceMut<'_>],
270        meta: &mut [RecvMeta],
271    ) -> io::Result<usize> {
272        recv_single(socket.0, bufs, meta)
273    }
274
275    /// Maximum number of segments to transmit if Generic Send Offload (GSO) is enabled.
276    ///
277    /// This is 1 if the platform doesn't support GSO.
278    ///
279    /// Subject to change if errors are detected while using GSO.
280    #[inline]
281    pub fn max_gso_segments(&self) -> NonZeroUsize {
282        self.max_gso_segments
283            .load(Ordering::Relaxed)
284            .try_into()
285            .expect("must have non zero GSO segments")
286    }
287
288    /// The number of segments to read when GRO is enabled.
289    ///
290    /// Used as a factor to compute the receive buffer size.
291    ///
292    /// Returns 1 if the platform doesn't support GRO.
293    #[inline]
294    pub fn gro_segments(&self) -> NonZeroUsize {
295        self.gro_segments
296    }
297
298    /// Resize the send buffer of `socket` to `bytes`
299    #[inline]
300    pub fn set_send_buffer_size(&self, socket: UdpSockRef<'_>, bytes: usize) -> io::Result<()> {
301        socket.0.set_send_buffer_size(bytes)
302    }
303
304    /// Resize the receive buffer of `socket` to `bytes`
305    #[inline]
306    pub fn set_recv_buffer_size(&self, socket: UdpSockRef<'_>, bytes: usize) -> io::Result<()> {
307        socket.0.set_recv_buffer_size(bytes)
308    }
309
310    /// Get the size of the `socket` send buffer
311    #[inline]
312    pub fn send_buffer_size(&self, socket: UdpSockRef<'_>) -> io::Result<usize> {
313        socket.0.send_buffer_size()
314    }
315
316    /// Get the size of the `socket` receive buffer
317    #[inline]
318    pub fn recv_buffer_size(&self, socket: UdpSockRef<'_>) -> io::Result<usize> {
319        socket.0.recv_buffer_size()
320    }
321
322    /// Whether transmitted datagrams might get fragmented by the IP layer
323    ///
324    /// Returns `false` on targets which employ e.g. the `IPV6_DONTFRAG` socket option.
325    #[inline]
326    pub fn may_fragment(&self) -> bool {
327        self.may_fragment
328    }
329
330    /// Returns true if we previously got an EINVAL error from `sendmsg` syscall.
331    pub(crate) fn sendmsg_einval(&self) -> bool {
332        self.sendmsg_einval.load(Ordering::Relaxed)
333    }
334
335    /// Sets the flag indicating we got EINVAL error from `sendmsg` syscall.
336    #[cfg(not(any(apple, target_os = "openbsd", target_os = "netbsd")))]
337    fn set_sendmsg_einval(&self) {
338        self.sendmsg_einval.store(true, Ordering::Relaxed)
339    }
340
341    /// Enables Apple's fast UDP datapath using private `sendmsg_x`/`recvmsg_x` APIs.
342    ///
343    /// Once enabled, this also updates [`max_gso_segments`] to allow batched sends.
344    ///
345    /// # Safety
346    ///
347    /// These APIs may crash on unsupported OS versions, so callers must verify
348    /// availability before enabling.
349    ///
350    /// [`max_gso_segments`]: Self::max_gso_segments
351    #[cfg(apple_fast)]
352    pub unsafe fn set_apple_fast_path(&self) {
353        self.apple_fast_path.store(true, Ordering::Relaxed);
354        self.max_gso_segments.store(BATCH_SIZE, Ordering::Relaxed);
355    }
356
357    /// Returns whether Apple's fast UDP datapath is enabled for this socket.
358    #[cfg(apple_fast)]
359    pub fn is_apple_fast_path_enabled(&self) -> bool {
360        self.apple_fast_path.load(Ordering::Relaxed)
361    }
362
363    /// Disables Apple's fast UDP datapath, reverting to `sendmsg`/`recvmsg`.
364    #[cfg(apple_fast)]
365    fn disable_apple_fast_path(&self) {
366        self.apple_fast_path.store(false, Ordering::Relaxed);
367        self.max_gso_segments.store(1, Ordering::Relaxed);
368    }
369
370    /// Resolves an Apple fast-path function pointer via `resolver`.
371    ///
372    /// Disables the fast path if the symbol is absent so that future calls use the slow path
373    /// directly.
374    #[cfg(apple_fast)]
375    pub(crate) fn resolve_apple_fast_fn<T>(&self, resolver: fn() -> Option<T>) -> Option<T> {
376        let f = resolver();
377        if f.is_none() {
378            self.disable_apple_fast_path();
379        }
380        f
381    }
382}
383
384#[cfg(not(any(apple, target_os = "openbsd", target_os = "netbsd")))]
385fn send(
386    #[allow(unused_variables)] // only used on Linux
387    state: &UdpSocketState,
388    io: SockRef<'_>,
389    transmit: &Transmit<'_>,
390) -> io::Result<()> {
391    #[allow(unused_mut)] // only mutable on FreeBSD
392    let mut encode_src_ip = true;
393    #[cfg(target_os = "freebsd")]
394    {
395        let addr = io.local_addr()?;
396        let is_ipv4 = addr.family() == libc::AF_INET as libc::sa_family_t;
397        if is_ipv4 {
398            if let Some(socket) = addr.as_socket_ipv4() {
399                encode_src_ip = socket.ip() == &Ipv4Addr::UNSPECIFIED;
400            }
401        }
402    }
403    let mut msg_hdr: libc::msghdr = unsafe { mem::zeroed() };
404    let mut iovec: libc::iovec = unsafe { mem::zeroed() };
405    let mut cmsgs = cmsg::Aligned([0u8; cmsg::LEN]);
406    let dst_addr = socket2::SockAddr::from(transmit.destination);
407    prepare_msg(
408        transmit,
409        &dst_addr,
410        &mut msg_hdr,
411        &mut iovec,
412        &mut cmsgs,
413        encode_src_ip,
414        state.sendmsg_einval(),
415    );
416
417    loop {
418        let n = unsafe { libc::sendmsg(io.as_raw_fd(), &msg_hdr, 0) };
419
420        if n >= 0 {
421            return Ok(());
422        }
423
424        let e = io::Error::last_os_error();
425        match e.kind() {
426            // Retry the transmission
427            io::ErrorKind::Interrupted => continue,
428            io::ErrorKind::WouldBlock => return Err(e),
429            _ => {
430                // Some network adapters and drivers do not support GSO. Unfortunately, Linux
431                // offers no easy way for us to detect this short of an EIO or sometimes EINVAL
432                // when we try to actually send datagrams using it.
433                #[cfg(any(target_os = "linux", target_os = "android"))]
434                if let Some(libc::EIO) | Some(libc::EINVAL) = e.raw_os_error() {
435                    // Prevent new transmits from being scheduled using GSO. Existing GSO transmits
436                    // may already be in the pipeline, so we need to tolerate additional failures.
437                    if state.max_gso_segments().get() > 1 {
438                        crate::log::info!(
439                            "`libc::sendmsg` failed with {e}; halting segmentation offload"
440                        );
441                        state.max_gso_segments.store(1, Ordering::Relaxed);
442                    }
443                }
444
445                // Some arguments to `sendmsg` are not supported. Switch to
446                // fallback mode and retry if we haven't already.
447                if e.raw_os_error() == Some(libc::EINVAL) && !state.sendmsg_einval() {
448                    state.set_sendmsg_einval();
449                    prepare_msg(
450                        transmit,
451                        &dst_addr,
452                        &mut msg_hdr,
453                        &mut iovec,
454                        &mut cmsgs,
455                        encode_src_ip,
456                        state.sendmsg_einval(),
457                    );
458                    continue;
459                }
460
461                return Err(e);
462            }
463        }
464    }
465}
466
467#[cfg(any(target_os = "openbsd", target_os = "netbsd", apple_slow))]
468fn send(state: &UdpSocketState, io: SockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
469    send_single(state, io, transmit)
470}
471
472#[cfg(any(target_os = "openbsd", target_os = "netbsd", apple))]
473#[cfg_attr(apple_fast, allow(dead_code))] // Unused when apple_fast is enabled
474pub(crate) fn send_single(
475    state: &UdpSocketState,
476    io: SockRef<'_>,
477    transmit: &Transmit<'_>,
478) -> io::Result<()> {
479    let mut hdr: libc::msghdr = unsafe { mem::zeroed() };
480    let mut iov: libc::iovec = unsafe { mem::zeroed() };
481    let mut ctrl = cmsg::Aligned([0u8; cmsg::LEN]);
482    let addr = socket2::SockAddr::from(transmit.destination);
483    prepare_msg(
484        transmit,
485        &addr,
486        &mut hdr,
487        &mut iov,
488        &mut ctrl,
489        cfg!(apple) || cfg!(target_os = "openbsd") || cfg!(target_os = "netbsd"),
490        state.sendmsg_einval(),
491    );
492    retry_if_interrupted(|| unsafe { libc::sendmsg(io.as_raw_fd(), &hdr, 0) })?;
493    Ok(())
494}
495
496/// Receive using the batched `recvmmsg` syscall.
497#[cfg(not(any(
498    apple,
499    target_os = "openbsd",
500    target_os = "netbsd",
501    target_os = "dragonfly",
502    target_os = "redox",
503    solarish
504)))]
505fn recv_via_recvmmsg(
506    io: SockRef<'_>,
507    bufs: &mut [IoSliceMut<'_>],
508    meta: &mut [RecvMeta],
509) -> io::Result<usize> {
510    let mut names = [MaybeUninit::<libc::sockaddr_storage>::uninit(); BATCH_SIZE];
511    let mut ctrls = [cmsg::Aligned(MaybeUninit::<[u8; cmsg::LEN]>::uninit()); BATCH_SIZE];
512    let mut hdrs = unsafe { mem::zeroed::<[libc::mmsghdr; BATCH_SIZE]>() };
513    let max_msg_count = bufs.len().min(BATCH_SIZE);
514    for i in 0..max_msg_count {
515        prepare_recv(
516            &mut bufs[i],
517            &mut names[i],
518            &mut ctrls[i],
519            &mut hdrs[i].msg_hdr,
520        );
521    }
522    let msg_count = retry_if_interrupted(|| unsafe {
523        libc::recvmmsg(
524            io.as_raw_fd(),
525            hdrs.as_mut_ptr(),
526            bufs.len().min(BATCH_SIZE) as _,
527            0,
528            ptr::null_mut::<libc::timespec>(),
529        ) as isize
530    })?;
531    for i in 0..(msg_count as usize) {
532        meta[i] = decode_recv(&names[i], &hdrs[i].msg_hdr, hdrs[i].msg_len as usize)?;
533    }
534    Ok(msg_count as usize)
535}
536
537#[cfg(any(
538    target_os = "openbsd",
539    target_os = "netbsd",
540    target_os = "dragonfly",
541    target_os = "redox",
542    solarish,
543    apple
544))]
545#[cfg_attr(apple_fast, allow(dead_code))] // Unused when apple_fast is enabled
546pub(crate) fn recv_single(
547    io: SockRef<'_>,
548    bufs: &mut [IoSliceMut<'_>],
549    meta: &mut [RecvMeta],
550) -> io::Result<usize> {
551    let mut name = MaybeUninit::<libc::sockaddr_storage>::uninit();
552    let mut ctrl = cmsg::Aligned(MaybeUninit::<[u8; cmsg::LEN]>::uninit());
553    let mut hdr = unsafe { mem::zeroed::<libc::msghdr>() };
554    prepare_recv(&mut bufs[0], &mut name, &mut ctrl, &mut hdr);
555    let n = loop {
556        let n = unsafe { libc::recvmsg(io.as_raw_fd(), &mut hdr, 0) };
557
558        if hdr.msg_flags & libc::MSG_TRUNC != 0 {
559            continue;
560        }
561
562        if n >= 0 {
563            break n;
564        }
565
566        let e = io::Error::last_os_error();
567        match e.kind() {
568            // Retry receiving
569            io::ErrorKind::Interrupted => continue,
570            _ => return Err(e),
571        }
572    };
573    meta[0] = decode_recv(&name, &hdr, n as usize)?;
574    Ok(1)
575}
576
577#[cfg_attr(apple_fast, allow(dead_code))] // Unused when apple_fast is enabled
578fn prepare_msg(
579    transmit: &Transmit<'_>,
580    dst_addr: &socket2::SockAddr,
581    hdr: &mut libc::msghdr,
582    iov: &mut libc::iovec,
583    ctrl: &mut cmsg::Aligned<[u8; cmsg::LEN]>,
584    #[allow(unused_variables)] // only used on FreeBSD & macOS
585    encode_src_ip: bool,
586    sendmsg_einval: bool,
587) {
588    iov.iov_base = transmit.contents.as_ptr() as *const _ as *mut _;
589    iov.iov_len = transmit.contents.len();
590
591    // SAFETY: Casting the pointer to a mutable one is legal,
592    // as sendmsg is guaranteed to not alter the mutable pointer
593    // as per the POSIX spec. See the section on the sys/socket.h
594    // header for details. The type is only mutable in the first
595    // place because it is reused by recvmsg as well.
596    let name = dst_addr.as_ptr() as *mut libc::c_void;
597    let namelen = dst_addr.len();
598    hdr.msg_name = name as *mut _;
599    hdr.msg_namelen = namelen;
600    hdr.msg_iov = iov;
601    hdr.msg_iovlen = 1;
602
603    hdr.msg_control = ctrl.0.as_mut_ptr() as _;
604    hdr.msg_controllen = cmsg::LEN as _;
605    let mut encoder = unsafe { cmsg::Encoder::new(hdr) };
606    let ecn = transmit.ecn.map_or(0, |x| x as libc::c_int);
607    // True for IPv4 or IPv4-Mapped IPv6
608    let is_ipv4 = transmit.destination.is_ipv4()
609        || matches!(transmit.destination.ip(), IpAddr::V6(addr) if addr.to_ipv4_mapped().is_some());
610    if is_ipv4 {
611        if !sendmsg_einval {
612            #[cfg(not(target_os = "netbsd"))]
613            {
614                encoder.push(libc::IPPROTO_IP, libc::IP_TOS, ecn as IpTosTy);
615            }
616        }
617    } else {
618        #[cfg(not(target_os = "redox"))]
619        encoder.push(libc::IPPROTO_IPV6, libc::IPV6_TCLASS, ecn);
620    }
621
622    // On apple_fast, prepare_msg is only compiled for send_single (fallback path), while the main
623    // send path uses prepare_msg_x with msghdr_x. gso::set_segment_size has a different signature
624    // when apple_fast is enabled, and it's a no-op on non-Linux platforms anyway.
625    #[cfg(not(apple_fast))]
626    if let Some(segment_size) = transmit.effective_segment_size() {
627        gso::set_segment_size(&mut encoder, segment_size as u16);
628    }
629
630    if let Some(ip) = &transmit.src_ip {
631        match ip {
632            IpAddr::V4(v4) => {
633                #[cfg(any(target_os = "linux", target_os = "android"))]
634                {
635                    let pktinfo = libc::in_pktinfo {
636                        ipi_ifindex: 0,
637                        ipi_spec_dst: libc::in_addr {
638                            s_addr: u32::from_ne_bytes(v4.octets()),
639                        },
640                        ipi_addr: libc::in_addr { s_addr: 0 },
641                    };
642                    encoder.push(libc::IPPROTO_IP, libc::IP_PKTINFO, pktinfo);
643                }
644                #[cfg(any(bsd, apple, solarish))]
645                {
646                    if encode_src_ip {
647                        let addr = libc::in_addr {
648                            s_addr: u32::from_ne_bytes(v4.octets()),
649                        };
650                        encoder.push(libc::IPPROTO_IP, libc::IP_RECVDSTADDR, addr);
651                    }
652                }
653            }
654            #[cfg(target_os = "redox")]
655            IpAddr::V6(_) => {}
656            #[cfg(not(target_os = "redox"))]
657            IpAddr::V6(v6) => {
658                let pktinfo = libc::in6_pktinfo {
659                    ipi6_ifindex: 0,
660                    ipi6_addr: libc::in6_addr {
661                        s6_addr: v6.octets(),
662                    },
663                };
664                encoder.push(libc::IPPROTO_IPV6, libc::IPV6_PKTINFO, pktinfo);
665            }
666        }
667    }
668
669    encoder.finish();
670}
671
672#[cfg_attr(apple_fast, allow(dead_code))] // Unused when apple_fast is enabled
673fn prepare_recv(
674    buf: &mut IoSliceMut<'_>,
675    name: &mut MaybeUninit<libc::sockaddr_storage>,
676    ctrl: &mut cmsg::Aligned<MaybeUninit<[u8; cmsg::LEN]>>,
677    hdr: &mut libc::msghdr,
678) {
679    hdr.msg_name = name.as_mut_ptr() as _;
680    hdr.msg_namelen = size_of::<libc::sockaddr_storage>() as _;
681    hdr.msg_iov = buf as *mut IoSliceMut<'_> as *mut libc::iovec;
682    hdr.msg_iovlen = 1;
683    hdr.msg_control = ctrl.0.as_mut_ptr() as _;
684    hdr.msg_controllen = cmsg::LEN as _;
685    hdr.msg_flags = 0;
686}
687
688pub(crate) fn decode_recv<M: cmsg::MsgHdr<ControlMessage = libc::cmsghdr>>(
689    name: &MaybeUninit<libc::sockaddr_storage>,
690    hdr: &M,
691    len: usize,
692) -> io::Result<RecvMeta> {
693    let name = unsafe { name.assume_init() };
694    let mut ctrl = ControlMetadata {
695        ecn_bits: 0,
696        dst_ip: None,
697        interface_index: None,
698        stride: len,
699        timestamp: None,
700    };
701
702    let cmsg_iter = unsafe { cmsg::Iter::new(hdr) };
703    for cmsg in cmsg_iter {
704        ctrl.decode(cmsg);
705    }
706
707    Ok(RecvMeta {
708        len,
709        stride: ctrl.stride,
710        addr: decode_socket_addr(&name)?,
711        ecn: EcnCodepoint::from_bits(ctrl.ecn_bits),
712        dst_ip: ctrl.dst_ip,
713        interface_index: ctrl.interface_index,
714        timestamp: ctrl.timestamp,
715    })
716}
717
718/// Metadata decoded from control messages
719struct ControlMetadata {
720    ecn_bits: u8,
721    dst_ip: Option<IpAddr>,
722    interface_index: Option<u32>,
723    stride: usize,
724    timestamp: Option<Duration>,
725}
726
727impl ControlMetadata {
728    /// Decodes a control message and updates the metadata state
729    fn decode(&mut self, cmsg: &libc::cmsghdr) {
730        match (cmsg.cmsg_level, cmsg.cmsg_type) {
731            (libc::IPPROTO_IP, libc::IP_TOS) => unsafe {
732                self.ecn_bits = cmsg::decode::<u8, libc::cmsghdr>(cmsg);
733            },
734            // FreeBSD uses IP_RECVTOS here, and we can be liberal because cmsgs are opt-in.
735            #[cfg(not(any(
736                target_os = "openbsd",
737                target_os = "netbsd",
738                target_os = "dragonfly",
739                solarish
740            )))]
741            (libc::IPPROTO_IP, libc::IP_RECVTOS) => unsafe {
742                self.ecn_bits = cmsg::decode::<u8, libc::cmsghdr>(cmsg);
743            },
744            #[cfg(not(target_os = "redox",))]
745            (libc::IPPROTO_IPV6, libc::IPV6_TCLASS) => unsafe {
746                // Temporary hack around broken macos ABI. Remove once upstream fixes it.
747                // https://bugreport.apple.com/web/?problemID=48761855
748                #[allow(clippy::unnecessary_cast)] // cmsg.cmsg_len defined as size_t
749                if cfg!(apple)
750                    && cmsg.cmsg_len as usize == libc::CMSG_LEN(size_of::<u8>() as _) as usize
751                {
752                    self.ecn_bits = cmsg::decode::<u8, libc::cmsghdr>(cmsg);
753                } else {
754                    self.ecn_bits = cmsg::decode::<libc::c_int, libc::cmsghdr>(cmsg) as u8;
755                }
756            },
757            #[cfg(any(target_os = "linux", target_os = "android"))]
758            (libc::IPPROTO_IP, libc::IP_PKTINFO) => {
759                let pktinfo = unsafe { cmsg::decode::<libc::in_pktinfo, libc::cmsghdr>(cmsg) };
760                self.dst_ip = Some(IpAddr::V4(Ipv4Addr::from(
761                    pktinfo.ipi_addr.s_addr.to_ne_bytes(),
762                )));
763                self.interface_index = Some(pktinfo.ipi_ifindex as u32);
764            }
765            #[cfg(any(bsd, apple))]
766            (libc::IPPROTO_IP, libc::IP_RECVDSTADDR) => {
767                let in_addr = unsafe { cmsg::decode::<libc::in_addr, libc::cmsghdr>(cmsg) };
768                self.dst_ip = Some(IpAddr::V4(Ipv4Addr::from(in_addr.s_addr.to_ne_bytes())));
769            }
770            #[cfg(not(target_os = "redox",))]
771            (libc::IPPROTO_IPV6, libc::IPV6_PKTINFO) => {
772                let pktinfo = unsafe { cmsg::decode::<libc::in6_pktinfo, libc::cmsghdr>(cmsg) };
773                self.dst_ip = Some(IpAddr::V6(Ipv6Addr::from(pktinfo.ipi6_addr.s6_addr)));
774                #[cfg_attr(not(target_os = "android"), expect(clippy::unnecessary_cast))]
775                {
776                    self.interface_index = Some(pktinfo.ipi6_ifindex as u32);
777                }
778            }
779            #[cfg(any(target_os = "linux", target_os = "android"))]
780            (libc::SOL_UDP, libc::UDP_GRO) => unsafe {
781                self.stride = cmsg::decode::<libc::c_int, libc::cmsghdr>(cmsg) as usize;
782            },
783            #[cfg(any(target_os = "linux", target_os = "android"))]
784            (libc::SOL_SOCKET, libc::SCM_TIMESTAMPNS) => {
785                let ts = unsafe { cmsg::decode::<libc::timespec, libc::cmsghdr>(cmsg) };
786                let secs = u64::try_from(ts.tv_sec).unwrap_or(0);
787                let nsecs = u32::try_from(ts.tv_nsec).unwrap_or(0);
788                self.timestamp = Some(Duration::new(secs, nsecs));
789            }
790            _ => {}
791        }
792    }
793}
794
795/// Decodes a `sockaddr_storage` into a `SocketAddr`
796pub(crate) fn decode_socket_addr(name: &libc::sockaddr_storage) -> io::Result<SocketAddr> {
797    match libc::c_int::from(name.ss_family) {
798        libc::AF_INET => {
799            // Safety: if the ss_family field is AF_INET then storage must be a sockaddr_in.
800            let addr: &libc::sockaddr_in =
801                unsafe { &*(name as *const _ as *const libc::sockaddr_in) };
802            Ok(SocketAddr::V4(SocketAddrV4::new(
803                Ipv4Addr::from(addr.sin_addr.s_addr.to_ne_bytes()),
804                u16::from_be(addr.sin_port),
805            )))
806        }
807        libc::AF_INET6 => {
808            // Safety: if the ss_family field is AF_INET6 then storage must be a sockaddr_in6.
809            let addr: &libc::sockaddr_in6 =
810                unsafe { &*(name as *const _ as *const libc::sockaddr_in6) };
811            Ok(SocketAddr::V6(SocketAddrV6::new(
812                Ipv6Addr::from(addr.sin6_addr.s6_addr),
813                u16::from_be(addr.sin6_port),
814                addr.sin6_flowinfo,
815                addr.sin6_scope_id,
816            )))
817        }
818        f => Err(io::Error::other(format!(
819            "expected AF_INET or AF_INET6, got {f}"
820        ))),
821    }
822}
823
824#[cfg(not(apple_slow))]
825// Chosen somewhat arbitrarily; might benefit from additional tuning.
826pub(crate) const BATCH_SIZE: usize = 32;
827
828#[cfg(apple_slow)]
829pub(crate) const BATCH_SIZE: usize = 1;
830
831// On Apple platforms using the `sendmsg_x` call, UDP datagram segmentation is not
832// offloaded to the NIC or even the kernel, but instead done here in user space in
833// [`send`]) and then passed to the OS as individual `iovec`s (up to `BATCH_SIZE`).
834// The initial value is 1 (no batching); callers can enable batching via
835// `UdpSocketState::set_apple_fast_path()` which updates `max_gso_segments`.
836#[cfg(not(any(target_os = "linux", target_os = "android")))]
837mod gso {
838    use super::*;
839
840    pub(super) fn max_gso_segments(_socket: &impl AsRawFd) -> usize {
841        1
842    }
843
844    #[cfg_attr(apple_fast, allow(dead_code))] // Unused when apple_fast is enabled
845    pub(super) fn set_segment_size(
846        #[cfg(not(apple_fast))] _encoder: &mut cmsg::Encoder<'_, libc::msghdr>,
847        #[cfg(apple_fast)] _encoder: &mut cmsg::Encoder<'_, msghdr_x>,
848        _segment_size: u16,
849    ) {
850    }
851}
852
853#[cfg(target_os = "freebsd")]
854type IpTosTy = libc::c_uchar;
855#[cfg(not(any(target_os = "freebsd", target_os = "netbsd")))]
856pub(crate) type IpTosTy = libc::c_int;
857
858/// Returns whether the given socket option is supported on the current platform
859///
860/// Yields `Ok(true)` if the option was set successfully, `Ok(false)` if setting
861/// the option raised an `ENOPROTOOPT` or `EOPNOTSUPP` error, and `Err` for any other error.
862fn set_socket_option_supported(
863    socket: &impl AsRawFd,
864    level: libc::c_int,
865    name: libc::c_int,
866    value: libc::c_int,
867) -> io::Result<bool> {
868    match set_socket_option(socket, level, name, value) {
869        Ok(()) => Ok(true),
870        Err(err) if err.raw_os_error() == Some(libc::ENOPROTOOPT) => Ok(false),
871        Err(err) if err.raw_os_error() == Some(libc::EOPNOTSUPP) => Ok(false),
872        Err(err) => Err(err),
873    }
874}
875
876pub(crate) fn set_socket_option(
877    socket: &impl AsRawFd,
878    level: libc::c_int,
879    name: libc::c_int,
880    value: libc::c_int,
881) -> io::Result<()> {
882    let rc = unsafe {
883        libc::setsockopt(
884            socket.as_raw_fd(),
885            level,
886            name,
887            &value as *const _ as _,
888            size_of_val(&value) as _,
889        )
890    };
891
892    match rc == 0 {
893        true => Ok(()),
894        false => Err(io::Error::last_os_error()),
895    }
896}
897
898const OPTION_ON: libc::c_int = 1;
899
900/// Calls `f` in a loop, retrying on `EINTR`.
901///
902/// Returns the non-negative result or the first non-`EINTR` error.
903pub(crate) fn retry_if_interrupted(mut f: impl FnMut() -> isize) -> io::Result<isize> {
904    loop {
905        let n = f();
906        if n >= 0 {
907            return Ok(n);
908        }
909        let e = io::Error::last_os_error();
910        if e.kind() != io::ErrorKind::Interrupted {
911            return Err(e);
912        }
913    }
914}
noq_udp/unix.rs

noq_udp/
unix.rs
