perf(transport): auto-tune stream receive window

neqo-http3/src/connection_client.rs

@@ -1279,7 +1279,7 @@ mod tests {
     use neqo_qpack::{encoder::QPackEncoder, QpackSettings};
     use neqo_transport::{
         CloseReason, ConnectionEvent, ConnectionParameters, Output, State, StreamId, StreamType,
-        Version, MIN_INITIAL_PACKET_SIZE, RECV_BUFFER_SIZE, SEND_BUFFER_SIZE,
+        Version, INITIAL_RECV_WINDOW_SIZE, MIN_INITIAL_PACKET_SIZE,
     };
     use test_fixture::{
         anti_replay, default_server_h3, fixture_init, new_server, now,
@@ -2730,7 +2730,7 @@ mod tests {
             if let ConnectionEvent::RecvStreamReadable { stream_id } = e {
                 if stream_id == request_stream_id {
                     // Read the DATA frame.
-                    let mut buf = vec![1_u8; RECV_BUFFER_SIZE];
+                    let mut buf = vec![1_u8; INITIAL_RECV_WINDOW_SIZE];
                     let (amount, fin) = server.conn.stream_recv(stream_id, &mut buf).unwrap();
                     assert!(fin);
                     assert_eq!(
@@ -2803,7 +2803,7 @@ mod tests {
         assert_eq!(sent, Ok(first_frame.len()));
 
         // The second frame cannot fit.
-        let sent = client.send_data(request_stream_id, &vec![0_u8; SEND_BUFFER_SIZE]);
+        let sent = client.send_data(request_stream_id, &vec![0_u8; INITIAL_RECV_WINDOW_SIZE]);
         assert_eq!(sent, Ok(expected_second_data_frame.len()));
 
         // Close stream.
@@ -2812,7 +2812,7 @@ mod tests {
         let mut out = client.process_output(now());
         // We need to loop a bit until all data has been sent. Once for every 1K
         // of data.
-        for _i in 0..SEND_BUFFER_SIZE / 1000 {
+        for _i in 0..INITIAL_RECV_WINDOW_SIZE / 1000 {
             out = server.conn.process(out.dgram(), now());
             out = client.process(out.dgram(), now());
         }
@@ -2822,7 +2822,7 @@ mod tests {
             if let ConnectionEvent::RecvStreamReadable { stream_id } = e {
                 if stream_id == request_stream_id {
                     // Read DATA frames.
-                    let mut buf = vec![1_u8; RECV_BUFFER_SIZE];
+                    let mut buf = vec![1_u8; INITIAL_RECV_WINDOW_SIZE];
                     let (amount, fin) = server.conn.stream_recv(stream_id, &mut buf).unwrap();
                     assert!(fin);
                     assert_eq!(
@@ -2875,7 +2875,7 @@ mod tests {
     // After the first frame there is exactly 63+2 bytes left in the send buffer.
     #[test]
     fn fetch_two_data_frame_second_63bytes() {
-        let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 88);
+        let (buf, hdr) = alloc_buffer(INITIAL_RECV_WINDOW_SIZE - 88);
         fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x3f], &[0_u8; 63]);
     }
 
@@ -2884,7 +2884,7 @@ mod tests {
     // but we can only send 63 bytes.
     #[test]
     fn fetch_two_data_frame_second_63bytes_place_for_66() {
-        let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 89);
+        let (buf, hdr) = alloc_buffer(INITIAL_RECV_WINDOW_SIZE - 89);
         fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x3f], &[0_u8; 63]);
     }
 
@@ -2893,15 +2893,15 @@ mod tests {
     // but we can only send 64 bytes.
     #[test]
     fn fetch_two_data_frame_second_64bytes_place_for_67() {
-        let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 90);
+        let (buf, hdr) = alloc_buffer(INITIAL_RECV_WINDOW_SIZE - 90);
         fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x40, 0x40], &[0_u8; 64]);
     }
 
     // Send 2 frames. For the second one we can only send 16383 bytes.
     // After the first frame there is exactly 16383+3 bytes left in the send buffer.
     #[test]
     fn fetch_two_data_frame_second_16383bytes() {
-        let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 16409);
+        let (buf, hdr) = alloc_buffer(INITIAL_RECV_WINDOW_SIZE - 16409);
         fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x7f, 0xff], &[0_u8; 16383]);
     }
 
@@ -2910,7 +2910,7 @@ mod tests {
     // send 16383 bytes.
     #[test]
     fn fetch_two_data_frame_second_16383bytes_place_for_16387() {
-        let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 16410);
+        let (buf, hdr) = alloc_buffer(INITIAL_RECV_WINDOW_SIZE - 16410);
         fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x7f, 0xff], &[0_u8; 16383]);
     }
 
@@ -2919,7 +2919,7 @@ mod tests {
     // send 16383 bytes.
     #[test]
     fn fetch_two_data_frame_second_16383bytes_place_for_16388() {
-        let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 16411);
+        let (buf, hdr) = alloc_buffer(INITIAL_RECV_WINDOW_SIZE - 16411);
         fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x7f, 0xff], &[0_u8; 16383]);
     }
 
@@ -2928,7 +2928,7 @@ mod tests {
     // 16384 bytes.
     #[test]
     fn fetch_two_data_frame_second_16384bytes_place_for_16389() {
-        let (buf, hdr) = alloc_buffer(SEND_BUFFER_SIZE - 16412);
+        let (buf, hdr) = alloc_buffer(INITIAL_RECV_WINDOW_SIZE - 16412);
         fetch_with_two_data_frames(&buf, &hdr, &[0x0, 0x80, 0x0, 0x40, 0x0], &[0_u8; 16384]);
     }
 

neqo-transport/Cargo.toml

@@ -67,3 +67,8 @@ required-features = ["bench"]
 name = "sent_packets"
 harness = false
 required-features = ["bench"]
+
+[[bench]]
+name = "min_bandwidth"
+harness = false
+required-features = ["bench"]

neqo-transport/benches/min_bandwidth.rs

@@ -0,0 +1,79 @@
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+//! A simulated transfer benchmark, asserting a minimum bandwidth.
+//!
+//! This is using [`test_fixture::sim`], i.e. does no I/O beyond the process
+//! boundary and runs in simulated time. Given that [`test_fixture::sim`] is
+//! deterministic, there is no need for multiple benchmark iterations. Still it
+//! is a Rust benchmark instead of a unit test due to its runtime (> 10s) even
+//! in Rust release mode.
+
+use std::time::Duration;
+
+use neqo_transport::{ConnectionParameters, State};
+use test_fixture::{
+    boxed,
+    sim::{
+        connection::{ConnectionNode, ReachState, ReceiveData, SendData},
+        network::{Delay, Mtu, TailDrop},
+        Simulator,
+    },
+};
+
+#[allow(clippy::cast_precision_loss)]
+pub fn main() {
+    const MIB: usize = 1_024 * 1_024;
+    const GIB: usize = 1_024 * MIB;
+
+    const MBIT: usize = 1_000 * 1_000;
+    const GBIT: usize = 1_000 * MBIT;
+
+    const TRANSFER_AMOUNT: usize = GIB;
+    const LINK_BANDWIDTH: usize = GBIT;
+    const LINK_RTT_MS: usize = 40;
+    const MINIMUM_EXPECTED_UTILIZATION: f64 = 0.5;
+
+    let gbit_link = || {
+        let rate_byte = LINK_BANDWIDTH / 8;
+        // Router buffer set to bandwidth-delay product.
+        let capacity_byte = rate_byte * LINK_RTT_MS / 1_000;
+        TailDrop::new(rate_byte, capacity_byte, Duration::ZERO)
+    };
+
+    let simulated_time = Simulator::new(
+        "gbit-bandwidth",
+        boxed![
+            ConnectionNode::new_client(
+                ConnectionParameters::default(),
+                boxed![ReachState::new(State::Confirmed)],
+                boxed![ReceiveData::new(TRANSFER_AMOUNT)]
+            ),
+            Mtu::new(1500),
+            gbit_link(),
+            Delay::new(Duration::from_millis(LINK_RTT_MS as u64 / 2)),
+            ConnectionNode::new_server(
+                ConnectionParameters::default(),
+                boxed![ReachState::new(State::Confirmed)],
+                boxed![SendData::new(TRANSFER_AMOUNT)]
+            ),
+            Mtu::new(1500),
+            gbit_link(),
+            Delay::new(Duration::from_millis(LINK_RTT_MS as u64 / 2)),
+        ],
+    )
+    .setup()
+    .run();
+
+    let achieved_bandwidth = TRANSFER_AMOUNT as f64 * 8.0 / simulated_time.as_secs_f64();
+
+    assert!(
+        LINK_BANDWIDTH as f64 * MINIMUM_EXPECTED_UTILIZATION < achieved_bandwidth,
+        "expected to reach {MINIMUM_EXPECTED_UTILIZATION} of maximum bandwidth ({} Mbit/s) but got {} Mbit/s",
+        LINK_BANDWIDTH  / MBIT,
+        achieved_bandwidth / MBIT as f64,
+    );
+}

neqo-transport/benches/transfer.rs

@@ -12,7 +12,7 @@ use test_fixture::{
     boxed,
     sim::{
         connection::{ConnectionNode, ReachState, ReceiveData, SendData},
-        network::{Delay, TailDrop},
+        network::{RandomDelay, TailDrop},
         ReadySimulator, Simulator,
     },
 };
@@ -38,14 +38,14 @@ fn benchmark_transfer(c: &mut Criterion, label: &str, seed: Option<impl AsRef<st
                             boxed![SendData::new(TRANSFER_AMOUNT)]
                         ),
                         TailDrop::dsl_uplink(),
-                        Delay::new(ZERO..JITTER),
+                        RandomDelay::new(ZERO..JITTER),
                         ConnectionNode::new_server(
                             ConnectionParameters::default().pmtud(true).pacing(pacing),
                             boxed![ReachState::new(State::Confirmed)],
                             boxed![ReceiveData::new(TRANSFER_AMOUNT)]
                         ),
                         TailDrop::dsl_downlink(),
-                        Delay::new(ZERO..JITTER),
+                        RandomDelay::new(ZERO..JITTER),
                     ];
                     let mut sim = Simulator::new(label, nodes);
                     if let Some(seed) = &seed {

neqo-transport/src/connection/mod.rs

@@ -2109,7 +2109,13 @@ impl Connection {
         &mut self,
         builder: &mut PacketBuilder,
         tokens: &mut Vec<RecoveryToken>,
+        now: Instant,
     ) {
+        let rtt = self.paths.primary().map_or_else(
+            || RttEstimate::default().estimate(),
+            |p| p.borrow().rtt().estimate(),
+        );
+
         let stats = &mut self.stats.borrow_mut();
         let frame_stats = &mut stats.frame_tx;
         if self.role == Role::Server {
@@ -2124,7 +2130,7 @@ impl Connection {
             TransmissionPriority::Important,
         ] {
             self.streams
-                .write_frames(prio, builder, tokens, frame_stats);
+                .write_frames(prio, builder, tokens, frame_stats, now, rtt);
             if builder.is_full() {
                 return;
             }
@@ -2143,7 +2149,7 @@ impl Connection {
 
         for prio in [TransmissionPriority::High, TransmissionPriority::Normal] {
             self.streams
-                .write_frames(prio, builder, tokens, &mut stats.frame_tx);
+                .write_frames(prio, builder, tokens, &mut stats.frame_tx, now, rtt);
             if builder.is_full() {
                 return;
             }
@@ -2173,8 +2179,14 @@ impl Connection {
             return;
         }
 
-        self.streams
-            .write_frames(TransmissionPriority::Low, builder, tokens, frame_stats);
+        self.streams.write_frames(
+            TransmissionPriority::Low,
+            builder,
+            tokens,
+            frame_stats,
+            now,
+            rtt,
+        );
 
         #[cfg(test)]
         if let Some(w) = &mut self.test_frame_writer {
@@ -2291,7 +2303,7 @@ impl Connection {
                         .send_probe(builder, &mut self.stats.borrow_mut());
                     ack_eliciting = true;
                 }
-                self.write_appdata_frames(builder, &mut tokens);
+                self.write_appdata_frames(builder, &mut tokens, now);
             } else {
                 let stats = &mut self.stats.borrow_mut().frame_tx;
                 self.crypto.write_frame(space, builder, &mut tokens, stats);

neqo-transport/src/connection/params.rs

@@ -9,7 +9,7 @@ use std::{cmp::max, time::Duration};
 pub use crate::recovery::FAST_PTO_SCALE;
 use crate::{
     connection::{ConnectionIdManager, Role, LOCAL_ACTIVE_CID_LIMIT},
-    recv_stream::RECV_BUFFER_SIZE,
+    recv_stream::INITIAL_RECV_WINDOW_SIZE,
     rtt::GRANULARITY,
     stream_id::StreamType,
     tparams::{self, PreferredAddress, TransportParameter, TransportParametersHandler},
@@ -25,7 +25,7 @@ const LOCAL_STREAM_LIMIT_UNI: u64 = 16;
 pub const ACK_RATIO_SCALE: u8 = 10;
 /// By default, aim to have the peer acknowledge 4 times per round trip time.
 /// See `ConnectionParameters.ack_ratio` for more.
-const DEFAULT_ACK_RATIO: u8 = 4 * ACK_RATIO_SCALE;
+pub const DEFAULT_ACK_RATIO: u8 = 4 * ACK_RATIO_SCALE;
 /// The local value for the idle timeout period.
 const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
 const MAX_QUEUED_DATAGRAMS_DEFAULT: usize = 10;
@@ -91,9 +91,9 @@ impl Default for ConnectionParameters {
             versions: VersionConfig::default(),
             cc_algorithm: CongestionControlAlgorithm::NewReno,
             max_data: LOCAL_MAX_DATA,
-            max_stream_data_bidi_remote: u64::try_from(RECV_BUFFER_SIZE).unwrap(),
-            max_stream_data_bidi_local: u64::try_from(RECV_BUFFER_SIZE).unwrap(),
-            max_stream_data_uni: u64::try_from(RECV_BUFFER_SIZE).unwrap(),
+            max_stream_data_bidi_remote: u64::try_from(INITIAL_RECV_WINDOW_SIZE).unwrap(),
+            max_stream_data_bidi_local: u64::try_from(INITIAL_RECV_WINDOW_SIZE).unwrap(),
+            max_stream_data_uni: u64::try_from(INITIAL_RECV_WINDOW_SIZE).unwrap(),
             max_streams_bidi: LOCAL_STREAM_LIMIT_BIDI,
             max_streams_uni: LOCAL_STREAM_LIMIT_UNI,
             ack_ratio: DEFAULT_ACK_RATIO,

neqo-transport/src/connection/tests/stream.rs

@@ -21,8 +21,8 @@ use crate::{
         FRAME_TYPE_STREAM_CLIENT_BIDI, FRAME_TYPE_STREAM_DATA_BLOCKED,
     },
     packet::PacketBuilder,
-    recv_stream::RECV_BUFFER_SIZE,
-    send_stream::{OrderGroup, SendStreamState, SEND_BUFFER_SIZE},
+    recv_stream::INITIAL_RECV_WINDOW_SIZE,
+    send_stream::{OrderGroup, SendStreamState},
     streams::{SendOrder, StreamOrder},
     tparams::{self, TransportParameter},
     CloseReason, Connection, ConnectionParameters, Error, StreamId, StreamType,
@@ -437,18 +437,7 @@ fn max_data() {
     client.streams.handle_max_data(100_000_000);
     assert_eq!(
         client.stream_avail_send_space(stream_id).unwrap(),
-        SEND_BUFFER_SIZE - SMALL_MAX_DATA
-    );
-
-    // Increase max stream data. Avail space now limited by tx buffer
-    client
-        .streams
-        .get_send_stream_mut(stream_id)
-        .unwrap()
-        .set_max_stream_data(100_000_000);
-    assert_eq!(
-        client.stream_avail_send_space(stream_id).unwrap(),
-        SEND_BUFFER_SIZE - SMALL_MAX_DATA + 4096
+        INITIAL_RECV_WINDOW_SIZE - SMALL_MAX_DATA
     );
 
     let evts = client.events().collect::<Vec<_>>();
@@ -887,7 +876,7 @@ fn stream_data_blocked_generates_max_stream_data() {
         }
         written += amount;
     }
-    assert_eq!(written, RECV_BUFFER_SIZE);
+    assert_eq!(written, INITIAL_RECV_WINDOW_SIZE);
 }
 
 /// See <https://github.com/mozilla/neqo/issues/871>