diff --git a/crates/lanspread-peer/src/services/server.rs b/crates/lanspread-peer/src/services/server.rs index 89d6d77..39cde7a 100644 --- a/crates/lanspread-peer/src/services/server.rs +++ b/crates/lanspread-peer/src/services/server.rs @@ -35,10 +35,13 @@ const MAX_ESTABLISHED_CONNECTIONS: usize = 64; /// Mirrors the transport stream limit and bounds application stream futures. const MAX_CONTROL_STREAM_TASKS: usize = 32; /// Server-wide cap acquired before any length-delimited decoder is allocated. -const MAX_GLOBAL_CONTROL_STREAM_TASKS: usize = 64; +const MAX_GLOBAL_CONTROL_STREAM_TASKS: usize = 16; /// Long-lived transfers move from the decoder pool to this smaller pool so /// saturated bulk egress cannot consume every control-plane permit. const MAX_GLOBAL_BULK_TRANSFER_TASKS: usize = 48; +/// Native archive extraction is substantially more expensive than ordinary +/// catalog egress, so keep it behind a separate small global budget. +const MAX_GLOBAL_STREAM_INSTALL_TASKS: usize = 2; /// Application idle bound for an established connection with no active /// request streams. This is independent of transport keepalive traffic. const CONNECTION_NO_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(10); @@ -125,6 +128,7 @@ async fn run_server_body( let mut connection_tasks = JoinSet::new(); let control_stream_permits = Arc::new(Semaphore::new(MAX_GLOBAL_CONTROL_STREAM_TASKS)); let bulk_transfer_permits = Arc::new(Semaphore::new(MAX_GLOBAL_BULK_TRANSFER_TASKS)); + let stream_install_permits = Arc::new(Semaphore::new(MAX_GLOBAL_STREAM_INSTALL_TASKS)); let ready_addr = (*ctx.local_peer_addr.read().await).unwrap_or_else(|| direct_connect_addr(server_addr)); @@ -174,6 +178,7 @@ async fn run_server_body( server_children_shutdown.clone(), Arc::clone(&control_stream_permits), Arc::clone(&bulk_transfer_permits), + Arc::clone(&stream_install_permits), )); } } @@ -265,6 +270,7 @@ async fn handle_peer_connection( server_shutdown: CancellationToken, control_stream_permits: Arc, bulk_transfer_permits: Arc, + stream_install_permits: Arc, ) -> eyre::Result<()> { let remote_addr = connection.remote_addr()?; log::info!("{remote_addr} peer connected"); @@ -313,6 +319,7 @@ async fn handle_peer_connection( stream_shutdown, control_permit, Arc::clone(&bulk_transfer_permits), + Arc::clone(&stream_install_permits), )); } Ok(None) => break Ok(()), @@ -353,6 +360,7 @@ async fn handle_admitted_peer_stream( stream_shutdown: CancellationToken, control_permit: OwnedSemaphorePermit, bulk_transfer_permits: Arc, + stream_install_permits: Arc, ) -> eyre::Result<()> { handle_peer_stream( stream, @@ -361,6 +369,7 @@ async fn handle_admitted_peer_stream( stream_shutdown, control_permit, bulk_transfer_permits, + stream_install_permits, ) .await } @@ -410,6 +419,7 @@ mod tests { MAX_ESTABLISHED_CONNECTIONS, MAX_GLOBAL_BULK_TRANSFER_TASKS, MAX_GLOBAL_CONTROL_STREAM_TASKS, + MAX_GLOBAL_STREAM_INSTALL_TASKS, MAX_INFLIGHT_HANDSHAKES, bounded_endpoint_limits, drain_joined_child_tasks, @@ -423,8 +433,9 @@ mod tests { assert_eq!(MAX_INFLIGHT_HANDSHAKES, 64); assert_eq!(MAX_ESTABLISHED_CONNECTIONS, 64); assert_eq!(MAX_CONTROL_STREAM_TASKS, 32); - assert_eq!(MAX_GLOBAL_CONTROL_STREAM_TASKS, 64); + assert_eq!(MAX_GLOBAL_CONTROL_STREAM_TASKS, 16); assert_eq!(MAX_GLOBAL_BULK_TRANSFER_TASKS, 48); + assert_eq!(MAX_GLOBAL_STREAM_INSTALL_TASKS, 2); assert_eq!( u64::try_from(MAX_CONTROL_STREAM_TASKS).expect("stream task bound fits u64"), crate::quic_runtime::MAX_OPEN_BIDIRECTIONAL_STREAMS, diff --git a/crates/lanspread-peer/src/services/stream.rs b/crates/lanspread-peer/src/services/stream.rs index de6d43c..c495e74 100644 --- a/crates/lanspread-peer/src/services/stream.rs +++ b/crates/lanspread-peer/src/services/stream.rs @@ -61,6 +61,7 @@ pub(super) async fn handle_peer_stream( stream_shutdown: CancellationToken, control_permit: OwnedSemaphorePermit, bulk_transfer_permits: Arc, + stream_install_permits: Arc, ) -> eyre::Result<()> { let (rx, tx) = stream.split(); let mut framed_rx = FramedRead::new(rx, request_codec()); @@ -94,6 +95,7 @@ pub(super) async fn handle_peer_stream( source_ip, framed_tx, &stream_shutdown, + &stream_install_permits, ) .await; framed_tx = dispatched.writer; @@ -111,6 +113,7 @@ pub(super) async fn handle_peer_stream( source_ip, framed_tx, &stream_shutdown, + &stream_install_permits, ) .await; framed_tx = dispatched.writer; @@ -229,6 +232,7 @@ async fn dispatch_request( source_ip: Option, framed_tx: ResponseWriter, stream_shutdown: &CancellationToken, + stream_install_permits: &Arc, ) -> DispatchResult { match request { Request::Ping => { @@ -322,10 +326,25 @@ async fn dispatch_request( Request::StreamInstall { game_id, content_id, - } => DispatchResult::close( - handle_stream_install_request(ctx, game_id, content_id, framed_tx, stream_shutdown) + } => { + let Ok(stream_install_permit) = Arc::clone(stream_install_permits).try_acquire_owned() + else { + let mut tx = framed_tx.into_inner(); + let _ = tx.reset(application::Error::UNKNOWN); + return DispatchResult::reset(FramedWrite::new(tx, control_codec())); + }; + DispatchResult::close( + handle_stream_install_request( + ctx, + game_id, + content_id, + framed_tx, + stream_shutdown, + stream_install_permit, + ) .await, - ), + ) + } } } diff --git a/crates/lanspread-peer/src/services/transfer.rs b/crates/lanspread-peer/src/services/transfer.rs index 6826ab0..0000080 100644 --- a/crates/lanspread-peer/src/services/transfer.rs +++ b/crates/lanspread-peer/src/services/transfer.rs @@ -438,6 +438,7 @@ pub(super) async fn handle_stream_install_request( content_id: ContentId, framed_tx: ResponseWriter, stream_shutdown: &CancellationToken, + _stream_install_permit: tokio::sync::OwnedSemaphorePermit, ) -> ResponseWriter { log::info!("Received StreamInstall request for {game_id} from peer"); let mut tx = framed_tx.into_inner();