From 5fbbf4c8f30cba779135b269b1adceecaa9c4e99 Mon Sep 17 00:00:00 2001 From: Benjamin Saunders Date: Wed, 23 Oct 2019 18:41:51 -0700 Subject: [PATCH] Limit outgoing datagram buffer size --- quinn-proto/src/connection.rs | 26 +++++++++++-------------- quinn-proto/src/shared.rs | 14 ++++++++++--- quinn-proto/src/tests/mod.rs | 4 ++-- quinn-proto/src/transport_parameters.rs | 2 +- 4 files changed, 25 insertions(+), 21 deletions(-) diff --git a/quinn-proto/src/connection.rs b/quinn-proto/src/connection.rs index 33d14f270..bb5c594ff 100644 --- a/quinn-proto/src/connection.rs +++ b/quinn-proto/src/connection.rs @@ -356,7 +356,6 @@ where if ack.largest >= self.space(space).next_packet_number { return Err(TransportError::PROTOCOL_VIOLATION("unsent packet acked")); } - let was_congested = self.congestion_blocked(); let was_blocked = self.blocked(); let new_largest = { let space = self.space_mut(space); @@ -446,12 +445,6 @@ where self.events.push_back(Event::StreamWritable { stream }); } } - if was_congested - && !self.congestion_blocked() - && mem::replace(&mut self.datagrams.send_blocked, false) - { - self.events.push_back(Event::DatagramSendUnblocked); - } Ok(()) } @@ -1955,7 +1948,7 @@ where // TODO: Cache, or perhaps forward to user? } Frame::Datagram(datagram) => { - let window = match self.config.datagram_window { + let window = match self.config.datagram_receive_window { None => { return Err(TransportError::PROTOCOL_VIOLATION( "unexpected DATAGRAM frame", @@ -2576,6 +2569,10 @@ where self.datagrams.outgoing.push_front(datagram); break; } + if self.datagrams.outgoing_total >= self.config.datagram_send_window { + self.events.push_back(Event::DatagramSendUnblocked); + } + self.datagrams.outgoing_total -= datagram.data.len(); datagram.encode(true, buf); } @@ -2989,14 +2986,13 @@ where /// If `Err(SendDatagramError::Blocked)` is returned, `Event::DatagramSendUnblocked` may be /// emitted in the future. pub fn send_datagram(&mut self) -> Result, SendDatagramError> { - if self.config.datagram_window.is_none() { + if self.config.datagram_receive_window.is_none() { return Err(SendDatagramError::Disabled); } let max = self .max_datagram_size() .ok_or(SendDatagramError::UnsupportedByPeer)?; - if self.congestion_blocked() { - self.datagrams.send_blocked = true; + if self.datagrams.outgoing_total >= self.config.datagram_send_window { return Err(SendDatagramError::Blocked); } Ok(DatagramSender { max, conn: self }) @@ -3026,7 +3022,7 @@ where - 4 // worst-case packet number size - self.space(SpaceId::Data).crypto.as_ref().or(self.zero_rtt_crypto.as_ref()).unwrap().packet.tag_len() - Datagram::SIZE_BOUND; - self.config.datagram_window?; + self.config.datagram_receive_window?; let limit = self.params.max_datagram_frame_size?.into_inner(); Some(limit.min(max_size as u64) as usize) } @@ -3616,6 +3612,7 @@ impl DatagramSender<'_, S> { if data.len() > self.max { return Err(DatagramTooLarge); } + self.conn.datagrams.outgoing_total += data.len(); self.conn.datagrams.outgoing.push_back(Datagram { data }); Ok(()) } @@ -3625,19 +3622,18 @@ struct DatagramState { /// Number of bytes of datagrams that have been received by the local transport but not /// delivered to the application recv_buffered: usize, - /// Whether a `send_datagram` call failed due to congestion - send_blocked: bool, incoming: VecDeque, outgoing: VecDeque, + outgoing_total: usize, } impl DatagramState { fn new() -> Self { Self { recv_buffered: 0, - send_blocked: false, incoming: VecDeque::new(), outgoing: VecDeque::new(), + outgoing_total: 0, } } } diff --git a/quinn-proto/src/shared.rs b/quinn-proto/src/shared.rs index 965152545..c4d50dd80 100644 --- a/quinn-proto/src/shared.rs +++ b/quinn-proto/src/shared.rs @@ -118,12 +118,19 @@ pub struct TransportConfig { /// This allows passive observers to easily judge the round trip time of a connection, which can /// be useful for network administration but sacrifices a small amount of privacy. pub allow_spin: bool, - /// Maximum number of application datagram bytes to buffer, or None to disable datagrams + /// Maximum number of incoming application datagram bytes to buffer, or None to disable + /// datagrams /// /// The peer is forbidden to send single datagrams larger than this size. If the aggregate size /// of all datagrams that have been received from the peer but not consumed by the application /// exceeds this value, old datagrams are dropped until it is no longer exceeded. - pub datagram_window: Option, + pub datagram_receive_window: Option, + /// Maximum number of outgoing application datagram bytes to buffer + /// + /// While datagrams are sent ASAP, it is possible for an application to generate data faster + /// than the link, or even the underlying hardware, can transmit them. This limits the amount of + /// memory that may be consumed in that case. + pub datagram_send_window: usize, } impl Default for TransportConfig { @@ -160,7 +167,8 @@ impl Default for TransportConfig { keep_alive_interval: 0, crypto_buffer_size: 16 * 1024, allow_spin: true, - datagram_window: Some(STREAM_RWND as usize), + datagram_receive_window: Some(STREAM_RWND as usize), + datagram_send_window: 1 * 1024 * 1024, } } } diff --git a/quinn-proto/src/tests/mod.rs b/quinn-proto/src/tests/mod.rs index 9d8ef40b2..8f43ebebe 100644 --- a/quinn-proto/src/tests/mod.rs +++ b/quinn-proto/src/tests/mod.rs @@ -1175,7 +1175,7 @@ fn datagram_window() { const WINDOW: usize = 100; let server = ServerConfig { transport: Arc::new(TransportConfig { - datagram_window: Some(WINDOW), + datagram_receive_window: Some(WINDOW), ..TransportConfig::default() }), ..server_config() @@ -1238,7 +1238,7 @@ fn datagram_window() { fn datagram_unsupported() { let server = ServerConfig { transport: Arc::new(TransportConfig { - datagram_window: None, + datagram_receive_window: None, ..TransportConfig::default() }), ..server_config() diff --git a/quinn-proto/src/transport_parameters.rs b/quinn-proto/src/transport_parameters.rs index fd9d5301d..983e0667d 100644 --- a/quinn-proto/src/transport_parameters.rs +++ b/quinn-proto/src/transport_parameters.rs @@ -89,7 +89,7 @@ impl TransportParameters { disable_active_migration: server_config.map_or(false, |c| !c.migration), active_connection_id_limit: REM_CID_COUNT, max_datagram_frame_size: config - .datagram_window + .datagram_receive_window .map(|x| (x.min(u16::max_value().into()) as u16).into()), ..Self::default() }