test disable stream timeout

core/src/tpu.rs

@@ -50,7 +50,7 @@ use {
 };
 
 // allow multiple connections for NAT and any open/close overlap
-pub const MAX_QUIC_CONNECTIONS_PER_PEER: usize = 8;
+pub const MAX_QUIC_CONNECTIONS_PER_PEER: usize = 8000;
 
 pub struct TpuSockets {
     pub transactions: Vec<UdpSocket>,

net-utils/src/lib.rs

@@ -28,7 +28,7 @@ pub struct UdpSocketPair {
 
 pub type PortRange = (u16, u16);
 
-pub const VALIDATOR_PORT_RANGE: PortRange = (8000, 10_000);
+pub const VALIDATOR_PORT_RANGE: PortRange = (8000, 20_000);
 pub const MINIMUM_VALIDATOR_PORT_RANGE_WIDTH: u16 = 14; // VALIDATOR_PORT_RANGE must be at least this wide
 
 pub(crate) const HEADER_LENGTH: usize = 4;

quic-client/Cargo.toml

@@ -32,5 +32,6 @@ tokio = { workspace = true, features = ["full"] }
 
 [dev-dependencies]
 crossbeam-channel = { workspace = true }
+rayon = { workspace = true }
 solana-logger = { workspace = true }
 solana-perf = { workspace = true }

quic-client/src/nonblocking/quic_client.rs

@@ -1,6 +1,7 @@
 //! Simple nonblocking client that connects to a given UDP port with the QUIC protocol
 //! and provides an interface for sending data which is restricted by the
 //! server's flow control.
+
 use {
     async_mutex::Mutex,
     async_trait::async_trait,
@@ -9,7 +10,7 @@ use {
     log::*,
     quinn::{
         ClientConfig, ConnectError, Connection, ConnectionError, Endpoint, EndpointConfig,
-        IdleTimeout, TokioRuntime, TransportConfig, WriteError,
+        IdleTimeout, SendStream, TokioRuntime, TransportConfig, WriteError,
     },
     solana_connection_cache::{
         client_connection::ClientStats, connection_cache_stats::ConnectionCacheStats,
@@ -252,6 +253,7 @@ pub struct QuicClient {
     addr: SocketAddr,
     stats: Arc<ClientStats>,
     chunk_size: usize,
+    streams: Mutex<Vec<SendStream>>,
 }
 
 impl QuicClient {
@@ -266,17 +268,26 @@ impl QuicClient {
             addr,
             stats: Arc::new(ClientStats::default()),
             chunk_size,
+            streams: Vec::default().into(),
         }
     }
 
     async fn _send_buffer_using_conn(
+        &self,
         data: &[u8],
         connection: &Connection,
     ) -> Result<(), QuicError> {
+        println!("Opening stream...");
         let mut send_stream = connection.open_uni().await?;
-
-        send_stream.write_all(data).await?;
-        send_stream.finish().await?;
+        println!("Writing data...");
+
+        send_stream.write(data).await?;
+        // intentionally holding off finalizing the stream
+        // send_stream.finish().await?;
+        println!("Taking a lock");
+        let mut lock = self.streams.lock().await;
+        lock.push(send_stream);
+        println!("Stored send_stream");
         Ok(())
     }
 
@@ -339,12 +350,13 @@ impl QuicClient {
                         match conn {
                             Ok(conn) => {
                                 *conn_guard = Some(conn.clone());
-                                info!(
-                                    "Made connection to {} id {} try_count {}, from connection cache warming?: {}",
+                                println!(
+                                    "Made connection to {} id {} try_count {}, from connection cache warming?: {} {:?}",
                                     self.addr,
                                     conn.connection.stable_id(),
                                     connection_try_count,
                                     data.is_empty(),
+                                    std::thread::current().id()
                                 );
                                 connection_try_count += 1;
                                 conn.connection.clone()
@@ -395,7 +407,11 @@ impl QuicClient {
             last_connection_id = connection.stable_id();
             measure_prepare_connection.stop();
 
-            match Self::_send_buffer_using_conn(data, &connection).await {
+            println!(
+                "To call _send_buffer_using_conn... {:?}",
+                std::thread::current().id()
+            );
+            match self._send_buffer_using_conn(data, &connection).await {
                 Ok(()) => {
                     measure_send_packet.stop();
                     stats.successful_packets.fetch_add(1, Ordering::Relaxed);
@@ -500,7 +516,7 @@ impl QuicClient {
                 join_all(
                     buffs
                         .into_iter()
-                        .map(|buf| Self::_send_buffer_using_conn(buf.as_ref(), connection_ref)),
+                        .map(|buf| self._send_buffer_using_conn(buf.as_ref(), connection_ref)),
                 )
             })
             .collect();
@@ -568,12 +584,15 @@ impl ClientConnection for QuicClientConnection {
     }
 
     async fn send_data_batch(&self, buffers: &[Vec<u8>]) -> TransportResult<()> {
+        println!("Executing send_data_batch");
         let stats = ClientStats::default();
         let len = buffers.len();
         let res = self
             .client
             .send_batch(buffers, &stats, self.connection_stats.clone())
             .await;
+
+        println!("Result send_data_batch {res:?}");
         self.connection_stats
             .add_client_stats(&stats, len, res.is_ok());
         res?;

quic-client/src/quic_client.rs

@@ -69,6 +69,7 @@ lazy_static! {
     static ref ASYNC_TASK_SEMAPHORE: AsyncTaskSemaphore =
         AsyncTaskSemaphore::new(MAX_OUTSTANDING_TASK);
     static ref RUNTIME: Runtime = tokio::runtime::Builder::new_multi_thread()
+        //.worker_threads(1)
         .thread_name("quic-client")
         .enable_all()
         .build()

quic-client/tests/quic_client.rs

@@ -3,12 +3,16 @@ mod tests {
     use {
         crossbeam_channel::{unbounded, Receiver},
         log::*,
+        rayon::iter::{IntoParallelIterator, ParallelIterator},
         solana_connection_cache::connection_cache_stats::ConnectionCacheStats,
         solana_perf::packet::PacketBatch,
         solana_quic_client::nonblocking::quic_client::{
             QuicClientCertificate, QuicLazyInitializedEndpoint,
         },
-        solana_sdk::{net::DEFAULT_TPU_COALESCE, packet::PACKET_DATA_SIZE, signature::Keypair},
+        solana_sdk::{
+            net::DEFAULT_TPU_COALESCE, packet::PACKET_DATA_SIZE,
+            quic::QUIC_MAX_STAKED_CONCURRENT_STREAMS, signature::Keypair,
+        },
         solana_streamer::{
             nonblocking::quic::DEFAULT_WAIT_FOR_CHUNK_TIMEOUT, quic::SpawnServerResult,
             streamer::StakedNodes, tls_certificates::new_self_signed_tls_certificate,
@@ -319,4 +323,114 @@ mod tests {
         response_recv_thread.join().unwrap();
         info!("Response receiver exited!");
     }
+
+    #[tokio::test]
+    async fn test_connection_cache_memory_usage() {
+        use {
+            solana_connection_cache::client_connection::ClientConnection,
+            solana_quic_client::quic_client::QuicClientConnection,
+        };
+        solana_logger::setup();
+
+        let addr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
+        let port = 8009;
+        let tpu_addr = SocketAddr::new(addr, port);
+        let connection_cache_stats = Arc::new(ConnectionCacheStats::default());
+        let mut clients = Vec::default();
+
+        let thread_pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(512)
+            .thread_name(|i| format!("concame{i:02}"))
+            .build()
+            .unwrap();
+
+        for i in 0..8000 {
+            println!("Connection {i}");
+            let client = QuicClientConnection::new(
+                Arc::new(QuicLazyInitializedEndpoint::default()),
+                tpu_addr,
+                connection_cache_stats.clone(),
+            );
+
+            // Send a full size packet with single byte writes.
+            let num_expected_packets: usize = 1;
+
+            thread_pool.install(|| {
+                (0..QUIC_MAX_STAKED_CONCURRENT_STREAMS)
+                    .into_par_iter()
+                    .for_each(|i| {
+                        let packets = vec![vec![0u8; PACKET_DATA_SIZE]; num_expected_packets];
+                        let rslt = client.send_data_batch(&packets);
+                        if let Err(rslt) = rslt {
+                            info!("Connection {i} error {rslt:?}");
+                        }
+                    });
+            });
+
+            clients.push(client);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_connection_cache_memory_usage_2() {
+        use {
+            solana_connection_cache::client_connection::ClientConnection,
+            solana_quic_client::quic_client::QuicClientConnection,
+        };
+
+        solana_logger::setup();
+
+        // Request Sender, it uses the same endpoint as the response receiver:
+        let addr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
+        let port = 8009;
+        let tpu_addr = SocketAddr::new(addr, port);
+        let connection_cache_stats = Arc::new(ConnectionCacheStats::default());
+
+        let (cert, priv_key) =
+            new_self_signed_tls_certificate(&Keypair::new(), IpAddr::V4(Ipv4Addr::UNSPECIFIED))
+                .expect("Failed to initialize QUIC client certificates");
+        let client_certificate = Arc::new(QuicClientCertificate {
+            certificate: cert,
+            key: priv_key,
+        });
+
+        let endpoint = Arc::new(QuicLazyInitializedEndpoint::new(client_certificate, None));
+
+        let thread_pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(128)
+            .thread_name(|i| format!("concame{i:02}"))
+            .build()
+            .unwrap();
+
+        let mut clients = Vec::default();
+        for i in 0..8000 {
+            println!("Connection {i}");
+            let client = QuicClientConnection::new(
+                endpoint.clone(),
+                tpu_addr,
+                connection_cache_stats.clone(),
+            );
+            // Send a full size packet with single byte writes.
+            let num_expected_packets: usize = 1;
+
+            thread_pool.install(|| {
+                (0..QUIC_MAX_STAKED_CONCURRENT_STREAMS)
+                    .into_par_iter()
+                    .for_each(|_i| {
+                        let packets = vec![
+                            vec![0u8; PACKET_DATA_SIZE-1 /*PACKET_DATA_SIZE*/];
+                            num_expected_packets
+                        ];
+                        println!("Sending data in thread {:?}", std::thread::current().id());
+                        let rslt = client.send_data_batch(&packets);
+                        println!("Sent data: {rslt:?}");
+                        if let Err(rslt) = rslt {
+                            info!("Connection {i} error {_i} {rslt:?}");
+                        }
+                    });
+            });
+
+            clients.push(client);
+        }
+    }
 }

sdk/src/quic.rs

@@ -6,18 +6,18 @@ pub const QUIC_PORT_OFFSET: u16 = 6;
 // Empirically found max number of concurrent streams
 // that seems to maximize TPS on GCE (higher values don't seem to
 // give significant improvement or seem to impact stability)
-pub const QUIC_MAX_UNSTAKED_CONCURRENT_STREAMS: usize = 128;
-pub const QUIC_MIN_STAKED_CONCURRENT_STREAMS: usize = 128;
+pub const QUIC_MAX_UNSTAKED_CONCURRENT_STREAMS: usize = 512;
+pub const QUIC_MIN_STAKED_CONCURRENT_STREAMS: usize = 512;
 
-pub const QUIC_TOTAL_STAKED_CONCURRENT_STREAMS: usize = 100_000;
+pub const QUIC_TOTAL_STAKED_CONCURRENT_STREAMS: usize = 1_024_000;
 
 // Set the maximum concurrent stream numbers to avoid excessive streams.
 // The value was lowered from 2048 to reduce contention of the limited
 // receive_window among the streams which is observed in CI bench-tests with
 // forwarded packets from staked nodes.
 pub const QUIC_MAX_STAKED_CONCURRENT_STREAMS: usize = 512;
 
-pub const QUIC_MAX_TIMEOUT: Duration = Duration::from_secs(2);
+pub const QUIC_MAX_TIMEOUT: Duration = Duration::from_secs(3600);
 pub const QUIC_KEEP_ALIVE: Duration = Duration::from_secs(1);
 
 // Based on commonly-used handshake timeouts for various TCP