streamer: refactor config structs

bench-vote/src/main.rs

@@ -267,12 +267,14 @@ fn main() -> Result<()> {
                 max_connections_per_ipaddr_per_min: max_connections_per_ipaddr_per_min
                     .try_into()
                     .unwrap(),
+                ..Default::default()
+            };
+            let qos_config = SwQosConfig {
                 max_connections_per_unstaked_peer: max_connections_per_peer,
                 max_staked_connections: max_connections,
                 max_unstaked_connections: 0,
                 ..Default::default()
             };
-            let qos_config = SwQosConfig::default();
             let (s_reader, r_reader) = unbounded();
             read_channels.push(r_reader);
 

core/src/validator.rs

@@ -589,17 +589,18 @@ impl ValidatorTpuConfig {
         let tpu_fwd_quic_server_config = SwQosQuicStreamerConfig {
             quic_streamer_config: QuicStreamerConfig {
                 max_connections_per_ipaddr_per_min: 32,
+                ..Default::default()
+            },
+            qos_config: SwQosConfig {
                 max_unstaked_connections: 0,
                 ..Default::default()
             },
-            qos_config: SwQosConfig::default(),
         };
 
         // vote and tpu_fwd share the same characteristics -- disallow non-staked connections:
         let vote_quic_server_config = SimpleQosQuicStreamerConfig {
             quic_streamer_config: QuicStreamerConfig {
                 max_connections_per_ipaddr_per_min: 32,
-                max_unstaked_connections: 0,
                 ..Default::default()
             },
             qos_config: SimpleQosConfig::default(),

streamer/examples/swqos.rs

@@ -68,8 +68,10 @@ pub fn load_staked_nodes_overrides(path: &String) -> anyhow::Result<HashMap<Pubk
 
 #[derive(Debug, Parser)]
 struct Cli {
-    #[arg(short, long, default_value_t = 1)]
-    max_connections_per_peer: usize,
+    #[arg(short, long, default_value_t = 10)]
+    max_connections_per_staked_peer: usize,
+    #[arg(short, long, default_value_t = 10)]
+    max_connections_per_unstaked_peer: usize,
 
     #[arg(short, long, default_value = "0.0.0.0:8008")]
     bind_to: SocketAddr,
@@ -120,10 +122,13 @@ async fn main() -> anyhow::Result<()> {
         sender,
         staked_nodes,
         QuicStreamerConfig {
-            max_connections_per_unstaked_peer: cli.max_connections_per_peer,
             ..QuicStreamerConfig::default()
         },
-        SwQosConfig::default(),
+        SwQosConfig {
+            max_connections_per_staked_peer: cli.max_connections_per_staked_peer,
+            max_connections_per_unstaked_peer: cli.max_connections_per_unstaked_peer,
+            ..Default::default()
+        },
         cancel.clone(),
     )?;
     info!("Server listening on {}", socket.local_addr()?);

streamer/src/nonblocking/qos.rs

@@ -51,4 +51,9 @@ pub(crate) trait QosController<C: ConnectionContext> {
         context: &C,
         connection: Connection,
     ) -> impl Future<Output = usize> + Send;
+
+    /// How many concurrent
+    fn max_concurrent_connections(&self) -> usize;
 }
+
+pub trait QosConfig {}

streamer/src/nonblocking/quic.rs

@@ -157,7 +157,7 @@ where
         })
         .collect::<Result<Vec<_>, _>>()?;
 
-    let max_concurrent_connections = quic_server_params.max_concurrent_connections();
+    let max_concurrent_connections = qos.max_concurrent_connections();
     let handle = tokio::spawn({
         let endpoints = endpoints.clone();
         let stats = stats.clone();
@@ -334,10 +334,9 @@ where
                 continue;
             }
 
-            let Ok(client_connection_tracker) = ClientConnectionTracker::new(
-                stats.clone(),
-                quic_server_params.max_concurrent_connections(),
-            ) else {
+            let Ok(client_connection_tracker) =
+                ClientConnectionTracker::new(stats.clone(), qos.max_concurrent_connections())
+            else {
                 stats
                     .refused_connections_too_many_open_connections
                     .fetch_add(1, Ordering::Relaxed);
@@ -1351,7 +1350,7 @@ pub mod test {
         } = setup_quic_server(
             None,
             QuicStreamerConfig::default_for_tests(),
-            SwQosConfig::default(),
+            SwQosConfig::default_for_tests(),
         );
         check_block_multiple_connections(server_address).await;
         cancel.cancel();
@@ -1372,10 +1371,12 @@ pub mod test {
         } = setup_quic_server(
             None,
             QuicStreamerConfig {
-                max_connections_per_unstaked_peer: 2,
                 ..QuicStreamerConfig::default_for_tests()
             },
-            SwQosConfig::default(),
+            SwQosConfig {
+                max_connections_per_unstaked_peer: 2,
+                ..SwQosConfig::default_for_tests()
+            },
         );
 
         let client_socket = bind_to_localhost_unique().expect("should bind - client");
@@ -1582,10 +1583,12 @@ pub mod test {
             sender,
             staked_nodes,
             QuicStreamerConfig {
-                max_unstaked_connections: 0, // Do not allow any connection from unstaked clients/nodes
                 ..QuicStreamerConfig::default_for_tests()
             },
-            SwQosConfig::default(),
+            SwQosConfig {
+                max_unstaked_connections: 0, // Do not allow any connection from unstaked clients/nodes
+                ..Default::default()
+            },
             cancel.clone(),
         )
         .unwrap();
@@ -1616,10 +1619,12 @@ pub mod test {
             sender,
             staked_nodes,
             QuicStreamerConfig {
-                max_connections_per_unstaked_peer: 2,
                 ..QuicStreamerConfig::default_for_tests()
             },
-            SwQosConfig::default(),
+            SwQosConfig {
+                max_connections_per_unstaked_peer: 2,
+                ..Default::default()
+            },
             cancel.clone(),
         )
         .unwrap();