From d075733dda3cf89abed643ea3c93cd88dc51e10a Mon Sep 17 00:00:00 2001 From: Benjamin Saunders Date: Mon, 18 Dec 2023 19:55:18 -0800 Subject: [PATCH] Don't buffer endpoint-generated datagrams Dropping these has low cost because they're not associated with any connection. --- quinn/src/endpoint.rs | 140 ++++++++++++------------------------------ quinn/src/lib.rs | 8 --- 2 files changed, 39 insertions(+), 109 deletions(-) diff --git a/quinn/src/endpoint.rs b/quinn/src/endpoint.rs index bb76faa53..589eef2ca 100644 --- a/quinn/src/endpoint.rs +++ b/quinn/src/endpoint.rs @@ -13,7 +13,7 @@ use std::{ }; use crate::{ - runtime::{default_runtime, AsyncUdpSocket, Runtime, UdpPoller}, + runtime::{default_runtime, AsyncUdpSocket, Runtime}, udp_transmit, }; use bytes::{Bytes, BytesMut}; @@ -29,8 +29,7 @@ use udp::{RecvMeta, BATCH_SIZE}; use crate::{ connection::Connecting, incoming::Incoming, work_limiter::WorkLimiter, ConnectionEvent, - EndpointConfig, VarInt, IO_LOOP_BOUND, MAX_INCOMING_CONNECTIONS, - MAX_TRANSMIT_QUEUE_CONTENTS_LEN, RECV_TIME_BOUND, SEND_TIME_BOUND, + EndpointConfig, VarInt, IO_LOOP_BOUND, MAX_INCOMING_CONNECTIONS, RECV_TIME_BOUND, }; /// A QUIC endpoint. @@ -224,7 +223,6 @@ impl Endpoint { let addr = socket.local_addr()?; let mut inner = self.inner.state.lock().unwrap(); inner.prev_socket = Some(mem::replace(&mut inner.socket, socket)); - inner.io_poller = inner.socket.clone().create_io_poller(); inner.ipv6 = addr.is_ipv6(); // Update connection socket references @@ -330,7 +328,6 @@ impl Future for EndpointDriver { let mut keep_going = false; keep_going |= endpoint.drive_recv(cx, now)?; keep_going |= endpoint.handle_events(cx, &self.0.shared); - keep_going |= endpoint.drive_send(cx)?; if !endpoint.recv_state.incoming.is_empty() { self.0.shared.incoming.notify_waiters(); @@ -392,7 +389,7 @@ impl EndpointInner { } Err(error) => { if let Some(transmit) = error.response { - state.transmit_state.respond(transmit, response_buffer); + respond(transmit, &mut response_buffer, &*state.socket); } Err(error.cause) } @@ -403,14 +400,14 @@ impl EndpointInner { let mut state = self.state.lock().unwrap(); let mut response_buffer = BytesMut::new(); let transmit = state.inner.refuse(incoming, &mut response_buffer); - state.transmit_state.respond(transmit, response_buffer); + respond(transmit, &mut response_buffer, &*state.socket); } pub(crate) fn retry(&self, incoming: proto::Incoming) -> Result<(), proto::RetryError> { let mut state = self.state.lock().unwrap(); let mut response_buffer = BytesMut::new(); let transmit = state.inner.retry(incoming, &mut response_buffer)?; - state.transmit_state.respond(transmit, response_buffer); + respond(transmit, &mut response_buffer, &*state.socket); Ok(()) } } @@ -421,9 +418,7 @@ pub(crate) struct State { /// During an active migration, abandoned_socket receives traffic /// until the first packet arrives on the new socket. prev_socket: Option>, - io_poller: Pin>, inner: proto::Endpoint, - transmit_state: TransmitState, recv_state: RecvState, driver: Option, ipv6: bool, @@ -431,7 +426,6 @@ pub(crate) struct State { /// Number of live handles that can be used to initiate or handle I/O; excludes the driver ref_count: usize, driver_lost: bool, - send_limiter: WorkLimiter, runtime: Arc, } @@ -446,24 +440,16 @@ impl State { self.recv_state.recv_limiter.start_cycle(); if let Some(socket) = &self.prev_socket { // We don't care about the `PollProgress` from old sockets. - let poll_res = self.recv_state.poll_socket( - cx, - &mut self.inner, - &mut self.transmit_state, - &**socket, - now, - ); + let poll_res = self + .recv_state + .poll_socket(cx, &mut self.inner, &**socket, now); if poll_res.is_err() { self.prev_socket = None; } }; - let poll_res = self.recv_state.poll_socket( - cx, - &mut self.inner, - &mut self.transmit_state, - &*self.socket, - now, - ); + let poll_res = self + .recv_state + .poll_socket(cx, &mut self.inner, &*self.socket, now); self.recv_state.recv_limiter.finish_cycle(); let poll_res = poll_res?; if poll_res.received_connection_packet { @@ -474,42 +460,6 @@ impl State { Ok(poll_res.keep_going) } - fn drive_send(&mut self, cx: &mut Context) -> Result { - self.send_limiter.start_cycle(); - - let result = loop { - if self.transmit_state.outgoing.is_empty() { - break Ok(false); - } - - if !self.send_limiter.allow_work() { - break Ok(true); - } - - if self.io_poller.as_mut().poll_writable(cx)?.is_pending() { - break Ok(false); - } - - match self.socket.try_send(self.transmit_state.transmits()) { - Ok(n) => { - self.transmit_state.dequeue(n); - // We count transmits instead of `try_send` calls since the cost - // of a `sendmmsg` still linearly increases with number of packets. - self.send_limiter.record_work(n); - } - Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => { - continue; - } - Err(e) => { - break Err(e); - } - } - }; - - self.send_limiter.finish_cycle(); - result - } - fn handle_events(&mut self, cx: &mut Context, shared: &Shared) -> bool { for _ in 0..IO_LOOP_BOUND { let (ch, event) = match self.events.poll_recv(cx) { @@ -543,40 +493,32 @@ impl State { } } -#[derive(Debug, Default)] -struct TransmitState { - outgoing: VecDeque, - /// The aggregateed contents length of the packets in the transmit queue - contents_len: usize, -} - -impl TransmitState { - fn respond(&mut self, transmit: proto::Transmit, mut response_buffer: BytesMut) { - // Limiting the memory usage for items queued in the outgoing queue from endpoint - // generated packets. Otherwise, we may see a build-up of the queue under test with - // flood of initial packets against the endpoint. The sender with the sender-limiter - // may not keep up the pace of these packets queued into the queue. - if self.contents_len >= MAX_TRANSMIT_QUEUE_CONTENTS_LEN { - return; - } - - let contents_len = transmit.size; - self.outgoing.push_back(udp_transmit( - transmit, - response_buffer.split_to(contents_len).freeze(), - )); - self.contents_len = self.contents_len.saturating_add(contents_len); - } - - fn dequeue(&mut self, sent: usize) { - self.contents_len = self - .contents_len - .saturating_sub(self.outgoing.drain(..sent).map(|t| t.contents.len()).sum()); - } - - fn transmits(&self) -> &[udp::Transmit] { - self.outgoing.as_slices().0 - } +fn respond(transmit: proto::Transmit, response_buffer: &mut BytesMut, socket: &dyn AsyncUdpSocket) { + // Send if there's kernel buffer space; otherwise, drop it + // + // As an endpoint-generated packet, we know this is an + // immediate, stateless response to an unconnected peer, + // one of: + // + // - A version negotiation response due to an unknown version + // - A `CLOSE` due to a malformed or unwanted connection attempt + // - A stateless reset due to an unrecognized connection + // - A `Retry` packet due to a connection attempt when + // `use_retry` is set + // + // In each case, a well-behaved peer can be trusted to retry a + // few times, which is guaranteed to produce the same response + // from us. Repeated failures might at worst cause a peer's new + // connection attempt to time out, which is acceptable if we're + // under such heavy load that there's never room for this code + // to transmit. This is morally equivalent to the packet getting + // lost due to congestion further along the link, which + // similarly relies on peer retries for recovery. + let contents_len = transmit.size; + _ = socket.try_send(&[udp_transmit( + transmit, + response_buffer.split_to(contents_len).freeze(), + )]); } #[inline] @@ -687,17 +629,14 @@ impl EndpointRef { idle: Notify::new(), }, state: Mutex::new(State { - io_poller: socket.clone().create_io_poller(), socket, prev_socket: None, inner, - transmit_state: TransmitState::default(), ipv6, events, driver: None, ref_count: 0, driver_lost: false, - send_limiter: WorkLimiter::new(SEND_TIME_BOUND), recv_state, runtime, }), @@ -772,7 +711,6 @@ impl RecvState { &mut self, cx: &mut Context, endpoint: &mut proto::Endpoint, - transmit_state: &mut TransmitState, socket: &dyn AsyncUdpSocket, now: Instant, ) -> Result { @@ -814,7 +752,7 @@ impl RecvState { } else { let transmit = endpoint.refuse(incoming, &mut response_buffer); - transmit_state.respond(transmit, response_buffer); + respond(transmit, &mut response_buffer, socket); } } Some(DatagramEvent::ConnectionEvent(handle, event)) => { @@ -828,7 +766,7 @@ impl RecvState { .send(ConnectionEvent::Proto(event)); } Some(DatagramEvent::Response(transmit)) => { - transmit_state.respond(transmit, response_buffer); + respond(transmit, &mut response_buffer, socket); } None => {} } diff --git a/quinn/src/lib.rs b/quinn/src/lib.rs index 168a754a7..6f6c39b00 100644 --- a/quinn/src/lib.rs +++ b/quinn/src/lib.rs @@ -132,14 +132,6 @@ const IO_LOOP_BOUND: usize = 160; /// batch of size 32 was observed to take 30us on some systems. const RECV_TIME_BOUND: Duration = Duration::from_micros(50); -/// The maximum amount of time that should be spent in `sendmsg()` calls per endpoint iteration -const SEND_TIME_BOUND: Duration = Duration::from_micros(50); - -/// The maximum size of content length of packets in the outgoing transmit queue. Transmit packets -/// generated from the endpoint (retry or initial close) can be dropped when this limit is being execeeded. -/// Chose to represent 100 MB of data. -const MAX_TRANSMIT_QUEUE_CONTENTS_LEN: usize = 100_000_000; - /// The maximum number of `IncomingConnection`s we allow to be enqueued at a time before we start /// rejecting new `IncomingConnection`s automatically. Assuming each `IncomingConnection` accounts /// for little over 1200 bytes of memory maximum, this should limit an endpoint's incoming