Merge remote-tracking branch 'n0/main' into encoder-helper

This commit is contained in:
Diva Martínez
2025-12-19 10:02:03 -05:00
19 changed files with 100 additions and 62 deletions
+1 -1
View File
@@ -14,7 +14,7 @@ struct CidData(ConnectionId, Option<ResetToken>);
/// - Zero to `Self::LEN - 1` reserved CIDs from `self.cursor` up to `self.cursor_reserved`.
/// - More "available"/"ready" CIDs after `self.cursor_reserved`.
///
/// The range of reserved CIDs is grown by calling [`CidQueue::next_reserved`], which takes one of
/// The range of reserved CIDs is grown by calling `CidQueue::next_reserved`, which takes one of
/// the available ones and returns the CID that was reserved.
///
/// New available/ready CIDs are added by calling [`CidQueue::insert`].
+1 -1
View File
@@ -10,7 +10,7 @@ use crate::{ConnectionId, Instant, Side};
///
/// This is set via [`TransportConfig::qlog_factory`].
///
/// [`TransportConfig::qlog_factory]: crate::config::TransportConfig::qlog_factory
/// [`TransportConfig::qlog_factory`]: crate::TransportConfig::qlog_factory
pub trait QlogFactory: Send + Sync + 'static {
/// Returns a [`QlogConfig`] for a connection, if logging should be enabled.
///
+4 -6
View File
@@ -4,7 +4,7 @@ use std::{
convert::TryFrom,
fmt, io, mem,
net::{IpAddr, SocketAddr},
num::NonZeroU32,
num::{NonZeroU32, NonZeroUsize},
ops::Not,
sync::Arc,
};
@@ -888,7 +888,7 @@ impl Connection {
pub fn poll_transmit(
&mut self,
now: Instant,
max_datagrams: usize,
max_datagrams: NonZeroUsize,
buf: &mut Vec<u8>,
) -> Option<Transmit> {
if let Some(probing) = self
@@ -911,9 +911,8 @@ impl Connection {
});
}
assert!(max_datagrams != 0);
let max_datagrams = match self.config.enable_segmentation_offload {
false => 1,
false => NonZeroUsize::MIN,
true => max_datagrams,
};
@@ -1154,7 +1153,7 @@ impl Connection {
// If the datagram is full, we need to start a new one.
if transmit.datagram_remaining_mut() == 0 {
if transmit.num_datagrams() >= transmit.max_datagrams() {
if transmit.num_datagrams() >= transmit.max_datagrams().get() {
// No more datagrams allowed
break;
}
@@ -3831,7 +3830,6 @@ impl Connection {
self.stats.frame_rx.record(&frame);
if let Frame::Close(_error) = frame {
trace!("draining");
self.state.move_to_draining(None);
break;
}
+5 -3
View File
@@ -141,7 +141,9 @@ pub(super) struct PathData {
pub(super) pacing: Pacer,
/// Actually sent challenges (on the wire).
pub(super) challenges_sent: IntMap<u64, SentChallengeInfo>,
/// Whether to *immediately* trigger another PATH_CHALLENGE (via [`super::Connection::can_send`])
/// Whether to *immediately* trigger another PATH_CHALLENGE.
///
/// This is picked up by [`super::Connection::space_can_send`].
pub(super) send_new_challenge: bool,
/// Pending responses to PATH_CHALLENGE frames
pub(super) path_responses: PathResponses,
@@ -273,8 +275,8 @@ impl PathData {
status: Default::default(),
first_packet: None,
pto_count: 0,
idle_timeout: None,
keep_alive: None,
idle_timeout: config.default_path_max_idle_timeout,
keep_alive: config.default_path_keep_alive_interval,
open: false,
last_allowed_receive: None,
#[cfg(feature = "qlog")]
+3 -2
View File
@@ -1,7 +1,8 @@
//! Implements support for emitting qlog events.
//!
//! This uses the [`n0-qlog`] crate to emit qlog events. The n0-qlog crate, and thus this implementation,
//! is currently based on [draft-ietf-quic-qlog-main-schema-13] an [draft-ietf-quic-qlog-quic-events-12].
//! This uses the [`qlog`] crate to emit qlog events. The n0-qlog crate, and thus this
//! implementation, is currently based on [draft-ietf-quic-qlog-main-schema-13] an
//! [draft-ietf-quic-qlog-quic-events-12].
//!
//! [draft-ietf-quic-qlog-main-schema-13]: https://www.ietf.org/archive/id/draft-ietf-quic-qlog-main-schema-13.html
//! [draft-ietf-quic-qlog-quic-events-12]: https://www.ietf.org/archive/id/draft-ietf-quic-qlog-quic-events-12.html
+16 -5
View File
@@ -1,4 +1,5 @@
use bytes::Bytes;
use tracing::trace;
use crate::frame::Close;
use crate::{ApplicationClose, ConnectionClose, ConnectionError, TransportError, TransportErrorCode};
@@ -54,13 +55,15 @@ impl State {
pub(super) fn move_to_handshake(&mut self, hs: Handshake) {
self.inner = InnerState::Handshake(hs);
trace!("connection state: handshake");
}
pub(super) fn move_to_established(&mut self) {
self.inner = InnerState::Established;
trace!("connection state: established");
}
/// Moves to a draining state.
/// Moves to the drained state.
///
/// Panics if the state was already drained.
pub(super) fn move_to_drained(&mut self, error: Option<ConnectionError>) {
@@ -93,6 +96,7 @@ impl State {
(error, self.is_local_close())
};
self.inner = InnerState::Drained { error, is_local };
trace!("connection state: drained");
}
/// Moves to a draining state.
@@ -109,6 +113,7 @@ impl State {
);
let is_local = self.is_local_close();
self.inner = InnerState::Draining { error, is_local };
trace!("connection state: draining");
}
fn is_local_close(&self) -> bool {
@@ -133,10 +138,13 @@ impl State {
"invalid state transition {:?} -> closed",
self.as_type()
);
let remote_reason = reason.into();
let is_local = false;
trace!(?remote_reason, ?is_local, "connection state: closed");
self.inner = InnerState::Closed {
error_read: false,
remote_reason: reason.into(),
is_local: false,
remote_reason,
is_local,
};
}
@@ -152,10 +160,13 @@ impl State {
"invalid state transition {:?} -> closed (local)",
self.as_type()
);
let remote_reason = reason.into();
let is_local = true;
trace!(?remote_reason, ?is_local, "connection state: closed");
self.inner = InnerState::Closed {
error_read: false,
remote_reason: reason.into(),
is_local: true,
remote_reason,
is_local,
};
}
+8 -6
View File
@@ -1,3 +1,5 @@
use std::num::NonZeroUsize;
use bytes::BufMut;
use tracing::trace;
@@ -40,7 +42,7 @@ pub(super) struct TransmitBuf<'a> {
/// size. All datagrams in between need to be exactly this size.
buf_capacity: usize,
/// The maximum number of datagrams allowed to write into [`TransmitBuf::buf`]
max_datagrams: usize,
max_datagrams: NonZeroUsize,
/// The number of datagrams already (partially) written into the buffer
///
/// Incremented by a call to [`TransmitBuf::start_new_datagram`].
@@ -57,7 +59,7 @@ pub(super) struct TransmitBuf<'a> {
}
impl<'a> TransmitBuf<'a> {
pub(super) fn new(buf: &'a mut Vec<u8>, max_datagrams: usize, mtu: usize) -> Self {
pub(super) fn new(buf: &'a mut Vec<u8>, max_datagrams: NonZeroUsize, mtu: usize) -> Self {
Self {
buf,
datagram_start: 0,
@@ -106,12 +108,12 @@ impl<'a> TransmitBuf<'a> {
// (e.g. purely containing ACKs), modern memory allocators (e.g. mimalloc and
// jemalloc) will pool certain allocation sizes and therefore this is still rather
// efficient.
let max_capacity_hint = self.max_datagrams * self.segment_size;
let max_capacity_hint = self.max_datagrams.get() * self.segment_size;
self.new_datagram_inner(self.segment_size, max_capacity_hint)
}
fn new_datagram_inner(&mut self, datagram_size: usize, max_capacity_hint: usize) {
debug_assert!(self.num_datagrams < self.max_datagrams);
debug_assert!(self.num_datagrams < self.max_datagrams.into());
if self.num_datagrams == 1 {
// Set the segment size to the size of the first datagram.
self.segment_size = self.buf.len();
@@ -121,7 +123,7 @@ impl<'a> TransmitBuf<'a> {
if datagram_size < self.segment_size {
// If this is a GSO batch and this datagram is smaller than the segment
// size, this must be the last datagram in the batch.
self.max_datagrams = self.num_datagrams + 1;
self.max_datagrams = NonZeroUsize::MIN.saturating_add(self.num_datagrams);
}
}
self.datagram_start = self.buf.len();
@@ -181,7 +183,7 @@ impl<'a> TransmitBuf<'a> {
}
/// Returns the maximum number of datagrams allowed to be written into the buffer
pub(super) fn max_datagrams(&self) -> usize {
pub(super) fn max_datagrams(&self) -> NonZeroUsize {
self.max_datagrams
}
+2 -1
View File
@@ -5,6 +5,7 @@ use std::{
io::{self, Write},
mem,
net::{Ipv6Addr, SocketAddr, UdpSocket},
num::NonZeroUsize,
ops::RangeFrom,
str,
sync::{Arc, LazyLock, Mutex},
@@ -791,7 +792,7 @@ pub(super) fn min_opt<T: Ord>(x: Option<T>, y: Option<T>) -> Option<T> {
}
/// The maximum of datagrams TestEndpoint will produce via `poll_transmit`
const MAX_DATAGRAMS: usize = 10;
const MAX_DATAGRAMS: NonZeroUsize = NonZeroUsize::new(10).expect("known");
fn split_transmit(transmit: Transmit, buffer: &[u8]) -> Vec<(Transmit, Bytes)> {
let mut buffer = Bytes::copy_from_slice(buffer);
+6 -4
View File
@@ -2,6 +2,7 @@ use std::{
cmp::min,
io::{ErrorKind, IoSliceMut},
net::{Ipv4Addr, Ipv6Addr, UdpSocket},
num::NonZeroUsize,
};
use criterion::{Criterion, criterion_group, criterion_main};
@@ -54,9 +55,9 @@ pub fn criterion_benchmark(c: &mut Criterion) {
let gso_segments = if gso_enabled {
send_state.max_gso_segments()
} else {
1
NonZeroUsize::MIN
};
let msg = vec![0xAB; min(MAX_DATAGRAM_SIZE, SEGMENT_SIZE * gso_segments)];
let msg = vec![0xAB; min(MAX_DATAGRAM_SIZE, SEGMENT_SIZE * gso_segments.get())];
let transmit = Transmit {
destination: dst_addr,
ecn: None,
@@ -67,13 +68,14 @@ pub fn criterion_benchmark(c: &mut Criterion) {
let gro_segments = if gro_enabled {
recv_state.gro_segments()
} else {
1
NonZeroUsize::MIN
};
let batch_size = if recvmmsg_enabled { BATCH_SIZE } else { 1 };
group.bench_function("throughput", |b| {
b.to_async(&rt).iter(|| async {
let mut receive_buffers = vec![vec![0; SEGMENT_SIZE * gro_segments]; batch_size];
let mut receive_buffers =
vec![vec![0; SEGMENT_SIZE * gro_segments.get()]; batch_size];
let mut receive_slices = receive_buffers
.iter_mut()
.map(|buf| IoSliceMut::new(buf))
+2 -2
View File
@@ -83,8 +83,8 @@ impl UdpSocketState {
}
#[inline]
pub fn gro_segments(&self) -> usize {
1
pub fn gro_segments(&self) -> NonZeroUsize {
NonZeroUsize::MIN
}
/// Resize the send buffer of `socket` to `bytes`
+17 -11
View File
@@ -4,6 +4,7 @@ use std::{
io::{self, IoSliceMut},
mem::{self, MaybeUninit},
net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6},
num::NonZeroUsize,
os::unix::io::AsRawFd,
sync::{
Mutex,
@@ -71,7 +72,7 @@ type IpTosTy = libc::c_int;
pub struct UdpSocketState {
last_send_error: Mutex<Instant>,
max_gso_segments: AtomicUsize,
gro_segments: usize,
gro_segments: NonZeroUsize,
may_fragment: bool,
/// True if we have received EINVAL error from `sendmsg` system call at least once.
@@ -244,8 +245,11 @@ impl UdpSocketState {
/// This is 1 if the platform doesn't support GSO. Subject to change if errors are detected
/// while using GSO.
#[inline]
pub fn max_gso_segments(&self) -> usize {
self.max_gso_segments.load(Ordering::Relaxed)
pub fn max_gso_segments(&self) -> NonZeroUsize {
self.max_gso_segments
.load(Ordering::Relaxed)
.try_into()
.expect("must have non zero GSO segments")
}
/// The number of segments to read when GRO is enabled. Used as a factor to
@@ -253,7 +257,7 @@ impl UdpSocketState {
///
/// Returns 1 if the platform doesn't support GRO.
#[inline]
pub fn gro_segments(&self) -> usize {
pub fn gro_segments(&self) -> NonZeroUsize {
self.gro_segments
}
@@ -354,7 +358,7 @@ fn send(
if let Some(libc::EIO) | Some(libc::EINVAL) = e.raw_os_error() {
// Prevent new transmits from being scheduled using GSO. Existing GSO transmits
// may already be in the pipeline, so we need to tolerate additional failures.
if state.max_gso_segments() > 1 {
if state.max_gso_segments().get() > 1 {
crate::log::info!(
"`libc::sendmsg` failed with {e}; halting segmentation offload"
);
@@ -1023,12 +1027,12 @@ mod gro {
// TODO: Add this to libc
pub(crate) const UDP_GRO: libc::c_int = 104;
pub(crate) fn gro_segments() -> usize {
pub(crate) fn gro_segments() -> NonZeroUsize {
let socket = match std::net::UdpSocket::bind("[::]:0")
.or_else(|_| std::net::UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)))
{
Ok(socket) => socket,
Err(_) => return 1,
Err(_) => return NonZeroUsize::MIN,
};
// As defined in net/ipv4/udp_offload.c
@@ -1039,8 +1043,8 @@ mod gro {
// list the kernel might potentially produce. See
// https://github.com/quinn-rs/quinn/pull/1354.
match set_socket_option(&socket, libc::SOL_UDP, UDP_GRO, OPTION_ON) {
Ok(()) => 64,
Err(_) => 1,
Ok(()) => NonZeroUsize::new(64).expect("known"),
Err(_) => NonZeroUsize::MIN,
}
}
}
@@ -1089,7 +1093,9 @@ const OPTION_ON: libc::c_int = 1;
#[cfg(not(any(target_os = "linux", target_os = "android")))]
mod gro {
pub(super) fn gro_segments() -> usize {
1
use std::num::NonZeroUsize;
pub(super) fn gro_segments() -> NonZeroUsize {
NonZeroUsize::MIN
}
}
+9 -5
View File
@@ -2,6 +2,7 @@ use std::{
io::{self, IoSliceMut},
mem,
net::{IpAddr, Ipv4Addr},
num::NonZeroUsize,
os::windows::io::AsRawSocket,
ptr,
sync::{
@@ -286,8 +287,11 @@ impl UdpSocketState {
/// This is 1 if the platform doesn't support GSO. Subject to change if errors are detected
/// while using GSO.
#[inline]
pub fn max_gso_segments(&self) -> usize {
self.max_gso_segments.load(Ordering::Relaxed)
pub fn max_gso_segments(&self) -> NonZeroUsize {
self.max_gso_segments
.load(Ordering::Relaxed)
.try_into()
.expect("must have non zero GSO segments")
}
/// The number of segments to read when GRO is enabled. Used as a factor to
@@ -295,9 +299,9 @@ impl UdpSocketState {
///
/// Returns 1 if the platform doesn't support GRO.
#[inline]
pub fn gro_segments(&self) -> usize {
pub fn gro_segments(&self) -> NonZeroUsize {
// Arbitrary reasonable value inspired by Linux and msquic
64
NonZeroUsize::new(64).expect("known")
}
/// Resize the send buffer of `socket` to `bytes`
@@ -426,7 +430,7 @@ fn send(state: &UdpSocketState, socket: UdpSockRef<'_>, transmit: &Transmit<'_>)
// GSO send failed. Some older versions of Windows report GSO support but
// fail on sending. Disable GSO for future sends. Existing GSO transmits may
// already be in the pipeline, so we need to tolerate additional failures.
if state.max_gso_segments() > 1 {
if state.max_gso_segments().get() > 1 {
crate::log::info!("WSASendMsg failed with {err}; halting segmentation offload");
state.max_gso_segments.store(1, Ordering::Relaxed);
}
+2 -2
View File
@@ -195,7 +195,7 @@ fn gso() {
.max_gso_segments();
let dst_addr = recv.local_addr().unwrap();
const SEGMENT_SIZE: usize = 128;
let msg = vec![0xAB; SEGMENT_SIZE * max_segments];
let msg = vec![0xAB; SEGMENT_SIZE * max_segments.get()];
test_send_recv(
&send.into(),
&recv.into(),
@@ -293,7 +293,7 @@ fn test_send_recv(send: &Socket, recv: &Socket, transmit: Transmit) {
Ok(_) => (),
Err(err) if err.kind() == io::ErrorKind::InvalidInput && cfg!(target_os = "windows") => {
// GSO can fail on windows. It should have disabled GSO now.
assert_eq!(send_state.max_gso_segments(), 1);
assert_eq!(send_state.max_gso_segments().get(), 1);
return;
}
Err(err) => panic!("send failed: {err:?}"),
+2 -1
View File
@@ -4,6 +4,7 @@ use std::{
future::Future,
io,
net::{IpAddr, SocketAddr},
num::NonZeroUsize,
pin::Pin,
sync::{Arc, Weak},
task::{Context, Poll, Waker, ready},
@@ -1728,4 +1729,4 @@ const MAX_TRANSMIT_DATAGRAMS: usize = 20;
/// This can be lower than the maximum platform capabilities, to avoid excessive
/// memory allocations when calling `poll_transmit()`. Benchmarks have shown
/// that numbers around 10 are a good compromise.
const MAX_TRANSMIT_SEGMENTS: usize = 10;
const MAX_TRANSMIT_SEGMENTS: NonZeroUsize = NonZeroUsize::new(10).expect("known");
+3 -2
View File
@@ -5,6 +5,7 @@ use std::{
io::{self, IoSliceMut},
mem,
net::{SocketAddr, SocketAddrV6},
num::NonZeroUsize,
pin::Pin,
str,
sync::{Arc, Mutex},
@@ -798,13 +799,13 @@ struct RecvState {
impl RecvState {
fn new(
sender: mpsc::UnboundedSender<(ConnectionHandle, EndpointEvent)>,
max_receive_segments: usize,
max_receive_segments: NonZeroUsize,
endpoint: &proto::Endpoint,
) -> Self {
let recv_buf = vec![
0;
endpoint.config().get_max_udp_payload_size().min(64 * 1024) as usize
* max_receive_segments
* max_receive_segments.get()
* BATCH_SIZE
];
Self {
+6
View File
@@ -211,6 +211,12 @@ impl Path {
let state = self.conn.state.lock("per_path_remote_address");
state.inner.path_remote_address(self.id)
}
/// Ping the remote endpoint over this path.
pub fn ping(&self) -> Result<(), ClosedPath> {
let mut state = self.conn.state.lock("ping");
state.inner.ping_path(self.id)
}
}
/// Future produced by [`Path::close`]
+7 -6
View File
@@ -5,6 +5,7 @@ use std::{
future::Future,
io::{self, IoSliceMut},
net::SocketAddr,
num::NonZeroUsize,
pin::Pin,
task::{Context, Poll},
};
@@ -65,8 +66,8 @@ pub trait AsyncUdpSocket: Send + Sync + Debug + 'static {
fn local_addr(&self) -> io::Result<SocketAddr>;
/// Maximum number of datagrams that might be described by a single [`RecvMeta`]
fn max_receive_segments(&self) -> usize {
1
fn max_receive_segments(&self) -> NonZeroUsize {
NonZeroUsize::MIN
}
/// Whether datagrams might get fragmented into multiple parts
@@ -100,8 +101,8 @@ pub trait UdpSender: Send + Sync + Debug + 'static {
) -> Poll<io::Result<()>>;
/// Maximum number of datagrams that a [`Transmit`] may encode.
fn max_transmit_segments(&self) -> usize {
1
fn max_transmit_segments(&self) -> NonZeroUsize {
NonZeroUsize::MIN
}
}
@@ -195,7 +196,7 @@ where
}
}
fn max_transmit_segments(&self) -> usize {
fn max_transmit_segments(&self) -> NonZeroUsize {
self.socket.max_transmit_segments()
}
}
@@ -212,7 +213,7 @@ trait UdpSenderHelperSocket: Send + Sync + 'static {
fn try_send(&self, transmit: &udp::Transmit) -> io::Result<()>;
/// See [`UdpSender::max_transmit_segments`].
fn max_transmit_segments(&self) -> usize;
fn max_transmit_segments(&self) -> NonZeroUsize;
}
/// Automatically select an appropriate runtime from those enabled at compile time
+3 -2
View File
@@ -1,5 +1,6 @@
use std::{
future::Future,
num::NonZeroUsize,
pin::Pin,
task::{Context, Poll},
time::Instant,
@@ -56,7 +57,7 @@ impl UdpSocket {
}
impl UdpSenderHelperSocket for UdpSocket {
fn max_transmit_segments(&self) -> usize {
fn max_transmit_segments(&self) -> NonZeroUsize {
self.inner.max_gso_segments()
}
@@ -95,7 +96,7 @@ impl AsyncUdpSocket for UdpSocket {
self.inner.may_fragment()
}
fn max_receive_segments(&self) -> usize {
fn max_receive_segments(&self) -> NonZeroUsize {
self.inner.gro_segments()
}
}
+3 -2
View File
@@ -2,6 +2,7 @@ use std::{
fmt::Debug,
future::Future,
io,
num::NonZeroUsize,
pin::Pin,
sync::Arc,
task::{Context, Poll, ready},
@@ -56,7 +57,7 @@ struct UdpSocket {
}
impl UdpSenderHelperSocket for UdpSocket {
fn max_transmit_segments(&self) -> usize {
fn max_transmit_segments(&self) -> NonZeroUsize {
self.inner.max_gso_segments()
}
@@ -99,7 +100,7 @@ impl AsyncUdpSocket for UdpSocket {
self.inner.may_fragment()
}
fn max_receive_segments(&self) -> usize {
fn max_receive_segments(&self) -> NonZeroUsize {
self.inner.gro_segments()
}
}