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

noq_udp/
unix.rs
