tokio_quiche/quic/io/
gso.rs

1// Copyright (C) 2025, Cloudflare, Inc.
2// All rights reserved.
3//
4// Redistribution and use in source and binary forms, with or without
5// modification, are permitted provided that the following conditions are
6// met:
7//
8//     * Redistributions of source code must retain the above copyright notice,
9//       this list of conditions and the following disclaimer.
10//
11//     * Redistributions in binary form must reproduce the above copyright
12//       notice, this list of conditions and the following disclaimer in the
13//       documentation and/or other materials provided with the distribution.
14//
15// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
16// IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
17// THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
18// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
19// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
20// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
21// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
22// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
23// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
24// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
25// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
26
27use std::io;
28use std::net::SocketAddr;
29use std::time::Instant;
30
31use foundations::telemetry::metrics::Counter;
32use foundations::telemetry::metrics::TimeHistogram;
33
34#[cfg(all(target_os = "linux", not(feature = "fuzzing")))]
35mod linux_imports {
36    pub(super) use nix::sys::socket::sendmsg;
37    pub(super) use nix::sys::socket::ControlMessage;
38    pub(super) use nix::sys::socket::MsgFlags;
39    pub(super) use nix::sys::socket::SockaddrStorage;
40    pub(super) use smallvec::SmallVec;
41    pub(super) use std::io::ErrorKind;
42    pub(super) use std::os::fd::AsRawFd;
43    pub(super) use tokio::io::Interest;
44}
45
46#[cfg(all(target_os = "linux", not(feature = "fuzzing")))]
47use self::linux_imports::*;
48
49// Maximum number of packets can be sent in UDP GSO.
50pub(crate) const UDP_MAX_SEGMENT_COUNT: usize = 64;
51
52#[cfg(not(feature = "gcongestion"))]
53/// Returns a new max send buffer size to avoid the fragmentation
54/// at the end. Maximum send buffer size is min(MAX_SEND_BUF_SIZE,
55/// connection's send_quantum).
56/// For example,
57///
58/// - max_send_buf = 1000 and mss = 100, return 1000
59/// - max_send_buf = 1000 and mss = 90, return 990
60///
61/// not to have last 10 bytes packet.
62pub(crate) fn tune_max_send_size(
63    segment_size: Option<usize>, send_quantum: usize, max_capacity: usize,
64) -> usize {
65    let max_send_buf_size = send_quantum.min(max_capacity);
66
67    if let Some(mss) = segment_size {
68        max_send_buf_size / mss * mss
69    } else {
70        max_send_buf_size
71    }
72}
73
74// https://wiki.cfdata.org/pages/viewpage.action?pageId=436188159
75pub(crate) const UDP_MAX_GSO_PACKET_SIZE: usize = 65507;
76
77#[cfg(all(target_os = "linux", not(feature = "fuzzing")))]
78#[derive(Copy, Clone, Debug)]
79pub(crate) enum PktInfo {
80    V4(libc::in_pktinfo),
81    V6(libc::in6_pktinfo),
82}
83
84#[cfg(all(target_os = "linux", not(feature = "fuzzing")))]
85impl PktInfo {
86    fn make_cmsg(&'_ self) -> ControlMessage<'_> {
87        match self {
88            Self::V4(pkt) => ControlMessage::Ipv4PacketInfo(pkt),
89            Self::V6(pkt) => ControlMessage::Ipv6PacketInfo(pkt),
90        }
91    }
92
93    fn from_socket_addr(addr: SocketAddr) -> Self {
94        match addr {
95            SocketAddr::V4(ipv4) => {
96                // This is basically a safe wrapper around `mem::transmute()`.
97                // Calling this on the raw octets will ensure they
98                // become a native-endian, kernel-readable u32
99                let s_addr = u32::from_ne_bytes(ipv4.ip().octets());
100
101                Self::V4(libc::in_pktinfo {
102                    ipi_ifindex: 0,
103                    ipi_spec_dst: libc::in_addr { s_addr },
104                    ipi_addr: libc::in_addr { s_addr: 0 },
105                })
106            },
107            SocketAddr::V6(ipv6) => Self::V6(libc::in6_pktinfo {
108                ipi6_ifindex: 0,
109                ipi6_addr: libc::in6_addr {
110                    s6_addr: ipv6.ip().octets(),
111                },
112            }),
113        }
114    }
115}
116
117#[cfg(all(target_os = "linux", not(feature = "fuzzing")))]
118#[allow(clippy::too_many_arguments)]
119pub async fn send_to(
120    socket: &tokio::net::UdpSocket, to: SocketAddr, from: Option<SocketAddr>,
121    send_buf: &[u8], segment_size: usize, tx_time: Option<Instant>,
122    would_block_metric: Counter, send_to_wouldblock_duration_s: TimeHistogram,
123) -> io::Result<usize> {
124    // SAFETY: This Linux-only code assumes `Instant` uses the same layout as a
125    // zero-valid timespec. The zero instant allows raw time extraction.
126    const INSTANT_ZERO: Instant = unsafe { std::mem::transmute(0u128) };
127
128    let mut sendmsg_retry_timer = None;
129    loop {
130        let iov = [std::io::IoSlice::new(send_buf)];
131        let segment_size_u16 = segment_size as u16;
132
133        let raw_time = tx_time
134            .map(|t| t.duration_since(INSTANT_ZERO).as_nanos() as u64)
135            .unwrap_or(0);
136
137        let pkt_info = from.map(PktInfo::from_socket_addr);
138
139        let mut cmsgs: SmallVec<[ControlMessage; 3]> = SmallVec::new();
140
141        // Create cmsg for UDP_SEGMENT.
142        cmsgs.push(ControlMessage::UdpGsoSegments(&segment_size_u16));
143
144        if tx_time.is_some() {
145            // Create cmsg for TXTIME.
146            cmsgs.push(ControlMessage::TxTime(&raw_time));
147        }
148
149        if let Some(pkt) = pkt_info.as_ref() {
150            // Create cmsg for IP(V6)_PKTINFO.
151            cmsgs.push(pkt.make_cmsg());
152        }
153
154        let addr = SockaddrStorage::from(to);
155
156        // Must use [`try_io`] so tokio can properly clear its readyness flag
157        let res = socket.try_io(Interest::WRITABLE, || {
158            let fd = socket.as_raw_fd();
159            sendmsg(fd, &iov, &cmsgs, MsgFlags::empty(), Some(&addr))
160                .map_err(Into::into)
161        });
162
163        match res {
164            // Wait for the socket to become writable and try again
165            Err(e) if e.kind() == ErrorKind::WouldBlock => {
166                if sendmsg_retry_timer.is_none() {
167                    sendmsg_retry_timer =
168                        Some(send_to_wouldblock_duration_s.start_timer());
169                }
170                would_block_metric.inc();
171                socket.writable().await?
172            },
173            res => return res,
174        }
175    }
176}
177
178#[cfg(any(not(target_os = "linux"), feature = "fuzzing"))]
179#[allow(clippy::too_many_arguments)]
180pub(crate) async fn send_to(
181    socket: &tokio::net::UdpSocket, to: SocketAddr, _from: Option<SocketAddr>,
182    send_buf: &[u8], _segment_size: usize, _tx_time: Option<Instant>,
183    _would_block_metric: Counter, _send_to_wouldblock_duration_s: TimeHistogram,
184) -> io::Result<usize> {
185    socket.send_to(send_buf, to).await
186}
187
188#[cfg(all(target_os = "linux", test))]
189mod test {
190    #[test]
191    /// If this test begins to fail, it means the implementation of [`Instant`]
192    /// has changed in the std library.
193    fn instant_zero() {
194        use std::time::Instant;
195
196        const INSTANT_ZERO: Instant = unsafe { std::mem::transmute(0u128) };
197        const NANOS_PER_SEC: u128 = 1_000_000_000;
198
199        // Define a [`Timespec`] similar to the one backing [`Instant`]
200        #[derive(Debug)]
201        struct Timespec {
202            tv_sec: i64,
203            tv_nsec: u32,
204        }
205
206        let now = Instant::now();
207        let now_timespec: Timespec = unsafe { std::mem::transmute(now) };
208
209        let ref_elapsed = now.duration_since(INSTANT_ZERO).as_nanos();
210        let raw_elapsed = now_timespec.tv_sec as u128 * NANOS_PER_SEC +
211            now_timespec.tv_nsec as u128;
212
213        assert_eq!(ref_elapsed, raw_elapsed);
214    }
215}
tokio_quiche/quic/io/gso.rs

tokio_quiche/quic/io/
gso.rs
