streamer: scale max_concurrent_uni_streams with BDP

Author: alexpyattaev

streamer/src/nonblocking/swqos.rs

@@ -49,6 +49,12 @@ pub const QUIC_MAX_STAKED_CONCURRENT_STREAMS: usize = 512;
 
 pub const QUIC_TOTAL_STAKED_CONCURRENT_STREAMS: usize = 100_000;
 
+/// RTT after which we start BDP scaling
+const REFERENCE_RTT_MS: u64 = 50;
+
+/// Above this RTT we stop scaling for BDP
+const MAX_RTT_MS: u64 = 350;
+
 #[derive(Clone)]
 pub struct SwQosConfig {
     pub max_streams_per_ms: u64,
@@ -140,8 +146,12 @@ impl SwQos {
     }
 }
 
-fn compute_max_allowed_uni_streams(peer_type: ConnectionPeerType, total_stake: u64) -> usize {
-    match peer_type {
+fn compute_max_allowed_uni_streams(
+    rtt_millis: u64,
+    peer_type: ConnectionPeerType,
+    total_stake: u64,
+) -> usize {
+    let streams = match peer_type {
         ConnectionPeerType::Staked(peer_stake) => {
             // No checked math for f64 type. So let's explicitly check for 0 here
             if total_stake == 0 || peer_stake > total_stake {
@@ -164,7 +174,9 @@ fn compute_max_allowed_uni_streams(peer_type: ConnectionPeerType, total_stake: u
             }
         }
         ConnectionPeerType::Unstaked => QUIC_MAX_UNSTAKED_CONCURRENT_STREAMS,
-    }
+    };
+    let streams = streams * rtt_millis.clamp(REFERENCE_RTT_MS, MAX_RTT_MS) / REFERENCE_RTT_MS;
+    streams as usize
 }
 
 impl SwQos {
@@ -182,7 +194,10 @@ impl SwQos {
         ),
         ConnectionHandlerError,
     > {
+        // get current RTT and limit it to MAX_RTT_MS
+        let rtt_millis = connection.rtt().as_millis() as u64;
         if let Ok(max_uni_streams) = VarInt::from_u64(compute_max_allowed_uni_streams(
+            rtt_millis,
             conn_context.peer_type(),
             conn_context.total_stake,
         ) as u64)
@@ -528,26 +543,34 @@ pub mod test {
     #[test]
     fn test_max_allowed_uni_streams() {
         assert_eq!(
-            compute_max_allowed_uni_streams(ConnectionPeerType::Unstaked, 0),
+            compute_max_allowed_uni_streams(REFERENCE_RTT_MS, ConnectionPeerType::Unstaked, 0),
             QUIC_MAX_UNSTAKED_CONCURRENT_STREAMS
         );
         assert_eq!(
-            compute_max_allowed_uni_streams(ConnectionPeerType::Staked(10), 0),
+            compute_max_allowed_uni_streams(REFERENCE_RTT_MS, ConnectionPeerType::Staked(10), 0),
             QUIC_MIN_STAKED_CONCURRENT_STREAMS
         );
         let delta =
             (QUIC_TOTAL_STAKED_CONCURRENT_STREAMS - QUIC_MIN_STAKED_CONCURRENT_STREAMS) as f64;
         assert_eq!(
-            compute_max_allowed_uni_streams(ConnectionPeerType::Staked(1000), 10000),
+            compute_max_allowed_uni_streams(
+                REFERENCE_RTT_MS,
+                ConnectionPeerType::Staked(1000),
+                10000
+            ),
             QUIC_MAX_STAKED_CONCURRENT_STREAMS,
         );
         assert_eq!(
-            compute_max_allowed_uni_streams(ConnectionPeerType::Staked(100), 10000),
+            compute_max_allowed_uni_streams(
+                REFERENCE_RTT_MS,
+                ConnectionPeerType::Staked(100),
+                10000
+            ),
             ((delta / (100_f64)) as usize + QUIC_MIN_STAKED_CONCURRENT_STREAMS)
                 .min(QUIC_MAX_STAKED_CONCURRENT_STREAMS)
         );
         assert_eq!(
-            compute_max_allowed_uni_streams(ConnectionPeerType::Unstaked, 10000),
+            compute_max_allowed_uni_streams(REFERENCE_RTT_MS, ConnectionPeerType::Unstaked, 10000),
             QUIC_MAX_UNSTAKED_CONCURRENT_STREAMS
         );
     }