From 1ac93f654dfdfec431459075226af78ad138329d Mon Sep 17 00:00:00 2001 From: Mygod Date: Thu, 22 Jan 2026 22:57:37 -0800 Subject: [PATCH 1/9] flow_control: reserve window for single stream, cap multi-stream stalls --- crates/slipstream-client/src/streams.rs | 257 ++++++++-- crates/slipstream-ffi/src/picoquic.rs | 5 + crates/slipstream-ffi/src/runtime.rs | 5 +- crates/slipstream-server/src/streams.rs | 333 +++++++++++-- .../tests/flow_control_e2e.rs | 464 ++++++++++++++++++ docs/config.md | 6 + docs/design.md | 25 +- docs/protocol.md | 13 +- 8 files changed, 1003 insertions(+), 105 deletions(-) create mode 100644 crates/slipstream-server/tests/flow_control_e2e.rs diff --git a/crates/slipstream-client/src/streams.rs b/crates/slipstream-client/src/streams.rs index a4654ca3..4134895f 100644 --- a/crates/slipstream-client/src/streams.rs +++ b/crates/slipstream-client/src/streams.rs @@ -3,11 +3,11 @@ use slipstream_ffi::picoquic::{ picoquic_add_to_stream, picoquic_call_back_event_t, picoquic_cnx_t, picoquic_current_time, picoquic_get_close_reasons, picoquic_get_cnx_state, picoquic_get_next_local_stream_id, picoquic_mark_active_stream, picoquic_provide_stream_data_buffer, picoquic_reset_stream, - picoquic_stream_data_consumed, + picoquic_stop_sending, picoquic_stream_data_consumed, }; use slipstream_ffi::{SLIPSTREAM_FILE_CANCEL_ERROR, SLIPSTREAM_INTERNAL_ERROR}; use std::collections::HashMap; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener as TokioTcpListener, TcpStream as TokioTcpStream}; use tokio::sync::{mpsc, oneshot, Notify}; @@ -16,11 +16,35 @@ use tracing::{debug, info, warn}; const STREAM_READ_CHUNK_BYTES: usize = 4096; const DEFAULT_TCP_RCVBUF_BYTES: usize = 256 * 1024; const CLIENT_WRITE_COALESCE_DEFAULT_BYTES: usize = 256 * 1024; +const DEFAULT_STREAM_QUEUE_MAX_BYTES: usize = 2 * 1024 * 1024; +const DEFAULT_CONN_RESERVE_BYTES: usize = 64 * 1024; + +fn stream_queue_max_bytes() -> usize { + static MAX_BYTES: OnceLock = OnceLock::new(); + *MAX_BYTES.get_or_init(|| { + std::env::var("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES") + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_STREAM_QUEUE_MAX_BYTES) + }) +} + +fn conn_reserve_bytes() -> usize { + static RESERVE_BYTES: OnceLock = OnceLock::new(); + *RESERVE_BYTES.get_or_init(|| { + std::env::var("SLIPSTREAM_CONN_RESERVE_BYTES") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(DEFAULT_CONN_RESERVE_BYTES) + }) +} pub(crate) struct ClientState { ready: bool, closing: bool, streams: HashMap, + multi_stream_mode: bool, command_tx: mpsc::UnboundedSender, data_notify: Arc, path_events: Vec, @@ -39,6 +63,7 @@ impl ClientState { ready: false, closing: false, streams: HashMap::new(), + multi_stream_mode: false, command_tx, data_notify, path_events: Vec::new(), @@ -81,6 +106,7 @@ impl ClientState { } self.ready = false; self.closing = false; + self.multi_stream_mode = false; self.path_events.clear(); self.debug_enqueued_bytes = 0; self.debug_last_enqueue_at = 0; @@ -97,6 +123,8 @@ struct ClientStream { consumed_offset: u64, fin_offset: Option, fin_enqueued: bool, + discarding: bool, + stop_sending_sent: bool, } enum StreamWrite { @@ -127,6 +155,24 @@ fn close_event_label(event: picoquic_call_back_event_t) -> &'static str { } } +fn promote_streams(cnx: *mut picoquic_cnx_t, state: &mut ClientState) { + for (stream_id, stream) in state.streams.iter_mut() { + if stream.discarding { + continue; + } + if stream.consumed_offset < stream.rx_bytes { + stream.consumed_offset = stream.rx_bytes; + let ret = unsafe { picoquic_stream_data_consumed(cnx, *stream_id, stream.rx_bytes) }; + if ret < 0 { + warn!( + "stream {}: stream_data_consumed failed during promote ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + } + } + } +} + pub(crate) unsafe extern "C" fn client_callback( cnx: *mut picoquic_cnx_t, stream_id: u64, @@ -242,6 +288,12 @@ fn handle_stream_data( let debug_streams = state.debug_streams; let mut reset_stream = false; let mut remove_stream = false; + let multi_stream = state.multi_stream_mode; + let reserve_bytes = if multi_stream { + 0 + } else { + conn_reserve_bytes() + }; { let Some(stream) = state.streams.get_mut(&stream_id) else { @@ -257,10 +309,84 @@ fn handle_stream_data( return; }; - if !data.is_empty() { - // Backpressure is enforced via connection-level max_data, not per-stream buffer caps. - stream.rx_bytes = stream.rx_bytes.saturating_add(data.len() as u64); - if stream + if stream.discarding { + if !data.is_empty() { + let incoming_len = data.len(); + stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); + if stream.consumed_offset < stream.rx_bytes { + stream.consumed_offset = stream.rx_bytes; + let ret = unsafe { + picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) + }; + if ret < 0 { + warn!( + "stream {}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + } + } + } + } else if !data.is_empty() { + let incoming_len = data.len(); + stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); + let projected = stream.queued_bytes.saturating_add(incoming_len); + if multi_stream { + let max_queue = stream_queue_max_bytes(); + if projected > max_queue { + warn!( + "stream {}: queued_bytes {} + {} exceeds limit {}; stopping", + stream_id, stream.queued_bytes, incoming_len, max_queue + ); + stream.discarding = true; + stream.queued_bytes = 0; + let (drain_tx, _drain_rx) = mpsc::unbounded_channel(); + stream.write_tx = drain_tx; + if stream.consumed_offset < stream.rx_bytes { + stream.consumed_offset = stream.rx_bytes; + let ret = unsafe { + picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) + }; + if ret < 0 { + warn!( + "stream {}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + } + } + if !stream.stop_sending_sent { + let _ = unsafe { + picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) + }; + stream.stop_sending_sent = true; + } + } else if stream + .write_tx + .send(StreamWrite::Data(data.to_vec())) + .is_err() + { + warn!( + "stream {}: tcp write channel closed queued={} rx_bytes={} tx_bytes={}", + stream_id, stream.queued_bytes, stream.rx_bytes, stream.tx_bytes + ); + reset_stream = true; + } else { + stream.queued_bytes = projected; + } + + if !stream.discarding && stream.consumed_offset < stream.rx_bytes { + stream.consumed_offset = stream.rx_bytes; + let ret = unsafe { + picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) + }; + if ret < 0 { + warn!( + "stream {}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + reset_stream = true; + } + } + } else if stream .write_tx .send(StreamWrite::Data(data.to_vec())) .is_err() @@ -271,32 +397,61 @@ fn handle_stream_data( ); reset_stream = true; } else { - stream.queued_bytes = stream.queued_bytes.saturating_add(data.len()); + stream.queued_bytes = projected; + } + + if !multi_stream && reserve_bytes > 0 && !stream.discarding { + let drained = stream.rx_bytes.saturating_sub(stream.queued_bytes as u64); + let mut target_offset = drained + .saturating_add(reserve_bytes as u64) + .min(stream.rx_bytes); + if let Some(fin_offset) = stream.fin_offset { + if target_offset > fin_offset { + target_offset = fin_offset; + } + } + if target_offset > stream.consumed_offset { + stream.consumed_offset = target_offset; + let ret = unsafe { + picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) + }; + if ret < 0 { + warn!( + "stream {}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + reset_stream = true; + } + } } } if fin { - if stream.fin_offset.is_none() { - stream.fin_offset = Some(stream.rx_bytes); - } - stream.data_rx = None; - if !stream.fin_enqueued { - if stream.write_tx.send(StreamWrite::Fin).is_err() { - warn!( - "stream {}: tcp write channel closed on fin queued={} rx_bytes={} tx_bytes={}", - stream_id, - stream.queued_bytes, - stream.rx_bytes, - stream.tx_bytes - ); - reset_stream = true; - } else { - stream.fin_enqueued = true; + if stream.discarding { + remove_stream = true; + } else { + if stream.fin_offset.is_none() { + stream.fin_offset = Some(stream.rx_bytes); + } + stream.data_rx = None; + if !stream.fin_enqueued { + if stream.write_tx.send(StreamWrite::Fin).is_err() { + warn!( + "stream {}: tcp write channel closed on fin queued={} rx_bytes={} tx_bytes={}", + stream_id, + stream.queued_bytes, + stream.rx_bytes, + stream.tx_bytes + ); + reset_stream = true; + } else { + stream.fin_enqueued = true; + } } } } - if !reset_stream && stream.fin_enqueued && stream.queued_bytes == 0 { + if !reset_stream && !stream.discarding && stream.fin_enqueued && stream.queued_bytes == 0 { remove_stream = true; } } @@ -412,8 +567,14 @@ pub(crate) fn handle_command( consumed_offset: 0, fin_offset: None, fin_enqueued: false, + discarding: false, + stop_sending_sent: false, }, ); + if !state.multi_stream_mode && state.streams.len() > 1 { + state.multi_stream_mode = true; + promote_streams(cnx, state); + } let _ = unsafe { picoquic_mark_active_stream(cnx, stream_id, 1, std::ptr::null_mut()) }; if state.debug_streams { debug!("stream {}: accepted", stream_id); @@ -484,33 +645,41 @@ pub(crate) fn handle_command( } Command::StreamWriteDrained { stream_id, bytes } => { let mut remove_stream = false; - let mut reset_stream = false; if let Some(stream) = state.streams.get_mut(&stream_id) { + if stream.discarding { + return; + } stream.queued_bytes = stream.queued_bytes.saturating_sub(bytes); - stream.consumed_offset = stream.consumed_offset.saturating_add(bytes as u64); - if let Some(fin_offset) = stream.fin_offset { - if stream.consumed_offset > fin_offset { - stream.consumed_offset = fin_offset; + if !state.multi_stream_mode { + let mut new_offset = stream.consumed_offset.saturating_add(bytes as u64); + if let Some(fin_offset) = stream.fin_offset { + if new_offset > fin_offset { + new_offset = fin_offset; + } + } + if new_offset > stream.consumed_offset { + stream.consumed_offset = new_offset; + let ret = unsafe { + picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) + }; + if ret < 0 { + warn!( + "stream {}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + let _ = unsafe { + picoquic_reset_stream(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) + }; + state.streams.remove(&stream_id); + return; + } } } - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - reset_stream = true; - } else if stream.fin_enqueued && stream.queued_bytes == 0 { + if stream.fin_enqueued && stream.queued_bytes == 0 { remove_stream = true; } } - if reset_stream { - let _ = - unsafe { picoquic_reset_stream(cnx, stream_id, SLIPSTREAM_FILE_CANCEL_ERROR) }; - state.streams.remove(&stream_id); - } else if remove_stream { + if remove_stream { state.streams.remove(&stream_id); } } diff --git a/crates/slipstream-ffi/src/picoquic.rs b/crates/slipstream-ffi/src/picoquic.rs index 705cc07d..8b157a86 100644 --- a/crates/slipstream-ffi/src/picoquic.rs +++ b/crates/slipstream-ffi/src/picoquic.rs @@ -397,6 +397,11 @@ extern "C" { stream_id: u64, local_stream_error: u64, ) -> c_int; + pub fn picoquic_stop_sending( + cnx: *mut picoquic_cnx_t, + stream_id: u64, + local_stream_error: u64, + ) -> c_int; pub fn picoquic_stream_data_consumed( cnx: *mut picoquic_cnx_t, stream_id: u64, diff --git a/crates/slipstream-ffi/src/runtime.rs b/crates/slipstream-ffi/src/runtime.rs index 1270b651..870cf042 100644 --- a/crates/slipstream-ffi/src/runtime.rs +++ b/crates/slipstream-ffi/src/runtime.rs @@ -61,8 +61,9 @@ pub unsafe fn configure_quic_with_custom( } /// Configure shared QUIC defaults. -/// Backpressure is enforced via a connection-level `max_data` cap (shared across streams), -/// rather than per-stream buffer limits/reset. +/// Connection-level `max_data` is still configured. Stream handlers apply a small reserve in +/// single-stream mode, then switch to per-stream caps with STOP_SENDING + discard when multiple +/// streams are active to avoid connection-wide stalls. /// /// # Safety /// `quic` must be a valid picoquic context and `mtu` must be non-zero. diff --git a/crates/slipstream-server/src/streams.rs b/crates/slipstream-server/src/streams.rs index 5ae9e317..1c32781e 100644 --- a/crates/slipstream-server/src/streams.rs +++ b/crates/slipstream-server/src/streams.rs @@ -4,13 +4,14 @@ use slipstream_ffi::picoquic::{ picoquic_call_back_event_t, picoquic_close, picoquic_close_immediate, picoquic_cnx_t, picoquic_get_first_cnx, picoquic_get_next_cnx, picoquic_mark_active_stream, picoquic_provide_stream_data_buffer, picoquic_quic_t, picoquic_reset_stream, - picoquic_stream_data_consumed, + picoquic_stop_sending, picoquic_stream_data_consumed, }; use slipstream_ffi::{SLIPSTREAM_FILE_CANCEL_ERROR, SLIPSTREAM_INTERNAL_ERROR}; -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; +use std::sync::OnceLock; use std::time::{Duration, Instant}; use tokio::sync::{mpsc, watch}; use tracing::{debug, error, warn}; @@ -18,6 +19,7 @@ use tracing::{debug, error, warn}; pub(crate) struct ServerState { target_addr: SocketAddr, streams: HashMap, + multi_streams: HashSet, command_tx: mpsc::UnboundedSender, debug_streams: bool, debug_commands: bool, @@ -35,6 +37,7 @@ impl ServerState { Self { target_addr, streams: HashMap::new(), + multi_streams: HashSet::new(), command_tx, debug_streams, debug_commands, @@ -99,6 +102,67 @@ struct ServerStream { pending_data: VecDeque>, pending_fin: bool, fin_enqueued: bool, + discarding: bool, + stop_sending_sent: bool, +} + +const DEFAULT_STREAM_QUEUE_MAX_BYTES: usize = 2 * 1024 * 1024; +const DEFAULT_CONN_RESERVE_BYTES: usize = 64 * 1024; + +fn stream_queue_max_bytes() -> usize { + static MAX_BYTES: OnceLock = OnceLock::new(); + *MAX_BYTES.get_or_init(|| { + std::env::var("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES") + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_STREAM_QUEUE_MAX_BYTES) + }) +} + +fn conn_reserve_bytes() -> usize { + static RESERVE_BYTES: OnceLock = OnceLock::new(); + *RESERVE_BYTES.get_or_init(|| { + std::env::var("SLIPSTREAM_CONN_RESERVE_BYTES") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(DEFAULT_CONN_RESERVE_BYTES) + }) +} + +fn mark_multi_stream(state: &mut ServerState, cnx_id: usize) -> bool { + if state.multi_streams.contains(&cnx_id) { + return false; + } + let count = state.streams.keys().filter(|key| key.cnx == cnx_id).count(); + if count > 1 { + state.multi_streams.insert(cnx_id); + true + } else { + false + } +} + +fn promote_connection_streams(cnx: *mut picoquic_cnx_t, state: &mut ServerState, cnx_id: usize) { + for (key, stream) in state + .streams + .iter_mut() + .filter(|(key, _)| key.cnx == cnx_id) + { + if stream.discarding { + continue; + } + if stream.consumed_offset < stream.rx_bytes { + stream.consumed_offset = stream.rx_bytes; + let ret = unsafe { picoquic_stream_data_consumed(cnx, key.stream_id, stream.rx_bytes) }; + if ret < 0 { + warn!( + "stream {:?}: stream_data_consumed failed during promote ret={} consumed_offset={}", + key.stream_id, ret, stream.consumed_offset + ); + } + } + } } pub(crate) unsafe extern "C" fn server_callback( @@ -301,20 +365,22 @@ fn handle_stream_data( }; let debug_streams = state.debug_streams; let mut reset_stream = false; + let mut remove_stream = false; - { - let stream = state.streams.entry(key).or_insert_with(|| { - let (shutdown_tx, shutdown_rx) = watch::channel(false); - if debug_streams { - debug!("stream {:?}: connecting", key.stream_id); - } - spawn_target_connector( - key, - state.target_addr, - state.command_tx.clone(), - debug_streams, - shutdown_rx, - ); + if !state.streams.contains_key(&key) { + let (shutdown_tx, shutdown_rx) = watch::channel(false); + if debug_streams { + debug!("stream {:?}: connecting", key.stream_id); + } + spawn_target_connector( + key, + state.target_addr, + state.command_tx.clone(), + debug_streams, + shutdown_rx, + ); + state.streams.insert( + key, ServerStream { write_tx: None, data_rx: None, @@ -331,50 +397,192 @@ fn handle_stream_data( pending_data: VecDeque::new(), pending_fin: false, fin_enqueued: false, + discarding: false, + stop_sending_sent: false, + }, + ); + } + + if mark_multi_stream(state, key.cnx) { + promote_connection_streams(cnx, state, key.cnx); + } + let multi_stream = state.multi_streams.contains(&key.cnx); + let reserve_bytes = if multi_stream { + 0 + } else { + conn_reserve_bytes() + }; + + { + let stream = match state.streams.get_mut(&key) { + Some(stream) => stream, + None => return, + }; + + if stream.discarding { + if !data.is_empty() { + let incoming_len = data.len(); + stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); + if stream.consumed_offset < stream.rx_bytes { + stream.consumed_offset = stream.rx_bytes; + let ret = unsafe { + picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) + }; + if ret < 0 { + warn!( + "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + } + } } - }); + } else if !data.is_empty() { + let incoming_len = data.len(); + stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); + let projected = stream.queued_bytes.saturating_add(incoming_len); + + if multi_stream { + let max_queue = stream_queue_max_bytes(); + if projected > max_queue { + warn!( + "stream {:?}: queued_bytes {} + {} exceeds limit {}; stopping", + stream_id, stream.queued_bytes, incoming_len, max_queue + ); + stream.discarding = true; + stream.pending_data.clear(); + stream.queued_bytes = 0; + stream.pending_fin = false; + stream.fin_enqueued = false; + stream.data_rx = None; + stream.write_tx = None; + stream.send_pending = None; + stream.send_stash = None; + stream.target_fin_pending = false; + stream.close_after_flush = false; + let _ = stream.shutdown_tx.send(true); + if stream.consumed_offset < stream.rx_bytes { + stream.consumed_offset = stream.rx_bytes; + let ret = unsafe { + picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) + }; + if ret < 0 { + warn!( + "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + } + } + if !stream.stop_sending_sent { + let _ = unsafe { + picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) + }; + stream.stop_sending_sent = true; + } + } else if let Some(write_tx) = stream.write_tx.as_ref() { + if write_tx.send(StreamWrite::Data(data.to_vec())).is_err() { + reset_stream = true; + } else { + stream.queued_bytes = projected; + } + } else { + stream.pending_data.push_back(data.to_vec()); + stream.queued_bytes = projected; + } - if !data.is_empty() { - // Backpressure is enforced via connection-level max_data, not per-stream buffer caps. - stream.rx_bytes = stream.rx_bytes.saturating_add(data.len() as u64); - if let Some(write_tx) = stream.write_tx.as_ref() { + if !stream.discarding && stream.consumed_offset < stream.rx_bytes { + stream.consumed_offset = stream.rx_bytes; + let ret = unsafe { + picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) + }; + if ret < 0 { + warn!( + "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + reset_stream = true; + } + } + } else if let Some(write_tx) = stream.write_tx.as_ref() { if write_tx.send(StreamWrite::Data(data.to_vec())).is_err() { reset_stream = true; } else { - stream.queued_bytes = stream.queued_bytes.saturating_add(data.len()); + stream.queued_bytes = projected; } } else { stream.pending_data.push_back(data.to_vec()); - stream.queued_bytes = stream.queued_bytes.saturating_add(data.len()); + stream.queued_bytes = projected; + } + + if !multi_stream && reserve_bytes > 0 && !stream.discarding { + let drained = stream.rx_bytes.saturating_sub(stream.queued_bytes as u64); + let mut target_offset = drained + .saturating_add(reserve_bytes as u64) + .min(stream.rx_bytes); + if let Some(fin_offset) = stream.fin_offset { + if target_offset > fin_offset { + target_offset = fin_offset; + } + } + if target_offset > stream.consumed_offset { + stream.consumed_offset = target_offset; + let ret = unsafe { + picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) + }; + if ret < 0 { + warn!( + "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + reset_stream = true; + } + } } } if fin { - if stream.fin_offset.is_none() { - stream.fin_offset = Some(stream.rx_bytes); - } - if !stream.fin_enqueued { - if stream.write_tx.is_some() && stream.pending_data.is_empty() { - if let Some(write_tx) = stream.write_tx.as_ref() { - if write_tx.send(StreamWrite::Fin).is_err() { - reset_stream = true; - } else { - stream.fin_enqueued = true; - stream.pending_fin = false; + if stream.discarding { + if !reset_stream { + remove_stream = true; + } + } else { + if stream.fin_offset.is_none() { + stream.fin_offset = Some(stream.rx_bytes); + } + if !stream.fin_enqueued { + if stream.write_tx.is_some() && stream.pending_data.is_empty() { + if let Some(write_tx) = stream.write_tx.as_ref() { + if write_tx.send(StreamWrite::Fin).is_err() { + reset_stream = true; + } else { + stream.fin_enqueued = true; + stream.pending_fin = false; + } } + } else { + stream.pending_fin = true; } - } else { - stream.pending_fin = true; } } } } + if remove_stream { + shutdown_stream(state, key); + return; + } + if reset_stream { if debug_streams { debug!("stream {:?}: resetting", stream_id); } - shutdown_stream(state, key); + if !state + .streams + .get(&key) + .map(|stream| stream.discarding) + .unwrap_or(false) + { + shutdown_stream(state, key); + } unsafe { let _ = picoquic_reset_stream(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR); } @@ -391,6 +599,7 @@ pub(crate) fn remove_connection_streams(state: &mut ServerState, cnx: usize) { for key in keys { shutdown_stream(state, key); } + state.multi_streams.remove(&cnx); } fn shutdown_stream(state: &mut ServerState, key: StreamKey) -> Option { @@ -435,6 +644,13 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { if state.debug_streams { debug!("stream {:?}: target connected", stream_id); } + if stream.discarding { + stream.pending_data.clear(); + stream.pending_fin = false; + stream.fin_enqueued = false; + let _ = stream.shutdown_tx.send(true); + return; + } stream.write_tx = Some(write_tx); stream.data_rx = Some(data_rx); stream.send_pending = Some(send_pending); @@ -584,26 +800,34 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { }; let mut reset_stream = false; if let Some(stream) = state.streams.get_mut(&key) { + if stream.discarding { + return; + } stream.queued_bytes = stream.queued_bytes.saturating_sub(bytes); - stream.consumed_offset = stream.consumed_offset.saturating_add(bytes as u64); - if let Some(fin_offset) = stream.fin_offset { - if stream.consumed_offset > fin_offset { - stream.consumed_offset = fin_offset; + if !state.multi_streams.contains(&cnx_id) { + let mut new_offset = stream.consumed_offset.saturating_add(bytes as u64); + if let Some(fin_offset) = stream.fin_offset { + if new_offset > fin_offset { + new_offset = fin_offset; + } + } + if new_offset > stream.consumed_offset { + stream.consumed_offset = new_offset; + let ret = unsafe { + picoquic_stream_data_consumed( + cnx_id as *mut picoquic_cnx_t, + stream_id, + stream.consumed_offset, + ) + }; + if ret < 0 { + warn!( + "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", + stream_id, ret, stream.consumed_offset + ); + reset_stream = true; + } } - } - let ret = unsafe { - picoquic_stream_data_consumed( - cnx_id as *mut picoquic_cnx_t, - stream_id, - stream.consumed_offset, - ) - }; - if ret < 0 { - warn!( - "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - reset_stream = true; } } if reset_stream { @@ -656,5 +880,6 @@ pub(crate) fn handle_shutdown(quic: *mut picoquic_quic_t, state: &mut ServerStat cnx = next; } state.streams.clear(); + state.multi_streams.clear(); true } diff --git a/crates/slipstream-server/tests/flow_control_e2e.rs b/crates/slipstream-server/tests/flow_control_e2e.rs new file mode 100644 index 00000000..1bd3052b --- /dev/null +++ b/crates/slipstream-server/tests/flow_control_e2e.rs @@ -0,0 +1,464 @@ +mod support; + +use std::io::{Read, Write}; +use std::net::{Ipv4Addr, SocketAddr, TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{self, Receiver}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +use support::{ + ensure_client_bin, log_snapshot, pick_tcp_port, pick_udp_port, server_bin_path, spawn_client, + spawn_server, wait_for_log, ClientArgs, LogCapture, ServerArgs, +}; + +const ENV_ENABLE: &str = "SLIPSTREAM_FLOW_CONTROL_TEST"; + +fn assert_log_absent(logs: &LogCapture, needle: &str, duration: Duration) { + let deadline = Instant::now() + duration; + while Instant::now() < deadline { + let snapshot = log_snapshot(logs); + if snapshot.contains(needle) { + panic!("unexpected log entry: {}", needle); + } + thread::sleep(Duration::from_millis(50)); + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum TargetMode { + Blackhole, + Echo, +} + +#[derive(Debug)] +enum TargetEvent { + Accepted { index: usize, mode: TargetMode }, +} + +struct SplitTarget { + addr: SocketAddr, + stop: Arc, + handle: Option>, + conn_handles: Arc>>>, + rx: Receiver, +} + +impl SplitTarget { + fn spawn() -> std::io::Result { + let listener = TcpListener::bind("127.0.0.1:0")?; + listener.set_nonblocking(true)?; + let addr = listener.local_addr()?; + let (tx, rx) = mpsc::channel(); + let stop = Arc::new(AtomicBool::new(false)); + let stop_flag = Arc::clone(&stop); + let conn_handles: Arc>>> = + Arc::new(Mutex::new(Vec::new())); + let conn_handles_clone = Arc::clone(&conn_handles); + + let handle = thread::spawn(move || { + let mut index = 0usize; + while !stop_flag.load(Ordering::Relaxed) { + match listener.accept() { + Ok((stream, _)) => { + let mode = if index == 0 { + TargetMode::Blackhole + } else { + TargetMode::Echo + }; + let _ = tx.send(TargetEvent::Accepted { index, mode }); + let stop_conn = Arc::clone(&stop_flag); + let join = thread::spawn(move || { + let _ = stream.set_nodelay(true); + match mode { + TargetMode::Blackhole => { + while !stop_conn.load(Ordering::Relaxed) { + thread::sleep(Duration::from_millis(100)); + } + } + TargetMode::Echo => { + let mut stream = stream; + let _ = + stream.set_read_timeout(Some(Duration::from_millis(200))); + let mut buf = [0u8; 4096]; + while !stop_conn.load(Ordering::Relaxed) { + match stream.read(&mut buf) { + Ok(0) => break, + Ok(n) => { + if stream.write_all(&buf[..n]).is_err() { + break; + } + } + Err(err) + if err.kind() == std::io::ErrorKind::TimedOut + || err.kind() + == std::io::ErrorKind::WouldBlock => + { + continue; + } + Err(_) => break, + } + } + } + } + }); + if let Ok(mut handles) = conn_handles_clone.lock() { + handles.push(join); + } + index = index.saturating_add(1); + } + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(10)); + } + Err(_) => break, + } + } + }); + + Ok(Self { + addr, + stop, + handle: Some(handle), + conn_handles, + rx, + }) + } + + fn recv_event(&self, timeout: Duration) -> Option { + self.rx.recv_timeout(timeout).ok() + } +} + +impl Drop for SplitTarget { + fn drop(&mut self) { + self.stop.store(true, Ordering::Relaxed); + if let Some(handle) = self.handle.take() { + let _ = handle.join(); + } + if let Ok(mut handles) = self.conn_handles.lock() { + for handle in handles.drain(..) { + let _ = handle.join(); + } + } + } +} + +#[test] +fn blocked_stream_should_not_stall_other_streams() { + if std::env::var(ENV_ENABLE).is_err() { + eprintln!( + "skipping flow control e2e test; set {}=1 to enable", + ENV_ENABLE + ); + return; + } + + let root = support::workspace_root(); + let client_bin = ensure_client_bin(&root); + let server_bin = server_bin_path(); + + let cert = root.join("fixtures/certs/cert.pem"); + let key = root.join("fixtures/certs/key.pem"); + assert!(cert.exists(), "missing fixtures/certs/cert.pem"); + assert!(key.exists(), "missing fixtures/certs/key.pem"); + + let dns_port = match pick_udp_port() { + Ok(port) => port, + Err(err) => { + eprintln!("skipping flow control e2e test: {}", err); + return; + } + }; + let tcp_port = match pick_tcp_port() { + Ok(port) => port, + Err(err) => { + eprintln!("skipping flow control e2e test: {}", err); + return; + } + }; + + let target = match SplitTarget::spawn() { + Ok(target) => target, + Err(err) => { + eprintln!("skipping flow control e2e test: {}", err); + return; + } + }; + let domain = "test.example.com"; + + std::env::set_var("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES", "65536"); + std::env::set_var("SLIPSTREAM_CONN_RESERVE_BYTES", "65536"); + std::env::set_var("SLIPSTREAM_STREAM_WRITE_BUFFER_BYTES", "8388608"); + + let (mut server, server_logs) = spawn_server(ServerArgs { + server_bin: &server_bin, + dns_listen_host: Some("127.0.0.1"), + dns_port, + target_address: &format!("127.0.0.1:{}", target.addr.port()), + domains: &[domain], + cert: &cert, + key: &key, + reset_seed_path: None, + fallback_addr: None, + idle_timeout_seconds: None, + rust_log: "info", + capture_logs: true, + }); + let server_logs = server_logs.expect("server logs"); + thread::sleep(Duration::from_millis(200)); + if server.has_exited() { + eprintln!("skipping flow control e2e test: server failed to start"); + return; + } + + let (_client, client_logs) = spawn_client(ClientArgs { + client_bin: &client_bin, + dns_port, + tcp_port, + domain, + cert: Some(&cert), + keep_alive_interval: Some(0), + rust_log: "info", + capture_logs: true, + }); + let client_logs = client_logs.expect("client logs"); + if !wait_for_log( + &client_logs, + "Listening on TCP port", + Duration::from_secs(5), + ) { + let snapshot = log_snapshot(&client_logs); + panic!("client did not start listening\n{}", snapshot); + } + if !wait_for_log(&client_logs, "Connection ready", Duration::from_secs(10)) { + let snapshot = log_snapshot(&client_logs); + panic!("client did not become ready\n{}", snapshot); + } + + let client_addr = SocketAddr::from((Ipv4Addr::LOCALHOST, tcp_port)); + let mut blocked = TcpStream::connect_timeout(&client_addr, Duration::from_secs(2)) + .expect("connect blocked stream"); + let _ = blocked.set_nodelay(true); + let _ = blocked.set_write_timeout(Some(Duration::from_millis(200))); + + if !wait_for_log(&client_logs, "Accepted TCP stream", Duration::from_secs(5)) { + let snapshot = log_snapshot(&client_logs); + panic!("client did not accept blocked stream\n{}", snapshot); + } + + let warmup = vec![0u8; 1024]; + let _ = blocked.write_all(&warmup); + + match target.recv_event(Duration::from_secs(5)) { + Some(TargetEvent::Accepted { index, mode }) => { + assert_eq!(index, 0, "expected first target connection to be index 0"); + assert_eq!(mode, TargetMode::Blackhole, "expected blackhole target"); + } + None => { + let snapshot = log_snapshot(&server_logs); + panic!("target did not accept blackhole connection\n{}", snapshot); + } + } + + let send_deadline = Instant::now() + Duration::from_secs(3); + let payload = vec![0u8; 32 * 1024]; + while Instant::now() < send_deadline { + match blocked.write(&payload) { + Ok(0) => break, + Ok(_) => continue, + Err(_) => break, + } + } + + let mut echo = TcpStream::connect_timeout(&client_addr, Duration::from_secs(2)) + .expect("connect echo stream"); + let _ = echo.set_nodelay(true); + let _ = echo.set_read_timeout(Some(Duration::from_millis(200))); + + if !wait_for_log(&client_logs, "Accepted TCP stream", Duration::from_secs(5)) { + let snapshot = log_snapshot(&client_logs); + panic!("client did not accept echo stream\n{}", snapshot); + } + + let ping = b"slipstream-flow-control"; + echo.write_all(ping).expect("write echo payload"); + + match target.recv_event(Duration::from_secs(5)) { + Some(TargetEvent::Accepted { index, mode }) => { + assert_eq!(index, 1, "expected second target connection to be index 1"); + assert_eq!(mode, TargetMode::Echo, "expected echo target"); + } + None => { + let snapshot = log_snapshot(&server_logs); + panic!("target did not accept echo connection\n{}", snapshot); + } + } + + let mut buf = vec![0u8; ping.len()]; + let mut read_total = 0usize; + let deadline = Instant::now() + Duration::from_secs(10); + while read_total < buf.len() && Instant::now() < deadline { + match echo.read(&mut buf[read_total..]) { + Ok(0) => break, + Ok(n) => read_total += n, + Err(err) + if err.kind() == std::io::ErrorKind::WouldBlock + || err.kind() == std::io::ErrorKind::TimedOut => + { + continue; + } + Err(err) => panic!("read echo payload failed: {}", err), + } + } + if read_total != buf.len() { + let client_snapshot = log_snapshot(&client_logs); + let server_snapshot = log_snapshot(&server_logs); + panic!( + "echo payload incomplete (got {} of {})\nclient logs:\n{}\nserver logs:\n{}", + read_total, + buf.len(), + client_snapshot, + server_snapshot + ); + } + if buf.as_slice() != ping { + let client_snapshot = log_snapshot(&client_logs); + let server_snapshot = log_snapshot(&server_logs); + panic!( + "echo payload mismatch\nclient logs:\n{}\nserver logs:\n{}", + client_snapshot, server_snapshot + ); + } +} + +#[test] +fn single_stream_slow_transfer_should_not_abort() { + if std::env::var(ENV_ENABLE).is_err() { + eprintln!( + "skipping flow control e2e test; set {}=1 to enable", + ENV_ENABLE + ); + return; + } + + let root = support::workspace_root(); + let client_bin = ensure_client_bin(&root); + let server_bin = server_bin_path(); + + let cert = root.join("fixtures/certs/cert.pem"); + let key = root.join("fixtures/certs/key.pem"); + assert!(cert.exists(), "missing fixtures/certs/cert.pem"); + assert!(key.exists(), "missing fixtures/certs/key.pem"); + + let dns_port = match pick_udp_port() { + Ok(port) => port, + Err(err) => { + eprintln!("skipping flow control e2e test: {}", err); + return; + } + }; + let tcp_port = match pick_tcp_port() { + Ok(port) => port, + Err(err) => { + eprintln!("skipping flow control e2e test: {}", err); + return; + } + }; + + let target = match SplitTarget::spawn() { + Ok(target) => target, + Err(err) => { + eprintln!("skipping flow control e2e test: {}", err); + return; + } + }; + let domain = "test.example.com"; + + std::env::set_var("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES", "32768"); + std::env::set_var("SLIPSTREAM_CONN_RESERVE_BYTES", "16384"); + std::env::set_var("SLIPSTREAM_STREAM_WRITE_BUFFER_BYTES", "8388608"); + + let (mut server, server_logs) = spawn_server(ServerArgs { + server_bin: &server_bin, + dns_listen_host: Some("127.0.0.1"), + dns_port, + target_address: &format!("127.0.0.1:{}", target.addr.port()), + domains: &[domain], + cert: &cert, + key: &key, + reset_seed_path: None, + fallback_addr: None, + idle_timeout_seconds: None, + rust_log: "info", + capture_logs: true, + }); + let server_logs = server_logs.expect("server logs"); + thread::sleep(Duration::from_millis(200)); + if server.has_exited() { + eprintln!("skipping flow control e2e test: server failed to start"); + return; + } + + let (_client, client_logs) = spawn_client(ClientArgs { + client_bin: &client_bin, + dns_port, + tcp_port, + domain, + cert: Some(&cert), + keep_alive_interval: Some(0), + rust_log: "info", + capture_logs: true, + }); + let client_logs = client_logs.expect("client logs"); + if !wait_for_log( + &client_logs, + "Listening on TCP port", + Duration::from_secs(5), + ) { + let snapshot = log_snapshot(&client_logs); + panic!("client did not start listening\n{}", snapshot); + } + if !wait_for_log(&client_logs, "Connection ready", Duration::from_secs(10)) { + let snapshot = log_snapshot(&client_logs); + panic!("client did not become ready\n{}", snapshot); + } + + let client_addr = SocketAddr::from((Ipv4Addr::LOCALHOST, tcp_port)); + let mut blocked = TcpStream::connect_timeout(&client_addr, Duration::from_secs(2)) + .expect("connect blocked stream"); + let _ = blocked.set_nodelay(true); + let _ = blocked.set_write_timeout(Some(Duration::from_millis(200))); + + if !wait_for_log(&client_logs, "Accepted TCP stream", Duration::from_secs(5)) { + let snapshot = log_snapshot(&client_logs); + panic!("client did not accept blocked stream\n{}", snapshot); + } + + let warmup = vec![0u8; 1024]; + let _ = blocked.write_all(&warmup); + + match target.recv_event(Duration::from_secs(5)) { + Some(TargetEvent::Accepted { index, mode }) => { + assert_eq!(index, 0, "expected first target connection to be index 0"); + assert_eq!(mode, TargetMode::Blackhole, "expected blackhole target"); + } + None => { + let snapshot = log_snapshot(&server_logs); + panic!("target did not accept blackhole connection\n{}", snapshot); + } + } + + let send_deadline = Instant::now() + Duration::from_secs(2); + let payload = vec![0u8; 64 * 1024]; + while Instant::now() < send_deadline { + match blocked.write(&payload) { + Ok(0) => break, + Ok(_) => continue, + Err(_) => break, + } + } + + assert_log_absent(&server_logs, "queued_bytes", Duration::from_secs(1)); + assert_log_absent(&client_logs, "reset event", Duration::from_secs(1)); +} diff --git a/docs/config.md b/docs/config.md index 584b93be..19a9c3e0 100644 --- a/docs/config.md +++ b/docs/config.md @@ -7,6 +7,12 @@ This page documents runtime knobs and environment variables. - SLIPSTREAM_STREAM_WRITE_BUFFER_BYTES Overrides the connection-level QUIC max_data limit used for backpressure. Default is 8 MiB. Values must be positive integers. +- SLIPSTREAM_STREAM_QUEUE_MAX_BYTES + Per-stream receive queue cap enforced when multiple QUIC streams are active. + Default is 2 MiB. Values must be positive integers. +- SLIPSTREAM_CONN_RESERVE_BYTES + Minimum connection-level receive window to keep available for new streams in + single-stream mode. Default is 64 KiB. Set to 0 to disable the reserve. ## TLS certificates diff --git a/docs/design.md b/docs/design.md index 0c8bf5ff..8f257b03 100644 --- a/docs/design.md +++ b/docs/design.md @@ -43,6 +43,28 @@ The runtime centers around a connection manager that owns QUIC state, timers, an per-connection queues. UDP receive/send and TCP accept/read/write are handled by separate tasks, with bounded channels used to limit memory growth under load. +## Flow control strategy + +Slipstream needs to satisfy two competing cases: + +- Slow, long-lived single-stream transfers should behave like TCP. We want + application backpressure to propagate to the sender so large uploads don't get + aborted just because the target is slow. +- If a second stream appears, a stalled or blackholed stream must not be able to + exhaust the connection-level window and block new streams. This showed up in + practice as “new TCP connections hang” even though the QUIC connection is + still alive. + +To cover both: + +- In single-stream mode, we rely on TCP-style backpressure (consume after TCP + writes) but keep a small reserve window (SLIPSTREAM_CONN_RESERVE_BYTES) so + a new stream can always send its first bytes and trigger mode switch. +- Once multiple streams are active, we switch to consume-on-receive and enforce + per-stream caps (SLIPSTREAM_STREAM_QUEUE_MAX_BYTES). If a stream exceeds its + cap, we send STOP_SENDING and discard further data for that stream while + continuing to consume, which prevents connection-wide stalls. + ## Rust vs C behavior notes - The Rust client clamps active DNS polling sleeps to `DNS_POLL_SLICE_US` (50 ms), @@ -63,7 +85,8 @@ separate tasks, with bounded channels used to limit memory growth under load. ## Safety and shutdown - CLI validation enforces required flags and valid host:port parsing. -- Backpressure is applied via connection-level max_data. +- Backpressure uses connection-level max_data with a single-stream reserve and + per-stream caps in multi-stream mode to avoid global stalls. - Shutdown follows explicit states (drain, close, force terminate) to avoid hangs and minimize data loss. diff --git a/docs/protocol.md b/docs/protocol.md index d8594b62..bc3fea34 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -95,10 +95,15 @@ Otherwise, the response is ignored (including NAME_ERROR, which signals no data) ## Backpressure and buffering - Connection-level max_data is set to stream_write_buffer_bytes (default 8 MiB). -- Stream receive buffering relies on connection-level flow control; there is no - per-stream buffer cap/reset when enqueueing data to TCP writers. -- picoquic_stream_data_consumed is called after TCP writes drain, so peers are - backpressured when the connection window is full. +- When only one stream is active, stream data consumption tracks TCP write + drain and a small reserve window (SLIPSTREAM_CONN_RESERVE_BYTES) is kept + open to allow new streams to send their first bytes. +- When multiple streams are active, each stream enforces a per-stream receive + cap (SLIPSTREAM_STREAM_QUEUE_MAX_BYTES). On overflow, the receiver sends + STOP_SENDING, discards data for that stream, and continues consuming to + avoid connection-level stalls. +- Once a connection enters multi-stream mode it stays there for the remainder + of the connection. ## Path handling From cc3d794644fac094764dc3887f8b29f779b00100 Mon Sep 17 00:00:00 2001 From: Mygod Date: Thu, 22 Jan 2026 23:23:56 -0800 Subject: [PATCH 2/9] =?UTF-8?q?Fix=20single=E2=80=91stream=20write?= =?UTF-8?q?=E2=80=91drain=20logic?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/slipstream-client/src/streams.rs | 10 +++++++++- crates/slipstream-server/src/streams.rs | 10 +++++++++- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/crates/slipstream-client/src/streams.rs b/crates/slipstream-client/src/streams.rs index 4134895f..81468838 100644 --- a/crates/slipstream-client/src/streams.rs +++ b/crates/slipstream-client/src/streams.rs @@ -651,7 +651,15 @@ pub(crate) fn handle_command( } stream.queued_bytes = stream.queued_bytes.saturating_sub(bytes); if !state.multi_stream_mode { - let mut new_offset = stream.consumed_offset.saturating_add(bytes as u64); + let reserve_bytes = conn_reserve_bytes(); + let drained = stream.rx_bytes.saturating_sub(stream.queued_bytes as u64); + let mut new_offset = if reserve_bytes > 0 { + drained + .saturating_add(reserve_bytes as u64) + .min(stream.rx_bytes) + } else { + drained + }; if let Some(fin_offset) = stream.fin_offset { if new_offset > fin_offset { new_offset = fin_offset; diff --git a/crates/slipstream-server/src/streams.rs b/crates/slipstream-server/src/streams.rs index 1c32781e..138054f9 100644 --- a/crates/slipstream-server/src/streams.rs +++ b/crates/slipstream-server/src/streams.rs @@ -805,7 +805,15 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { } stream.queued_bytes = stream.queued_bytes.saturating_sub(bytes); if !state.multi_streams.contains(&cnx_id) { - let mut new_offset = stream.consumed_offset.saturating_add(bytes as u64); + let reserve_bytes = conn_reserve_bytes(); + let drained = stream.rx_bytes.saturating_sub(stream.queued_bytes as u64); + let mut new_offset = if reserve_bytes > 0 { + drained + .saturating_add(reserve_bytes as u64) + .min(stream.rx_bytes) + } else { + drained + }; if let Some(fin_offset) = stream.fin_offset { if new_offset > fin_offset { new_offset = fin_offset; From 831210346fed314522fbdfaa8342d92ed98c2107 Mon Sep 17 00:00:00 2001 From: Mygod Date: Thu, 22 Jan 2026 23:57:35 -0800 Subject: [PATCH 3/9] Address review comments --- crates/slipstream-client/src/streams.rs | 6 ++++-- crates/slipstream-server/src/streams.rs | 6 ++++-- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/crates/slipstream-client/src/streams.rs b/crates/slipstream-client/src/streams.rs index 81468838..a1fd2bfb 100644 --- a/crates/slipstream-client/src/streams.rs +++ b/crates/slipstream-client/src/streams.rs @@ -161,13 +161,15 @@ fn promote_streams(cnx: *mut picoquic_cnx_t, state: &mut ClientState) { continue; } if stream.consumed_offset < stream.rx_bytes { - stream.consumed_offset = stream.rx_bytes; - let ret = unsafe { picoquic_stream_data_consumed(cnx, *stream_id, stream.rx_bytes) }; + let new_offset = stream.rx_bytes; + let ret = unsafe { picoquic_stream_data_consumed(cnx, *stream_id, new_offset) }; if ret < 0 { warn!( "stream {}: stream_data_consumed failed during promote ret={} consumed_offset={}", stream_id, ret, stream.consumed_offset ); + } else { + stream.consumed_offset = new_offset; } } } diff --git a/crates/slipstream-server/src/streams.rs b/crates/slipstream-server/src/streams.rs index 138054f9..01eec820 100644 --- a/crates/slipstream-server/src/streams.rs +++ b/crates/slipstream-server/src/streams.rs @@ -153,13 +153,15 @@ fn promote_connection_streams(cnx: *mut picoquic_cnx_t, state: &mut ServerState, continue; } if stream.consumed_offset < stream.rx_bytes { - stream.consumed_offset = stream.rx_bytes; - let ret = unsafe { picoquic_stream_data_consumed(cnx, key.stream_id, stream.rx_bytes) }; + let new_offset = stream.rx_bytes; + let ret = unsafe { picoquic_stream_data_consumed(cnx, key.stream_id, new_offset) }; if ret < 0 { warn!( "stream {:?}: stream_data_consumed failed during promote ret={} consumed_offset={}", key.stream_id, ret, stream.consumed_offset ); + } else { + stream.consumed_offset = new_offset; } } } From 2b06ebe62b9f7dd89ed36c85feb0ea7e7defabf8 Mon Sep 17 00:00:00 2001 From: Mygod Date: Fri, 23 Jan 2026 00:06:26 -0800 Subject: [PATCH 4/9] Reduce dup --- crates/slipstream-client/src/streams.rs | 80 +++------ crates/slipstream-core/src/lib.rs | 1 + crates/slipstream-server/src/streams.rs | 80 +++------ .../tests/flow_control_e2e.rs | 160 +++++++----------- 4 files changed, 105 insertions(+), 216 deletions(-) diff --git a/crates/slipstream-client/src/streams.rs b/crates/slipstream-client/src/streams.rs index a1fd2bfb..57ed68fc 100644 --- a/crates/slipstream-client/src/streams.rs +++ b/crates/slipstream-client/src/streams.rs @@ -1,3 +1,6 @@ +use slipstream_core::flow_control::{ + conn_reserve_bytes, promote_consumed_offset, reserve_target_offset, stream_queue_max_bytes, +}; use slipstream_core::tcp::{stream_read_limit_chunks, tcp_send_buffer_bytes}; use slipstream_ffi::picoquic::{ picoquic_add_to_stream, picoquic_call_back_event_t, picoquic_cnx_t, picoquic_current_time, @@ -7,7 +10,7 @@ use slipstream_ffi::picoquic::{ }; use slipstream_ffi::{SLIPSTREAM_FILE_CANCEL_ERROR, SLIPSTREAM_INTERNAL_ERROR}; use std::collections::HashMap; -use std::sync::{Arc, OnceLock}; +use std::sync::Arc; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener as TokioTcpListener, TcpStream as TokioTcpStream}; use tokio::sync::{mpsc, oneshot, Notify}; @@ -16,29 +19,6 @@ use tracing::{debug, info, warn}; const STREAM_READ_CHUNK_BYTES: usize = 4096; const DEFAULT_TCP_RCVBUF_BYTES: usize = 256 * 1024; const CLIENT_WRITE_COALESCE_DEFAULT_BYTES: usize = 256 * 1024; -const DEFAULT_STREAM_QUEUE_MAX_BYTES: usize = 2 * 1024 * 1024; -const DEFAULT_CONN_RESERVE_BYTES: usize = 64 * 1024; - -fn stream_queue_max_bytes() -> usize { - static MAX_BYTES: OnceLock = OnceLock::new(); - *MAX_BYTES.get_or_init(|| { - std::env::var("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES") - .ok() - .and_then(|value| value.parse::().ok()) - .filter(|value| *value > 0) - .unwrap_or(DEFAULT_STREAM_QUEUE_MAX_BYTES) - }) -} - -fn conn_reserve_bytes() -> usize { - static RESERVE_BYTES: OnceLock = OnceLock::new(); - *RESERVE_BYTES.get_or_init(|| { - std::env::var("SLIPSTREAM_CONN_RESERVE_BYTES") - .ok() - .and_then(|value| value.parse::().ok()) - .unwrap_or(DEFAULT_CONN_RESERVE_BYTES) - }) -} pub(crate) struct ClientState { ready: bool, @@ -160,18 +140,17 @@ fn promote_streams(cnx: *mut picoquic_cnx_t, state: &mut ClientState) { if stream.discarding { continue; } - if stream.consumed_offset < stream.rx_bytes { - let new_offset = stream.rx_bytes; - let ret = unsafe { picoquic_stream_data_consumed(cnx, *stream_id, new_offset) }; - if ret < 0 { + promote_consumed_offset( + stream.rx_bytes, + &mut stream.consumed_offset, + |new_offset| unsafe { picoquic_stream_data_consumed(cnx, *stream_id, new_offset) }, + |ret, consumed_offset| { warn!( "stream {}: stream_data_consumed failed during promote ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset + stream_id, ret, consumed_offset ); - } else { - stream.consumed_offset = new_offset; - } - } + }, + ); } } @@ -403,15 +382,12 @@ fn handle_stream_data( } if !multi_stream && reserve_bytes > 0 && !stream.discarding { - let drained = stream.rx_bytes.saturating_sub(stream.queued_bytes as u64); - let mut target_offset = drained - .saturating_add(reserve_bytes as u64) - .min(stream.rx_bytes); - if let Some(fin_offset) = stream.fin_offset { - if target_offset > fin_offset { - target_offset = fin_offset; - } - } + let target_offset = reserve_target_offset( + stream.rx_bytes, + stream.queued_bytes, + stream.fin_offset, + reserve_bytes, + ); if target_offset > stream.consumed_offset { stream.consumed_offset = target_offset; let ret = unsafe { @@ -653,20 +629,12 @@ pub(crate) fn handle_command( } stream.queued_bytes = stream.queued_bytes.saturating_sub(bytes); if !state.multi_stream_mode { - let reserve_bytes = conn_reserve_bytes(); - let drained = stream.rx_bytes.saturating_sub(stream.queued_bytes as u64); - let mut new_offset = if reserve_bytes > 0 { - drained - .saturating_add(reserve_bytes as u64) - .min(stream.rx_bytes) - } else { - drained - }; - if let Some(fin_offset) = stream.fin_offset { - if new_offset > fin_offset { - new_offset = fin_offset; - } - } + let new_offset = reserve_target_offset( + stream.rx_bytes, + stream.queued_bytes, + stream.fin_offset, + conn_reserve_bytes(), + ); if new_offset > stream.consumed_offset { stream.consumed_offset = new_offset; let ret = unsafe { diff --git a/crates/slipstream-core/src/lib.rs b/crates/slipstream-core/src/lib.rs index 7f88460a..9e2acd3f 100644 --- a/crates/slipstream-core/src/lib.rs +++ b/crates/slipstream-core/src/lib.rs @@ -1,5 +1,6 @@ use std::fmt; +pub mod flow_control; mod macros; pub mod net; pub mod sip003; diff --git a/crates/slipstream-server/src/streams.rs b/crates/slipstream-server/src/streams.rs index 01eec820..f26a0818 100644 --- a/crates/slipstream-server/src/streams.rs +++ b/crates/slipstream-server/src/streams.rs @@ -1,5 +1,8 @@ use crate::server::{Command, StreamKey, StreamWrite}; use crate::target::spawn_target_connector; +use slipstream_core::flow_control::{ + conn_reserve_bytes, promote_consumed_offset, reserve_target_offset, stream_queue_max_bytes, +}; use slipstream_ffi::picoquic::{ picoquic_call_back_event_t, picoquic_close, picoquic_close_immediate, picoquic_cnx_t, picoquic_get_first_cnx, picoquic_get_next_cnx, picoquic_mark_active_stream, @@ -11,7 +14,6 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; -use std::sync::OnceLock; use std::time::{Duration, Instant}; use tokio::sync::{mpsc, watch}; use tracing::{debug, error, warn}; @@ -106,30 +108,6 @@ struct ServerStream { stop_sending_sent: bool, } -const DEFAULT_STREAM_QUEUE_MAX_BYTES: usize = 2 * 1024 * 1024; -const DEFAULT_CONN_RESERVE_BYTES: usize = 64 * 1024; - -fn stream_queue_max_bytes() -> usize { - static MAX_BYTES: OnceLock = OnceLock::new(); - *MAX_BYTES.get_or_init(|| { - std::env::var("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES") - .ok() - .and_then(|value| value.parse::().ok()) - .filter(|value| *value > 0) - .unwrap_or(DEFAULT_STREAM_QUEUE_MAX_BYTES) - }) -} - -fn conn_reserve_bytes() -> usize { - static RESERVE_BYTES: OnceLock = OnceLock::new(); - *RESERVE_BYTES.get_or_init(|| { - std::env::var("SLIPSTREAM_CONN_RESERVE_BYTES") - .ok() - .and_then(|value| value.parse::().ok()) - .unwrap_or(DEFAULT_CONN_RESERVE_BYTES) - }) -} - fn mark_multi_stream(state: &mut ServerState, cnx_id: usize) -> bool { if state.multi_streams.contains(&cnx_id) { return false; @@ -152,18 +130,17 @@ fn promote_connection_streams(cnx: *mut picoquic_cnx_t, state: &mut ServerState, if stream.discarding { continue; } - if stream.consumed_offset < stream.rx_bytes { - let new_offset = stream.rx_bytes; - let ret = unsafe { picoquic_stream_data_consumed(cnx, key.stream_id, new_offset) }; - if ret < 0 { + promote_consumed_offset( + stream.rx_bytes, + &mut stream.consumed_offset, + |new_offset| unsafe { picoquic_stream_data_consumed(cnx, key.stream_id, new_offset) }, + |ret, consumed_offset| { warn!( "stream {:?}: stream_data_consumed failed during promote ret={} consumed_offset={}", - key.stream_id, ret, stream.consumed_offset + key.stream_id, ret, consumed_offset ); - } else { - stream.consumed_offset = new_offset; - } - } + }, + ); } } @@ -516,15 +493,12 @@ fn handle_stream_data( } if !multi_stream && reserve_bytes > 0 && !stream.discarding { - let drained = stream.rx_bytes.saturating_sub(stream.queued_bytes as u64); - let mut target_offset = drained - .saturating_add(reserve_bytes as u64) - .min(stream.rx_bytes); - if let Some(fin_offset) = stream.fin_offset { - if target_offset > fin_offset { - target_offset = fin_offset; - } - } + let target_offset = reserve_target_offset( + stream.rx_bytes, + stream.queued_bytes, + stream.fin_offset, + reserve_bytes, + ); if target_offset > stream.consumed_offset { stream.consumed_offset = target_offset; let ret = unsafe { @@ -807,20 +781,12 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { } stream.queued_bytes = stream.queued_bytes.saturating_sub(bytes); if !state.multi_streams.contains(&cnx_id) { - let reserve_bytes = conn_reserve_bytes(); - let drained = stream.rx_bytes.saturating_sub(stream.queued_bytes as u64); - let mut new_offset = if reserve_bytes > 0 { - drained - .saturating_add(reserve_bytes as u64) - .min(stream.rx_bytes) - } else { - drained - }; - if let Some(fin_offset) = stream.fin_offset { - if new_offset > fin_offset { - new_offset = fin_offset; - } - } + let new_offset = reserve_target_offset( + stream.rx_bytes, + stream.queued_bytes, + stream.fin_offset, + conn_reserve_bytes(), + ); if new_offset > stream.consumed_offset { stream.consumed_offset = new_offset; let ret = unsafe { diff --git a/crates/slipstream-server/tests/flow_control_e2e.rs b/crates/slipstream-server/tests/flow_control_e2e.rs index 1bd3052b..198b0ed2 100644 --- a/crates/slipstream-server/tests/flow_control_e2e.rs +++ b/crates/slipstream-server/tests/flow_control_e2e.rs @@ -10,10 +10,11 @@ use std::time::{Duration, Instant}; use support::{ ensure_client_bin, log_snapshot, pick_tcp_port, pick_udp_port, server_bin_path, spawn_client, - spawn_server, wait_for_log, ClientArgs, LogCapture, ServerArgs, + spawn_server, wait_for_log, ChildGuard, ClientArgs, LogCapture, ServerArgs, }; const ENV_ENABLE: &str = "SLIPSTREAM_FLOW_CONTROL_TEST"; +const DOMAIN: &str = "test.example.com"; fn assert_log_absent(logs: &LogCapture, needle: &str, duration: Duration) { let deadline = Instant::now() + duration; @@ -144,14 +145,22 @@ impl Drop for SplitTarget { } } -#[test] -fn blocked_stream_should_not_stall_other_streams() { +struct FlowControlHarness { + _server: ChildGuard, + _client: ChildGuard, + server_logs: LogCapture, + client_logs: LogCapture, + target: SplitTarget, + client_addr: SocketAddr, +} + +fn setup_flow_control(envs: &[(&str, &str)]) -> Option { if std::env::var(ENV_ENABLE).is_err() { eprintln!( "skipping flow control e2e test; set {}=1 to enable", ENV_ENABLE ); - return; + return None; } let root = support::workspace_root(); @@ -167,14 +176,14 @@ fn blocked_stream_should_not_stall_other_streams() { Ok(port) => port, Err(err) => { eprintln!("skipping flow control e2e test: {}", err); - return; + return None; } }; let tcp_port = match pick_tcp_port() { Ok(port) => port, Err(err) => { eprintln!("skipping flow control e2e test: {}", err); - return; + return None; } }; @@ -182,21 +191,20 @@ fn blocked_stream_should_not_stall_other_streams() { Ok(target) => target, Err(err) => { eprintln!("skipping flow control e2e test: {}", err); - return; + return None; } }; - let domain = "test.example.com"; - std::env::set_var("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES", "65536"); - std::env::set_var("SLIPSTREAM_CONN_RESERVE_BYTES", "65536"); - std::env::set_var("SLIPSTREAM_STREAM_WRITE_BUFFER_BYTES", "8388608"); + for (key, value) in envs { + std::env::set_var(key, value); + } let (mut server, server_logs) = spawn_server(ServerArgs { server_bin: &server_bin, dns_listen_host: Some("127.0.0.1"), dns_port, target_address: &format!("127.0.0.1:{}", target.addr.port()), - domains: &[domain], + domains: &[DOMAIN], cert: &cert, key: &key, reset_seed_path: None, @@ -209,14 +217,14 @@ fn blocked_stream_should_not_stall_other_streams() { thread::sleep(Duration::from_millis(200)); if server.has_exited() { eprintln!("skipping flow control e2e test: server failed to start"); - return; + return None; } - let (_client, client_logs) = spawn_client(ClientArgs { + let (client, client_logs) = spawn_client(ClientArgs { client_bin: &client_bin, dns_port, tcp_port, - domain, + domain: DOMAIN, cert: Some(&cert), keep_alive_interval: Some(0), rust_log: "info", @@ -237,6 +245,31 @@ fn blocked_stream_should_not_stall_other_streams() { } let client_addr = SocketAddr::from((Ipv4Addr::LOCALHOST, tcp_port)); + + Some(FlowControlHarness { + _server: server, + _client: client, + server_logs, + client_logs, + target, + client_addr, + }) +} + +#[test] +fn blocked_stream_should_not_stall_other_streams() { + let Some(harness) = setup_flow_control(&[ + ("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES", "65536"), + ("SLIPSTREAM_CONN_RESERVE_BYTES", "65536"), + ("SLIPSTREAM_STREAM_WRITE_BUFFER_BYTES", "8388608"), + ]) else { + return; + }; + + let server_logs = &harness.server_logs; + let client_logs = &harness.client_logs; + let target = &harness.target; + let client_addr = harness.client_addr; let mut blocked = TcpStream::connect_timeout(&client_addr, Duration::from_secs(2)) .expect("connect blocked stream"); let _ = blocked.set_nodelay(true); @@ -334,97 +367,18 @@ fn blocked_stream_should_not_stall_other_streams() { #[test] fn single_stream_slow_transfer_should_not_abort() { - if std::env::var(ENV_ENABLE).is_err() { - eprintln!( - "skipping flow control e2e test; set {}=1 to enable", - ENV_ENABLE - ); + let Some(harness) = setup_flow_control(&[ + ("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES", "32768"), + ("SLIPSTREAM_CONN_RESERVE_BYTES", "16384"), + ("SLIPSTREAM_STREAM_WRITE_BUFFER_BYTES", "8388608"), + ]) else { return; - } - - let root = support::workspace_root(); - let client_bin = ensure_client_bin(&root); - let server_bin = server_bin_path(); - - let cert = root.join("fixtures/certs/cert.pem"); - let key = root.join("fixtures/certs/key.pem"); - assert!(cert.exists(), "missing fixtures/certs/cert.pem"); - assert!(key.exists(), "missing fixtures/certs/key.pem"); - - let dns_port = match pick_udp_port() { - Ok(port) => port, - Err(err) => { - eprintln!("skipping flow control e2e test: {}", err); - return; - } - }; - let tcp_port = match pick_tcp_port() { - Ok(port) => port, - Err(err) => { - eprintln!("skipping flow control e2e test: {}", err); - return; - } }; - let target = match SplitTarget::spawn() { - Ok(target) => target, - Err(err) => { - eprintln!("skipping flow control e2e test: {}", err); - return; - } - }; - let domain = "test.example.com"; - - std::env::set_var("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES", "32768"); - std::env::set_var("SLIPSTREAM_CONN_RESERVE_BYTES", "16384"); - std::env::set_var("SLIPSTREAM_STREAM_WRITE_BUFFER_BYTES", "8388608"); - - let (mut server, server_logs) = spawn_server(ServerArgs { - server_bin: &server_bin, - dns_listen_host: Some("127.0.0.1"), - dns_port, - target_address: &format!("127.0.0.1:{}", target.addr.port()), - domains: &[domain], - cert: &cert, - key: &key, - reset_seed_path: None, - fallback_addr: None, - idle_timeout_seconds: None, - rust_log: "info", - capture_logs: true, - }); - let server_logs = server_logs.expect("server logs"); - thread::sleep(Duration::from_millis(200)); - if server.has_exited() { - eprintln!("skipping flow control e2e test: server failed to start"); - return; - } - - let (_client, client_logs) = spawn_client(ClientArgs { - client_bin: &client_bin, - dns_port, - tcp_port, - domain, - cert: Some(&cert), - keep_alive_interval: Some(0), - rust_log: "info", - capture_logs: true, - }); - let client_logs = client_logs.expect("client logs"); - if !wait_for_log( - &client_logs, - "Listening on TCP port", - Duration::from_secs(5), - ) { - let snapshot = log_snapshot(&client_logs); - panic!("client did not start listening\n{}", snapshot); - } - if !wait_for_log(&client_logs, "Connection ready", Duration::from_secs(10)) { - let snapshot = log_snapshot(&client_logs); - panic!("client did not become ready\n{}", snapshot); - } - - let client_addr = SocketAddr::from((Ipv4Addr::LOCALHOST, tcp_port)); + let server_logs = &harness.server_logs; + let client_logs = &harness.client_logs; + let target = &harness.target; + let client_addr = harness.client_addr; let mut blocked = TcpStream::connect_timeout(&client_addr, Duration::from_secs(2)) .expect("connect blocked stream"); let _ = blocked.set_nodelay(true); From eb8e7acc8f0a1253bc2e6eac5b856749e8cf399a Mon Sep 17 00:00:00 2001 From: Mygod Date: Fri, 23 Jan 2026 00:12:21 -0800 Subject: [PATCH 5/9] Fix env in test --- .../tests/cert_pinning_e2e.rs | 3 ++ .../tests/flow_control_e2e.rs | 36 +++++++++---------- crates/slipstream-server/tests/idle_gc_e2e.rs | 3 ++ .../tests/restart_reconnect_e2e.rs | 3 ++ crates/slipstream-server/tests/support/mod.rs | 12 +++++-- .../tests/udp_fallback_e2e.rs | 1 + 6 files changed, 37 insertions(+), 21 deletions(-) diff --git a/crates/slipstream-server/tests/cert_pinning_e2e.rs b/crates/slipstream-server/tests/cert_pinning_e2e.rs index ce1997ff..09a6c302 100644 --- a/crates/slipstream-server/tests/cert_pinning_e2e.rs +++ b/crates/slipstream-server/tests/cert_pinning_e2e.rs @@ -57,6 +57,7 @@ fn cert_pinning_e2e() { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: None, + envs: &[], rust_log: "info", capture_logs: false, }); @@ -74,6 +75,7 @@ fn cert_pinning_e2e() { domain, cert: Some(&cert), keep_alive_interval: None, + envs: &[], rust_log: "info", capture_logs: true, }); @@ -107,6 +109,7 @@ fn cert_pinning_e2e() { domain: alt_domain, cert: Some(&alt_cert), keep_alive_interval: None, + envs: &[], rust_log: "info", capture_logs: true, }); diff --git a/crates/slipstream-server/tests/flow_control_e2e.rs b/crates/slipstream-server/tests/flow_control_e2e.rs index 198b0ed2..939b0642 100644 --- a/crates/slipstream-server/tests/flow_control_e2e.rs +++ b/crates/slipstream-server/tests/flow_control_e2e.rs @@ -195,10 +195,6 @@ fn setup_flow_control(envs: &[(&str, &str)]) -> Option { } }; - for (key, value) in envs { - std::env::set_var(key, value); - } - let (mut server, server_logs) = spawn_server(ServerArgs { server_bin: &server_bin, dns_listen_host: Some("127.0.0.1"), @@ -210,6 +206,7 @@ fn setup_flow_control(envs: &[(&str, &str)]) -> Option { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: None, + envs, rust_log: "info", capture_logs: true, }); @@ -227,6 +224,7 @@ fn setup_flow_control(envs: &[(&str, &str)]) -> Option { domain: DOMAIN, cert: Some(&cert), keep_alive_interval: Some(0), + envs, rust_log: "info", capture_logs: true, }); @@ -275,8 +273,8 @@ fn blocked_stream_should_not_stall_other_streams() { let _ = blocked.set_nodelay(true); let _ = blocked.set_write_timeout(Some(Duration::from_millis(200))); - if !wait_for_log(&client_logs, "Accepted TCP stream", Duration::from_secs(5)) { - let snapshot = log_snapshot(&client_logs); + if !wait_for_log(client_logs, "Accepted TCP stream", Duration::from_secs(5)) { + let snapshot = log_snapshot(client_logs); panic!("client did not accept blocked stream\n{}", snapshot); } @@ -289,7 +287,7 @@ fn blocked_stream_should_not_stall_other_streams() { assert_eq!(mode, TargetMode::Blackhole, "expected blackhole target"); } None => { - let snapshot = log_snapshot(&server_logs); + let snapshot = log_snapshot(server_logs); panic!("target did not accept blackhole connection\n{}", snapshot); } } @@ -309,8 +307,8 @@ fn blocked_stream_should_not_stall_other_streams() { let _ = echo.set_nodelay(true); let _ = echo.set_read_timeout(Some(Duration::from_millis(200))); - if !wait_for_log(&client_logs, "Accepted TCP stream", Duration::from_secs(5)) { - let snapshot = log_snapshot(&client_logs); + if !wait_for_log(client_logs, "Accepted TCP stream", Duration::from_secs(5)) { + let snapshot = log_snapshot(client_logs); panic!("client did not accept echo stream\n{}", snapshot); } @@ -323,7 +321,7 @@ fn blocked_stream_should_not_stall_other_streams() { assert_eq!(mode, TargetMode::Echo, "expected echo target"); } None => { - let snapshot = log_snapshot(&server_logs); + let snapshot = log_snapshot(server_logs); panic!("target did not accept echo connection\n{}", snapshot); } } @@ -345,8 +343,8 @@ fn blocked_stream_should_not_stall_other_streams() { } } if read_total != buf.len() { - let client_snapshot = log_snapshot(&client_logs); - let server_snapshot = log_snapshot(&server_logs); + let client_snapshot = log_snapshot(client_logs); + let server_snapshot = log_snapshot(server_logs); panic!( "echo payload incomplete (got {} of {})\nclient logs:\n{}\nserver logs:\n{}", read_total, @@ -356,8 +354,8 @@ fn blocked_stream_should_not_stall_other_streams() { ); } if buf.as_slice() != ping { - let client_snapshot = log_snapshot(&client_logs); - let server_snapshot = log_snapshot(&server_logs); + let client_snapshot = log_snapshot(client_logs); + let server_snapshot = log_snapshot(server_logs); panic!( "echo payload mismatch\nclient logs:\n{}\nserver logs:\n{}", client_snapshot, server_snapshot @@ -384,8 +382,8 @@ fn single_stream_slow_transfer_should_not_abort() { let _ = blocked.set_nodelay(true); let _ = blocked.set_write_timeout(Some(Duration::from_millis(200))); - if !wait_for_log(&client_logs, "Accepted TCP stream", Duration::from_secs(5)) { - let snapshot = log_snapshot(&client_logs); + if !wait_for_log(client_logs, "Accepted TCP stream", Duration::from_secs(5)) { + let snapshot = log_snapshot(client_logs); panic!("client did not accept blocked stream\n{}", snapshot); } @@ -398,7 +396,7 @@ fn single_stream_slow_transfer_should_not_abort() { assert_eq!(mode, TargetMode::Blackhole, "expected blackhole target"); } None => { - let snapshot = log_snapshot(&server_logs); + let snapshot = log_snapshot(server_logs); panic!("target did not accept blackhole connection\n{}", snapshot); } } @@ -413,6 +411,6 @@ fn single_stream_slow_transfer_should_not_abort() { } } - assert_log_absent(&server_logs, "queued_bytes", Duration::from_secs(1)); - assert_log_absent(&client_logs, "reset event", Duration::from_secs(1)); + assert_log_absent(server_logs, "queued_bytes", Duration::from_secs(1)); + assert_log_absent(client_logs, "reset event", Duration::from_secs(1)); } diff --git a/crates/slipstream-server/tests/idle_gc_e2e.rs b/crates/slipstream-server/tests/idle_gc_e2e.rs index 9cf590cd..5d147bc2 100644 --- a/crates/slipstream-server/tests/idle_gc_e2e.rs +++ b/crates/slipstream-server/tests/idle_gc_e2e.rs @@ -65,6 +65,7 @@ fn idle_gc_closes_connection() { reset_seed_path: None, fallback_addr: None, idle_timeout_seconds: Some(1), + envs: &[], rust_log: "debug", capture_logs: true, }); @@ -82,6 +83,7 @@ fn idle_gc_closes_connection() { domain, cert: Some(&cert), keep_alive_interval: Some(0), + envs: &[], rust_log: "info", capture_logs: true, }); @@ -115,6 +117,7 @@ fn idle_gc_closes_connection() { domain, cert: Some(&cert), keep_alive_interval: Some(0), + envs: &[], rust_log: "info", capture_logs: true, }); diff --git a/crates/slipstream-server/tests/restart_reconnect_e2e.rs b/crates/slipstream-server/tests/restart_reconnect_e2e.rs index 21ad7a76..0e768a0f 100644 --- a/crates/slipstream-server/tests/restart_reconnect_e2e.rs +++ b/crates/slipstream-server/tests/restart_reconnect_e2e.rs @@ -49,6 +49,7 @@ fn restart_reconnects_idle_client() { reset_seed_path: Some(&reset_seed_path), fallback_addr: None, idle_timeout_seconds: None, + envs: &[], rust_log: "info", capture_logs: false, }); @@ -65,6 +66,7 @@ fn restart_reconnects_idle_client() { domain, cert: Some(&cert), keep_alive_interval: Some(0), + envs: &[], rust_log: "info", capture_logs: true, }); @@ -94,6 +96,7 @@ fn restart_reconnects_idle_client() { reset_seed_path: Some(&reset_seed_path), fallback_addr: None, idle_timeout_seconds: None, + envs: &[], rust_log: "info", capture_logs: false, }); diff --git a/crates/slipstream-server/tests/support/mod.rs b/crates/slipstream-server/tests/support/mod.rs index 118e5211..7a01ec77 100644 --- a/crates/slipstream-server/tests/support/mod.rs +++ b/crates/slipstream-server/tests/support/mod.rs @@ -72,6 +72,7 @@ pub struct ServerArgs<'a> { pub reset_seed_path: Option<&'a Path>, pub fallback_addr: Option, pub idle_timeout_seconds: Option, + pub envs: &'a [(&'a str, &'a str)], pub rust_log: &'a str, pub capture_logs: bool, } @@ -83,6 +84,7 @@ pub struct ClientArgs<'a> { pub domain: &'a str, pub cert: Option<&'a Path>, pub keep_alive_interval: Option, + pub envs: &'a [(&'a str, &'a str)], pub rust_log: &'a str, pub capture_logs: bool, } @@ -148,6 +150,9 @@ pub fn spawn_server(args: ServerArgs<'_>) -> (ChildGuard, Option) { cmd.arg("--idle-timeout-seconds") .arg(idle_timeout.to_string()); } + for (key, value) in args.envs { + cmd.env(key, value); + } cmd.arg("--cert") .arg(args.cert) .arg("--key") @@ -163,14 +168,17 @@ pub fn spawn_client(args: ClientArgs<'_>) -> (ChildGuard, Option) { .arg("--resolver") .arg(format!("127.0.0.1:{}", args.dns_port)) .arg("--domain") - .arg(args.domain) - .env("RUST_LOG", args.rust_log); + .arg(args.domain); if let Some(cert) = args.cert { cmd.arg("--cert").arg(cert); } if let Some(interval) = args.keep_alive_interval { cmd.arg("--keep-alive-interval").arg(interval.to_string()); } + for (key, value) in args.envs { + cmd.env(key, value); + } + cmd.env("RUST_LOG", args.rust_log); spawn_process(&mut cmd, args.capture_logs, "slipstream-client") } diff --git a/crates/slipstream-server/tests/udp_fallback_e2e.rs b/crates/slipstream-server/tests/udp_fallback_e2e.rs index 1fd9b26d..cb5f34bc 100644 --- a/crates/slipstream-server/tests/udp_fallback_e2e.rs +++ b/crates/slipstream-server/tests/udp_fallback_e2e.rs @@ -123,6 +123,7 @@ fn udp_fallback_e2e() { reset_seed_path: None, fallback_addr: Some(echo.addr), idle_timeout_seconds: None, + envs: &[], rust_log: "info", capture_logs: false, }); From 2e93da72bc58ef070b5bb2479f174868bd7fa118 Mon Sep 17 00:00:00 2001 From: Mygod Date: Fri, 23 Jan 2026 13:24:29 -0800 Subject: [PATCH 6/9] Refactor to remove dup --- crates/slipstream-client/src/streams.rs | 171 ++++++++--------- crates/slipstream-server/src/streams.rs | 172 +++++++++--------- .../tests/flow_control_e2e.rs | 79 ++++---- 3 files changed, 204 insertions(+), 218 deletions(-) diff --git a/crates/slipstream-client/src/streams.rs b/crates/slipstream-client/src/streams.rs index 57ed68fc..998933c3 100644 --- a/crates/slipstream-client/src/streams.rs +++ b/crates/slipstream-client/src/streams.rs @@ -1,5 +1,6 @@ use slipstream_core::flow_control::{ - conn_reserve_bytes, promote_consumed_offset, reserve_target_offset, stream_queue_max_bytes, + apply_consumed_offset, conn_reserve_bytes, handle_queue_overflow, promote_consumed_offset, + reserve_target_offset, stream_queue_max_bytes, }; use slipstream_core::tcp::{stream_read_limit_chunks, tcp_send_buffer_bytes}; use slipstream_ffi::picoquic::{ @@ -135,20 +136,40 @@ fn close_event_label(event: picoquic_call_back_event_t) -> &'static str { } } +fn log_consume_error(stream_id: u64, context: &'static str, ret: i32, current: u64, target: u64) { + warn!( + "stream {}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", + stream_id, context, ret, current, target + ); +} + +fn consume_stream_data( + cnx: *mut picoquic_cnx_t, + stream_id: u64, + consumed_offset: &mut u64, + target: u64, + context: &'static str, +) -> bool { + apply_consumed_offset( + consumed_offset, + target, + |new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, + |ret, current, target| log_consume_error(stream_id, context, ret, current, target), + ) +} + fn promote_streams(cnx: *mut picoquic_cnx_t, state: &mut ClientState) { for (stream_id, stream) in state.streams.iter_mut() { if stream.discarding { continue; } + let rx_bytes = stream.rx_bytes; promote_consumed_offset( - stream.rx_bytes, + rx_bytes, &mut stream.consumed_offset, |new_offset| unsafe { picoquic_stream_data_consumed(cnx, *stream_id, new_offset) }, |ret, consumed_offset| { - warn!( - "stream {}: stream_data_consumed failed during promote ret={} consumed_offset={}", - stream_id, ret, consumed_offset - ); + log_consume_error(*stream_id, " during promote", ret, consumed_offset, rx_bytes); }, ); } @@ -294,52 +315,45 @@ fn handle_stream_data( if !data.is_empty() { let incoming_len = data.len(); stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); - if stream.consumed_offset < stream.rx_bytes { - stream.consumed_offset = stream.rx_bytes; - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - } - } + let _ = consume_stream_data( + cnx, + stream_id, + &mut stream.consumed_offset, + stream.rx_bytes, + "", + ); } } else if !data.is_empty() { let incoming_len = data.len(); stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); - let projected = stream.queued_bytes.saturating_add(incoming_len); if multi_stream { let max_queue = stream_queue_max_bytes(); - if projected > max_queue { - warn!( - "stream {}: queued_bytes {} + {} exceeds limit {}; stopping", - stream_id, stream.queued_bytes, incoming_len, max_queue - ); + let overflowed = handle_queue_overflow( + stream.queued_bytes, + incoming_len, + max_queue, + stream.rx_bytes, + &mut stream.consumed_offset, + &mut stream.stop_sending_sent, + |queued, incoming, max| { + warn!( + "stream {}: queued_bytes {} + {} exceeds limit {}; stopping", + stream_id, queued, incoming, max + ); + }, + |new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, + || { + let _ = unsafe { + picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) + }; + }, + |ret, current, target| log_consume_error(stream_id, "", ret, current, target), + ); + if overflowed { stream.discarding = true; stream.queued_bytes = 0; let (drain_tx, _drain_rx) = mpsc::unbounded_channel(); stream.write_tx = drain_tx; - if stream.consumed_offset < stream.rx_bytes { - stream.consumed_offset = stream.rx_bytes; - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - } - } - if !stream.stop_sending_sent { - let _ = unsafe { - picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) - }; - stream.stop_sending_sent = true; - } } else if stream .write_tx .send(StreamWrite::Data(data.to_vec())) @@ -351,21 +365,19 @@ fn handle_stream_data( ); reset_stream = true; } else { - stream.queued_bytes = projected; + stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); } - if !stream.discarding && stream.consumed_offset < stream.rx_bytes { - stream.consumed_offset = stream.rx_bytes; - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - reset_stream = true; - } + if !stream.discarding + && !consume_stream_data( + cnx, + stream_id, + &mut stream.consumed_offset, + stream.rx_bytes, + "", + ) + { + reset_stream = true; } } else if stream .write_tx @@ -378,7 +390,7 @@ fn handle_stream_data( ); reset_stream = true; } else { - stream.queued_bytes = projected; + stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); } if !multi_stream && reserve_bytes > 0 && !stream.discarding { @@ -388,18 +400,14 @@ fn handle_stream_data( stream.fin_offset, reserve_bytes, ); - if target_offset > stream.consumed_offset { - stream.consumed_offset = target_offset; - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - reset_stream = true; - } + if !consume_stream_data( + cnx, + stream_id, + &mut stream.consumed_offset, + target_offset, + "", + ) { + reset_stream = true; } } } @@ -635,22 +643,17 @@ pub(crate) fn handle_command( stream.fin_offset, conn_reserve_bytes(), ); - if new_offset > stream.consumed_offset { - stream.consumed_offset = new_offset; - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - let _ = unsafe { - picoquic_reset_stream(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) - }; - state.streams.remove(&stream_id); - return; - } + if !consume_stream_data( + cnx, + stream_id, + &mut stream.consumed_offset, + new_offset, + "", + ) { + let _ = + unsafe { picoquic_reset_stream(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) }; + state.streams.remove(&stream_id); + return; } } if stream.fin_enqueued && stream.queued_bytes == 0 { diff --git a/crates/slipstream-server/src/streams.rs b/crates/slipstream-server/src/streams.rs index f26a0818..29189778 100644 --- a/crates/slipstream-server/src/streams.rs +++ b/crates/slipstream-server/src/streams.rs @@ -1,7 +1,8 @@ use crate::server::{Command, StreamKey, StreamWrite}; use crate::target::spawn_target_connector; use slipstream_core::flow_control::{ - conn_reserve_bytes, promote_consumed_offset, reserve_target_offset, stream_queue_max_bytes, + apply_consumed_offset, conn_reserve_bytes, handle_queue_overflow, promote_consumed_offset, + reserve_target_offset, stream_queue_max_bytes, }; use slipstream_ffi::picoquic::{ picoquic_call_back_event_t, picoquic_close, picoquic_close_immediate, picoquic_cnx_t, @@ -108,6 +109,28 @@ struct ServerStream { stop_sending_sent: bool, } +fn log_consume_error(stream_id: u64, context: &'static str, ret: i32, current: u64, target: u64) { + warn!( + "stream {:?}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", + stream_id, context, ret, current, target + ); +} + +fn consume_stream_data( + cnx: *mut picoquic_cnx_t, + stream_id: u64, + consumed_offset: &mut u64, + target: u64, + context: &'static str, +) -> bool { + apply_consumed_offset( + consumed_offset, + target, + |new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, + |ret, current, target| log_consume_error(stream_id, context, ret, current, target), + ) +} + fn mark_multi_stream(state: &mut ServerState, cnx_id: usize) -> bool { if state.multi_streams.contains(&cnx_id) { return false; @@ -130,15 +153,13 @@ fn promote_connection_streams(cnx: *mut picoquic_cnx_t, state: &mut ServerState, if stream.discarding { continue; } + let rx_bytes = stream.rx_bytes; promote_consumed_offset( - stream.rx_bytes, + rx_bytes, &mut stream.consumed_offset, |new_offset| unsafe { picoquic_stream_data_consumed(cnx, key.stream_id, new_offset) }, |ret, consumed_offset| { - warn!( - "stream {:?}: stream_data_consumed failed during promote ret={} consumed_offset={}", - key.stream_id, ret, consumed_offset - ); + log_consume_error(key.stream_id, " during promote", ret, consumed_offset, rx_bytes); }, ); } @@ -402,31 +423,42 @@ fn handle_stream_data( if !data.is_empty() { let incoming_len = data.len(); stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); - if stream.consumed_offset < stream.rx_bytes { - stream.consumed_offset = stream.rx_bytes; - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - } - } + let _ = consume_stream_data( + cnx, + stream_id, + &mut stream.consumed_offset, + stream.rx_bytes, + "", + ); } } else if !data.is_empty() { let incoming_len = data.len(); stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); - let projected = stream.queued_bytes.saturating_add(incoming_len); if multi_stream { let max_queue = stream_queue_max_bytes(); - if projected > max_queue { - warn!( - "stream {:?}: queued_bytes {} + {} exceeds limit {}; stopping", - stream_id, stream.queued_bytes, incoming_len, max_queue - ); + let overflowed = handle_queue_overflow( + stream.queued_bytes, + incoming_len, + max_queue, + stream.rx_bytes, + &mut stream.consumed_offset, + &mut stream.stop_sending_sent, + |queued, incoming, max| { + warn!( + "stream {:?}: queued_bytes {} + {} exceeds limit {}; stopping", + stream_id, queued, incoming, max + ); + }, + |new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, + || { + let _ = unsafe { + picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) + }; + }, + |ret, current, target| log_consume_error(stream_id, "", ret, current, target), + ); + if overflowed { stream.discarding = true; stream.pending_data.clear(); stream.queued_bytes = 0; @@ -439,57 +471,37 @@ fn handle_stream_data( stream.target_fin_pending = false; stream.close_after_flush = false; let _ = stream.shutdown_tx.send(true); - if stream.consumed_offset < stream.rx_bytes { - stream.consumed_offset = stream.rx_bytes; - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - } - } - if !stream.stop_sending_sent { - let _ = unsafe { - picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) - }; - stream.stop_sending_sent = true; - } } else if let Some(write_tx) = stream.write_tx.as_ref() { if write_tx.send(StreamWrite::Data(data.to_vec())).is_err() { reset_stream = true; } else { - stream.queued_bytes = projected; + stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); } } else { stream.pending_data.push_back(data.to_vec()); - stream.queued_bytes = projected; + stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); } - if !stream.discarding && stream.consumed_offset < stream.rx_bytes { - stream.consumed_offset = stream.rx_bytes; - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - reset_stream = true; - } + if !stream.discarding + && !consume_stream_data( + cnx, + stream_id, + &mut stream.consumed_offset, + stream.rx_bytes, + "", + ) + { + reset_stream = true; } } else if let Some(write_tx) = stream.write_tx.as_ref() { if write_tx.send(StreamWrite::Data(data.to_vec())).is_err() { reset_stream = true; } else { - stream.queued_bytes = projected; + stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); } } else { stream.pending_data.push_back(data.to_vec()); - stream.queued_bytes = projected; + stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); } if !multi_stream && reserve_bytes > 0 && !stream.discarding { @@ -499,18 +511,14 @@ fn handle_stream_data( stream.fin_offset, reserve_bytes, ); - if target_offset > stream.consumed_offset { - stream.consumed_offset = target_offset; - let ret = unsafe { - picoquic_stream_data_consumed(cnx, stream_id, stream.consumed_offset) - }; - if ret < 0 { - warn!( - "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - reset_stream = true; - } + if !consume_stream_data( + cnx, + stream_id, + &mut stream.consumed_offset, + target_offset, + "", + ) { + reset_stream = true; } } } @@ -787,22 +795,14 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { stream.fin_offset, conn_reserve_bytes(), ); - if new_offset > stream.consumed_offset { - stream.consumed_offset = new_offset; - let ret = unsafe { - picoquic_stream_data_consumed( - cnx_id as *mut picoquic_cnx_t, - stream_id, - stream.consumed_offset, - ) - }; - if ret < 0 { - warn!( - "stream {:?}: stream_data_consumed failed ret={} consumed_offset={}", - stream_id, ret, stream.consumed_offset - ); - reset_stream = true; - } + if !consume_stream_data( + cnx_id as *mut picoquic_cnx_t, + stream_id, + &mut stream.consumed_offset, + new_offset, + "", + ) { + reset_stream = true; } } } diff --git a/crates/slipstream-server/tests/flow_control_e2e.rs b/crates/slipstream-server/tests/flow_control_e2e.rs index 939b0642..423ca446 100644 --- a/crates/slipstream-server/tests/flow_control_e2e.rs +++ b/crates/slipstream-server/tests/flow_control_e2e.rs @@ -254,16 +254,12 @@ fn setup_flow_control(envs: &[(&str, &str)]) -> Option { }) } -#[test] -fn blocked_stream_should_not_stall_other_streams() { - let Some(harness) = setup_flow_control(&[ - ("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES", "65536"), - ("SLIPSTREAM_CONN_RESERVE_BYTES", "65536"), - ("SLIPSTREAM_STREAM_WRITE_BUFFER_BYTES", "8388608"), - ]) else { - return; - }; - +fn open_blackhole_stream( + harness: &FlowControlHarness, + expected_index: usize, + send_duration: Duration, + payload_len: usize, +) -> TcpStream { let server_logs = &harness.server_logs; let client_logs = &harness.client_logs; let target = &harness.target; @@ -283,7 +279,7 @@ fn blocked_stream_should_not_stall_other_streams() { match target.recv_event(Duration::from_secs(5)) { Some(TargetEvent::Accepted { index, mode }) => { - assert_eq!(index, 0, "expected first target connection to be index 0"); + assert_eq!(index, expected_index, "unexpected target index"); assert_eq!(mode, TargetMode::Blackhole, "expected blackhole target"); } None => { @@ -292,8 +288,8 @@ fn blocked_stream_should_not_stall_other_streams() { } } - let send_deadline = Instant::now() + Duration::from_secs(3); - let payload = vec![0u8; 32 * 1024]; + let send_deadline = Instant::now() + send_duration; + let payload = vec![0u8; payload_len]; while Instant::now() < send_deadline { match blocked.write(&payload) { Ok(0) => break, @@ -302,6 +298,26 @@ fn blocked_stream_should_not_stall_other_streams() { } } + blocked +} + +#[test] +fn blocked_stream_should_not_stall_other_streams() { + let Some(harness) = setup_flow_control(&[ + ("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES", "65536"), + ("SLIPSTREAM_CONN_RESERVE_BYTES", "65536"), + ("SLIPSTREAM_STREAM_WRITE_BUFFER_BYTES", "8388608"), + ]) else { + return; + }; + + let server_logs = &harness.server_logs; + let client_logs = &harness.client_logs; + let target = &harness.target; + let client_addr = harness.client_addr; + let _blocked = + open_blackhole_stream(&harness, 0, Duration::from_secs(3), 32 * 1024); + let mut echo = TcpStream::connect_timeout(&client_addr, Duration::from_secs(2)) .expect("connect echo stream"); let _ = echo.set_nodelay(true); @@ -375,41 +391,8 @@ fn single_stream_slow_transfer_should_not_abort() { let server_logs = &harness.server_logs; let client_logs = &harness.client_logs; - let target = &harness.target; - let client_addr = harness.client_addr; - let mut blocked = TcpStream::connect_timeout(&client_addr, Duration::from_secs(2)) - .expect("connect blocked stream"); - let _ = blocked.set_nodelay(true); - let _ = blocked.set_write_timeout(Some(Duration::from_millis(200))); - - if !wait_for_log(client_logs, "Accepted TCP stream", Duration::from_secs(5)) { - let snapshot = log_snapshot(client_logs); - panic!("client did not accept blocked stream\n{}", snapshot); - } - - let warmup = vec![0u8; 1024]; - let _ = blocked.write_all(&warmup); - - match target.recv_event(Duration::from_secs(5)) { - Some(TargetEvent::Accepted { index, mode }) => { - assert_eq!(index, 0, "expected first target connection to be index 0"); - assert_eq!(mode, TargetMode::Blackhole, "expected blackhole target"); - } - None => { - let snapshot = log_snapshot(server_logs); - panic!("target did not accept blackhole connection\n{}", snapshot); - } - } - - let send_deadline = Instant::now() + Duration::from_secs(2); - let payload = vec![0u8; 64 * 1024]; - while Instant::now() < send_deadline { - match blocked.write(&payload) { - Ok(0) => break, - Ok(_) => continue, - Err(_) => break, - } - } + let _blocked = + open_blackhole_stream(&harness, 0, Duration::from_secs(2), 64 * 1024); assert_log_absent(server_logs, "queued_bytes", Duration::from_secs(1)); assert_log_absent(client_logs, "reset event", Duration::from_secs(1)); From 8b123365b54eef195d3393f371308da45d6e5dcb Mon Sep 17 00:00:00 2001 From: Mygod Date: Fri, 23 Jan 2026 13:53:08 -0800 Subject: [PATCH 7/9] Refactor further --- crates/slipstream-client/src/streams.rs | 270 +++++++++--------- crates/slipstream-server/src/streams.rs | 266 +++++++++-------- .../tests/flow_control_e2e.rs | 6 +- 3 files changed, 260 insertions(+), 282 deletions(-) diff --git a/crates/slipstream-client/src/streams.rs b/crates/slipstream-client/src/streams.rs index 998933c3..a44a74e8 100644 --- a/crates/slipstream-client/src/streams.rs +++ b/crates/slipstream-client/src/streams.rs @@ -1,6 +1,7 @@ use slipstream_core::flow_control::{ - apply_consumed_offset, conn_reserve_bytes, handle_queue_overflow, promote_consumed_offset, - reserve_target_offset, stream_queue_max_bytes, + conn_reserve_bytes, consume_stream_data, handle_stream_receive, promote_streams, + reserve_target_offset, stream_queue_max_bytes, FlowControlStream, PromoteEntry, + StreamReceiveConfig, StreamReceiveOps, }; use slipstream_core::tcp::{stream_read_limit_chunks, tcp_send_buffer_bytes}; use slipstream_ffi::picoquic::{ @@ -108,6 +109,52 @@ struct ClientStream { stop_sending_sent: bool, } +impl FlowControlStream for ClientStream { + fn queued_bytes(&self) -> usize { + self.queued_bytes + } + + fn set_queued_bytes(&mut self, value: usize) { + self.queued_bytes = value; + } + + fn rx_bytes(&self) -> u64 { + self.rx_bytes + } + + fn set_rx_bytes(&mut self, value: u64) { + self.rx_bytes = value; + } + + fn consumed_offset(&self) -> u64 { + self.consumed_offset + } + + fn set_consumed_offset(&mut self, value: u64) { + self.consumed_offset = value; + } + + fn fin_offset(&self) -> Option { + self.fin_offset + } + + fn discarding(&self) -> bool { + self.discarding + } + + fn set_discarding(&mut self, value: bool) { + self.discarding = value; + } + + fn stop_sending_sent(&self) -> bool { + self.stop_sending_sent + } + + fn set_stop_sending_sent(&mut self, value: bool) { + self.stop_sending_sent = value; + } +} + enum StreamWrite { Data(Vec), Fin, @@ -136,45 +183,6 @@ fn close_event_label(event: picoquic_call_back_event_t) -> &'static str { } } -fn log_consume_error(stream_id: u64, context: &'static str, ret: i32, current: u64, target: u64) { - warn!( - "stream {}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", - stream_id, context, ret, current, target - ); -} - -fn consume_stream_data( - cnx: *mut picoquic_cnx_t, - stream_id: u64, - consumed_offset: &mut u64, - target: u64, - context: &'static str, -) -> bool { - apply_consumed_offset( - consumed_offset, - target, - |new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, - |ret, current, target| log_consume_error(stream_id, context, ret, current, target), - ) -} - -fn promote_streams(cnx: *mut picoquic_cnx_t, state: &mut ClientState) { - for (stream_id, stream) in state.streams.iter_mut() { - if stream.discarding { - continue; - } - let rx_bytes = stream.rx_bytes; - promote_consumed_offset( - rx_bytes, - &mut stream.consumed_offset, - |new_offset| unsafe { picoquic_stream_data_consumed(cnx, *stream_id, new_offset) }, - |ret, consumed_offset| { - log_consume_error(*stream_id, " during promote", ret, consumed_offset, rx_bytes); - }, - ); - } -} - pub(crate) unsafe extern "C" fn client_callback( cnx: *mut picoquic_cnx_t, stream_id: u64, @@ -311,105 +319,61 @@ fn handle_stream_data( return; }; - if stream.discarding { - if !data.is_empty() { - let incoming_len = data.len(); - stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); - let _ = consume_stream_data( - cnx, - stream_id, - &mut stream.consumed_offset, - stream.rx_bytes, - "", - ); - } - } else if !data.is_empty() { - let incoming_len = data.len(); - stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); - if multi_stream { - let max_queue = stream_queue_max_bytes(); - let overflowed = handle_queue_overflow( - stream.queued_bytes, - incoming_len, - max_queue, - stream.rx_bytes, - &mut stream.consumed_offset, - &mut stream.stop_sending_sent, - |queued, incoming, max| { + let max_queue = if multi_stream { + stream_queue_max_bytes() + } else { + 0 + }; + if handle_stream_receive( + stream, + data.len(), + StreamReceiveConfig { + multi_stream, + reserve_bytes, + max_queue, + }, + StreamReceiveOps { + enqueue: |stream: &mut ClientStream| { + if stream + .write_tx + .send(StreamWrite::Data(data.to_vec())) + .is_err() + { warn!( - "stream {}: queued_bytes {} + {} exceeds limit {}; stopping", - stream_id, queued, incoming, max + "stream {}: tcp write channel closed queued={} rx_bytes={} tx_bytes={}", + stream_id, stream.queued_bytes, stream.rx_bytes, stream.tx_bytes ); - }, - |new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, - || { - let _ = unsafe { - picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) - }; - }, - |ret, current, target| log_consume_error(stream_id, "", ret, current, target), - ); - if overflowed { - stream.discarding = true; - stream.queued_bytes = 0; + Err(()) + } else { + Ok(()) + } + }, + on_overflow: |stream: &mut ClientStream| { let (drain_tx, _drain_rx) = mpsc::unbounded_channel(); stream.write_tx = drain_tx; - } else if stream - .write_tx - .send(StreamWrite::Data(data.to_vec())) - .is_err() - { + }, + consume: |new_offset| unsafe { + picoquic_stream_data_consumed(cnx, stream_id, new_offset) + }, + stop_sending: || { + let _ = + unsafe { picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) }; + }, + log_overflow: |queued, incoming, max| { warn!( - "stream {}: tcp write channel closed queued={} rx_bytes={} tx_bytes={}", - stream_id, stream.queued_bytes, stream.rx_bytes, stream.tx_bytes + "stream {}: queued_bytes {} + {} exceeds limit {}; stopping", + stream_id, queued, incoming, max ); - reset_stream = true; - } else { - stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); - } - - if !stream.discarding - && !consume_stream_data( - cnx, - stream_id, - &mut stream.consumed_offset, - stream.rx_bytes, - "", - ) - { - reset_stream = true; - } - } else if stream - .write_tx - .send(StreamWrite::Data(data.to_vec())) - .is_err() - { - warn!( - "stream {}: tcp write channel closed queued={} rx_bytes={} tx_bytes={}", - stream_id, stream.queued_bytes, stream.rx_bytes, stream.tx_bytes - ); - reset_stream = true; - } else { - stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); - } - - if !multi_stream && reserve_bytes > 0 && !stream.discarding { - let target_offset = reserve_target_offset( - stream.rx_bytes, - stream.queued_bytes, - stream.fin_offset, - reserve_bytes, - ); - if !consume_stream_data( - cnx, - stream_id, - &mut stream.consumed_offset, - target_offset, - "", - ) { - reset_stream = true; - } - } + }, + on_consume_error: |ret, current, target| { + warn!( + "stream {}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", + stream_id, "", ret, current, target + ); + }, + }, + ) { + reset_stream = true; } if fin { @@ -559,7 +523,26 @@ pub(crate) fn handle_command( ); if !state.multi_stream_mode && state.streams.len() > 1 { state.multi_stream_mode = true; - promote_streams(cnx, state); + promote_streams( + state + .streams + .iter_mut() + .map(|(stream_id, stream)| PromoteEntry { + stream_id: *stream_id, + rx_bytes: stream.rx_bytes, + consumed_offset: &mut stream.consumed_offset, + discarding: stream.discarding, + }), + |stream_id, new_offset| unsafe { + picoquic_stream_data_consumed(cnx, stream_id, new_offset) + }, + |stream_id, ret, consumed_offset, rx_bytes| { + warn!( + "stream {}: stream_data_consumed failed during promote ret={} consumed_offset={} target={}", + stream_id, ret, consumed_offset, rx_bytes + ); + }, + ); } let _ = unsafe { picoquic_mark_active_stream(cnx, stream_id, 1, std::ptr::null_mut()) }; if state.debug_streams { @@ -644,14 +627,21 @@ pub(crate) fn handle_command( conn_reserve_bytes(), ); if !consume_stream_data( - cnx, - stream_id, &mut stream.consumed_offset, new_offset, - "", + |new_offset| unsafe { + picoquic_stream_data_consumed(cnx, stream_id, new_offset) + }, + |ret, current, target| { + warn!( + "stream {}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", + stream_id, "", ret, current, target + ); + }, ) { - let _ = - unsafe { picoquic_reset_stream(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) }; + let _ = unsafe { + picoquic_reset_stream(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) + }; state.streams.remove(&stream_id); return; } diff --git a/crates/slipstream-server/src/streams.rs b/crates/slipstream-server/src/streams.rs index 29189778..9b692fb3 100644 --- a/crates/slipstream-server/src/streams.rs +++ b/crates/slipstream-server/src/streams.rs @@ -1,8 +1,9 @@ use crate::server::{Command, StreamKey, StreamWrite}; use crate::target::spawn_target_connector; use slipstream_core::flow_control::{ - apply_consumed_offset, conn_reserve_bytes, handle_queue_overflow, promote_consumed_offset, - reserve_target_offset, stream_queue_max_bytes, + conn_reserve_bytes, consume_stream_data, handle_stream_receive, promote_streams, + reserve_target_offset, stream_queue_max_bytes, FlowControlStream, PromoteEntry, + StreamReceiveConfig, StreamReceiveOps, }; use slipstream_ffi::picoquic::{ picoquic_call_back_event_t, picoquic_close, picoquic_close_immediate, picoquic_cnx_t, @@ -109,26 +110,50 @@ struct ServerStream { stop_sending_sent: bool, } -fn log_consume_error(stream_id: u64, context: &'static str, ret: i32, current: u64, target: u64) { - warn!( - "stream {:?}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", - stream_id, context, ret, current, target - ); -} +impl FlowControlStream for ServerStream { + fn queued_bytes(&self) -> usize { + self.queued_bytes + } -fn consume_stream_data( - cnx: *mut picoquic_cnx_t, - stream_id: u64, - consumed_offset: &mut u64, - target: u64, - context: &'static str, -) -> bool { - apply_consumed_offset( - consumed_offset, - target, - |new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, - |ret, current, target| log_consume_error(stream_id, context, ret, current, target), - ) + fn set_queued_bytes(&mut self, value: usize) { + self.queued_bytes = value; + } + + fn rx_bytes(&self) -> u64 { + self.rx_bytes + } + + fn set_rx_bytes(&mut self, value: u64) { + self.rx_bytes = value; + } + + fn consumed_offset(&self) -> u64 { + self.consumed_offset + } + + fn set_consumed_offset(&mut self, value: u64) { + self.consumed_offset = value; + } + + fn fin_offset(&self) -> Option { + self.fin_offset + } + + fn discarding(&self) -> bool { + self.discarding + } + + fn set_discarding(&mut self, value: bool) { + self.discarding = value; + } + + fn stop_sending_sent(&self) -> bool { + self.stop_sending_sent + } + + fn set_stop_sending_sent(&mut self, value: bool) { + self.stop_sending_sent = value; + } } fn mark_multi_stream(state: &mut ServerState, cnx_id: usize) -> bool { @@ -144,27 +169,6 @@ fn mark_multi_stream(state: &mut ServerState, cnx_id: usize) -> bool { } } -fn promote_connection_streams(cnx: *mut picoquic_cnx_t, state: &mut ServerState, cnx_id: usize) { - for (key, stream) in state - .streams - .iter_mut() - .filter(|(key, _)| key.cnx == cnx_id) - { - if stream.discarding { - continue; - } - let rx_bytes = stream.rx_bytes; - promote_consumed_offset( - rx_bytes, - &mut stream.consumed_offset, - |new_offset| unsafe { picoquic_stream_data_consumed(cnx, key.stream_id, new_offset) }, - |ret, consumed_offset| { - log_consume_error(key.stream_id, " during promote", ret, consumed_offset, rx_bytes); - }, - ); - } -} - pub(crate) unsafe extern "C" fn server_callback( cnx: *mut picoquic_cnx_t, stream_id: u64, @@ -404,7 +408,27 @@ fn handle_stream_data( } if mark_multi_stream(state, key.cnx) { - promote_connection_streams(cnx, state, key.cnx); + promote_streams( + state + .streams + .iter_mut() + .filter(|(entry_key, _)| entry_key.cnx == key.cnx) + .map(|(entry_key, stream)| PromoteEntry { + stream_id: entry_key.stream_id, + rx_bytes: stream.rx_bytes, + consumed_offset: &mut stream.consumed_offset, + discarding: stream.discarding, + }), + |stream_id, new_offset| unsafe { + picoquic_stream_data_consumed(cnx, stream_id, new_offset) + }, + |stream_id, ret, consumed_offset, rx_bytes| { + warn!( + "stream {:?}: stream_data_consumed failed during promote ret={} consumed_offset={} target={}", + stream_id, ret, consumed_offset, rx_bytes + ); + }, + ); } let multi_stream = state.multi_streams.contains(&key.cnx); let reserve_bytes = if multi_stream { @@ -419,49 +443,32 @@ fn handle_stream_data( None => return, }; - if stream.discarding { - if !data.is_empty() { - let incoming_len = data.len(); - stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); - let _ = consume_stream_data( - cnx, - stream_id, - &mut stream.consumed_offset, - stream.rx_bytes, - "", - ); - } - } else if !data.is_empty() { - let incoming_len = data.len(); - stream.rx_bytes = stream.rx_bytes.saturating_add(incoming_len as u64); - - if multi_stream { - let max_queue = stream_queue_max_bytes(); - let overflowed = handle_queue_overflow( - stream.queued_bytes, - incoming_len, - max_queue, - stream.rx_bytes, - &mut stream.consumed_offset, - &mut stream.stop_sending_sent, - |queued, incoming, max| { - warn!( - "stream {:?}: queued_bytes {} + {} exceeds limit {}; stopping", - stream_id, queued, incoming, max - ); - }, - |new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, - || { - let _ = unsafe { - picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) - }; - }, - |ret, current, target| log_consume_error(stream_id, "", ret, current, target), - ); - if overflowed { - stream.discarding = true; + let max_queue = if multi_stream { + stream_queue_max_bytes() + } else { + 0 + }; + if handle_stream_receive( + stream, + data.len(), + StreamReceiveConfig { + multi_stream, + reserve_bytes, + max_queue, + }, + StreamReceiveOps { + enqueue: |stream: &mut ServerStream| { + if let Some(write_tx) = stream.write_tx.as_ref() { + if write_tx.send(StreamWrite::Data(data.to_vec())).is_err() { + return Err(()); + } + } else { + stream.pending_data.push_back(data.to_vec()); + } + Ok(()) + }, + on_overflow: |stream: &mut ServerStream| { stream.pending_data.clear(); - stream.queued_bytes = 0; stream.pending_fin = false; stream.fin_enqueued = false; stream.data_rx = None; @@ -471,56 +478,29 @@ fn handle_stream_data( stream.target_fin_pending = false; stream.close_after_flush = false; let _ = stream.shutdown_tx.send(true); - } else if let Some(write_tx) = stream.write_tx.as_ref() { - if write_tx.send(StreamWrite::Data(data.to_vec())).is_err() { - reset_stream = true; - } else { - stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); - } - } else { - stream.pending_data.push_back(data.to_vec()); - stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); - } - - if !stream.discarding - && !consume_stream_data( - cnx, - stream_id, - &mut stream.consumed_offset, - stream.rx_bytes, - "", - ) - { - reset_stream = true; - } - } else if let Some(write_tx) = stream.write_tx.as_ref() { - if write_tx.send(StreamWrite::Data(data.to_vec())).is_err() { - reset_stream = true; - } else { - stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); - } - } else { - stream.pending_data.push_back(data.to_vec()); - stream.queued_bytes = stream.queued_bytes.saturating_add(incoming_len); - } - - if !multi_stream && reserve_bytes > 0 && !stream.discarding { - let target_offset = reserve_target_offset( - stream.rx_bytes, - stream.queued_bytes, - stream.fin_offset, - reserve_bytes, - ); - if !consume_stream_data( - cnx, - stream_id, - &mut stream.consumed_offset, - target_offset, - "", - ) { - reset_stream = true; - } - } + }, + consume: |new_offset| unsafe { + picoquic_stream_data_consumed(cnx, stream_id, new_offset) + }, + stop_sending: || { + let _ = + unsafe { picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) }; + }, + log_overflow: |queued, incoming, max| { + warn!( + "stream {:?}: queued_bytes {} + {} exceeds limit {}; stopping", + stream_id, queued, incoming, max + ); + }, + on_consume_error: |ret, current, target| { + warn!( + "stream {:?}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", + stream_id, "", ret, current, target + ); + }, + }, + ) { + reset_stream = true; } if fin { @@ -796,11 +776,21 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { conn_reserve_bytes(), ); if !consume_stream_data( - cnx_id as *mut picoquic_cnx_t, - stream_id, &mut stream.consumed_offset, new_offset, - "", + |new_offset| unsafe { + picoquic_stream_data_consumed( + cnx_id as *mut picoquic_cnx_t, + stream_id, + new_offset, + ) + }, + |ret, current, target| { + warn!( + "stream {:?}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", + stream_id, "", ret, current, target + ); + }, ) { reset_stream = true; } diff --git a/crates/slipstream-server/tests/flow_control_e2e.rs b/crates/slipstream-server/tests/flow_control_e2e.rs index 423ca446..7f7f012b 100644 --- a/crates/slipstream-server/tests/flow_control_e2e.rs +++ b/crates/slipstream-server/tests/flow_control_e2e.rs @@ -315,8 +315,7 @@ fn blocked_stream_should_not_stall_other_streams() { let client_logs = &harness.client_logs; let target = &harness.target; let client_addr = harness.client_addr; - let _blocked = - open_blackhole_stream(&harness, 0, Duration::from_secs(3), 32 * 1024); + let _blocked = open_blackhole_stream(&harness, 0, Duration::from_secs(3), 32 * 1024); let mut echo = TcpStream::connect_timeout(&client_addr, Duration::from_secs(2)) .expect("connect echo stream"); @@ -391,8 +390,7 @@ fn single_stream_slow_transfer_should_not_abort() { let server_logs = &harness.server_logs; let client_logs = &harness.client_logs; - let _blocked = - open_blackhole_stream(&harness, 0, Duration::from_secs(2), 64 * 1024); + let _blocked = open_blackhole_stream(&harness, 0, Duration::from_secs(2), 64 * 1024); assert_log_absent(server_logs, "queued_bytes", Duration::from_secs(1)); assert_log_absent(client_logs, "reset event", Duration::from_secs(1)); From 0d353a75cbd4a045f0a95857d6ce1a52ccdf1d23 Mon Sep 17 00:00:00 2001 From: Mygod Date: Fri, 23 Jan 2026 13:53:52 -0800 Subject: [PATCH 8/9] Add missing file --- crates/slipstream-core/src/flow_control.rs | 303 +++++++++++++++++++++ 1 file changed, 303 insertions(+) create mode 100644 crates/slipstream-core/src/flow_control.rs diff --git a/crates/slipstream-core/src/flow_control.rs b/crates/slipstream-core/src/flow_control.rs new file mode 100644 index 00000000..18c7f7ed --- /dev/null +++ b/crates/slipstream-core/src/flow_control.rs @@ -0,0 +1,303 @@ +use std::sync::OnceLock; + +const DEFAULT_STREAM_QUEUE_MAX_BYTES: usize = 2 * 1024 * 1024; +const DEFAULT_CONN_RESERVE_BYTES: usize = 64 * 1024; + +pub trait FlowControlStream { + fn queued_bytes(&self) -> usize; + fn set_queued_bytes(&mut self, value: usize); + fn rx_bytes(&self) -> u64; + fn set_rx_bytes(&mut self, value: u64); + fn consumed_offset(&self) -> u64; + fn set_consumed_offset(&mut self, value: u64); + fn fin_offset(&self) -> Option; + fn discarding(&self) -> bool; + fn set_discarding(&mut self, value: bool); + fn stop_sending_sent(&self) -> bool; + fn set_stop_sending_sent(&mut self, value: bool); +} + +pub struct QueueOverflowOps { + pub log_overflow: Log, + pub consume: Consume, + pub stop_sending: Stop, + pub on_consume_error: Err, +} + +pub struct StreamReceiveConfig { + pub multi_stream: bool, + pub reserve_bytes: usize, + pub max_queue: usize, +} + +pub struct StreamReceiveOps { + pub enqueue: Enqueue, + pub on_overflow: Overflow, + pub consume: Consume, + pub stop_sending: Stop, + pub log_overflow: Log, + pub on_consume_error: Err, +} + +pub fn stream_queue_max_bytes() -> usize { + static MAX_BYTES: OnceLock = OnceLock::new(); + *MAX_BYTES.get_or_init(|| { + std::env::var("SLIPSTREAM_STREAM_QUEUE_MAX_BYTES") + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_STREAM_QUEUE_MAX_BYTES) + }) +} + +pub fn conn_reserve_bytes() -> usize { + static RESERVE_BYTES: OnceLock = OnceLock::new(); + *RESERVE_BYTES.get_or_init(|| { + std::env::var("SLIPSTREAM_CONN_RESERVE_BYTES") + .ok() + .and_then(|value| value.parse::().ok()) + .unwrap_or(DEFAULT_CONN_RESERVE_BYTES) + }) +} + +pub fn reserve_target_offset( + rx_bytes: u64, + queued_bytes: usize, + fin_offset: Option, + reserve_bytes: usize, +) -> u64 { + let drained = rx_bytes.saturating_sub(queued_bytes as u64); + let mut target = if reserve_bytes > 0 { + drained.saturating_add(reserve_bytes as u64).min(rx_bytes) + } else { + drained + }; + if let Some(fin) = fin_offset { + if target > fin { + target = fin; + } + } + target +} + +pub fn apply_consumed_offset( + consumed_offset: &mut u64, + target: u64, + mut consume_fn: F, + mut on_error: G, +) -> bool +where + F: FnMut(u64) -> i32, + G: FnMut(i32, u64, u64), +{ + if target <= *consumed_offset { + return true; + } + let ret = consume_fn(target); + if ret < 0 { + on_error(ret, *consumed_offset, target); + return false; + } + *consumed_offset = target; + true +} + +pub fn consume_stream_data( + consumed_offset: &mut u64, + target: u64, + consume_fn: F, + on_error: G, +) -> bool +where + F: FnMut(u64) -> i32, + G: FnMut(i32, u64, u64), +{ + apply_consumed_offset(consumed_offset, target, consume_fn, on_error) +} + +#[allow(clippy::too_many_arguments)] +pub fn handle_queue_overflow( + queued_bytes: usize, + incoming_len: usize, + max_queue: usize, + rx_bytes: u64, + consumed_offset: &mut u64, + stop_sending_sent: &mut bool, + mut ops: QueueOverflowOps, +) -> bool +where + Log: FnMut(usize, usize, usize), + Consume: FnMut(u64) -> i32, + Stop: FnMut(), + Err: FnMut(i32, u64, u64), +{ + let projected = queued_bytes.saturating_add(incoming_len); + if projected <= max_queue { + return false; + } + (ops.log_overflow)(queued_bytes, incoming_len, max_queue); + let _ = apply_consumed_offset( + consumed_offset, + rx_bytes, + &mut ops.consume, + &mut ops.on_consume_error, + ); + if !*stop_sending_sent { + (ops.stop_sending)(); + *stop_sending_sent = true; + } + true +} + +pub struct PromoteEntry<'a> { + pub stream_id: u64, + pub rx_bytes: u64, + pub consumed_offset: &'a mut u64, + pub discarding: bool, +} + +pub fn promote_consumed_offset( + rx_bytes: u64, + consumed_offset: &mut u64, + mut consume_fn: F, + mut on_error: G, +) where + F: FnMut(u64) -> i32, + G: FnMut(i32, u64), +{ + if *consumed_offset >= rx_bytes { + return; + } + let _ = apply_consumed_offset( + consumed_offset, + rx_bytes, + &mut consume_fn, + |ret, current, _| { + on_error(ret, current); + }, + ); +} + +pub fn promote_streams<'a, I, Consume, Log>(entries: I, mut consume_fn: Consume, mut on_error: Log) +where + I: IntoIterator>, + Consume: FnMut(u64, u64) -> i32, + Log: FnMut(u64, i32, u64, u64), +{ + for entry in entries { + if entry.discarding { + continue; + } + let stream_id = entry.stream_id; + let rx_bytes = entry.rx_bytes; + promote_consumed_offset( + rx_bytes, + entry.consumed_offset, + |new_offset| consume_fn(stream_id, new_offset), + |ret, consumed_offset| on_error(stream_id, ret, consumed_offset, rx_bytes), + ); + } +} + +#[allow(clippy::too_many_arguments)] +pub fn handle_stream_receive( + stream: &mut S, + incoming_len: usize, + config: StreamReceiveConfig, + mut ops: StreamReceiveOps, +) -> bool +where + S: FlowControlStream, + Enqueue: FnMut(&mut S) -> Result<(), ()>, + Overflow: FnMut(&mut S), + Consume: FnMut(u64) -> i32, + Stop: FnMut(), + Log: FnMut(usize, usize, usize), + Err: FnMut(i32, u64, u64), +{ + if incoming_len == 0 { + return false; + } + + let mut queued_bytes = stream.queued_bytes(); + let mut rx_bytes = stream.rx_bytes(); + let mut consumed_offset = stream.consumed_offset(); + let fin_offset = stream.fin_offset(); + let mut discarding = stream.discarding(); + let mut stop_sending_sent = stream.stop_sending_sent(); + let mut reset_stream = false; + + rx_bytes = rx_bytes.saturating_add(incoming_len as u64); + stream.set_rx_bytes(rx_bytes); + + if discarding { + let _ = consume_stream_data( + &mut consumed_offset, + rx_bytes, + &mut ops.consume, + &mut ops.on_consume_error, + ); + } else if config.multi_stream { + let overflowed = handle_queue_overflow( + queued_bytes, + incoming_len, + config.max_queue, + rx_bytes, + &mut consumed_offset, + &mut stop_sending_sent, + QueueOverflowOps { + log_overflow: &mut ops.log_overflow, + consume: &mut ops.consume, + stop_sending: &mut ops.stop_sending, + on_consume_error: &mut ops.on_consume_error, + }, + ); + if overflowed { + discarding = true; + queued_bytes = 0; + stream.set_discarding(true); + stream.set_queued_bytes(0); + (ops.on_overflow)(stream); + } else if (ops.enqueue)(stream).is_err() { + reset_stream = true; + } else { + queued_bytes = queued_bytes.saturating_add(incoming_len); + } + + if !discarding + && !consume_stream_data( + &mut consumed_offset, + rx_bytes, + &mut ops.consume, + &mut ops.on_consume_error, + ) + { + reset_stream = true; + } + } else { + if (ops.enqueue)(stream).is_err() { + reset_stream = true; + } else { + queued_bytes = queued_bytes.saturating_add(incoming_len); + } + + if config.reserve_bytes > 0 + && !discarding + && !consume_stream_data( + &mut consumed_offset, + reserve_target_offset(rx_bytes, queued_bytes, fin_offset, config.reserve_bytes), + &mut ops.consume, + &mut ops.on_consume_error, + ) + { + reset_stream = true; + } + } + + stream.set_queued_bytes(queued_bytes); + stream.set_consumed_offset(consumed_offset); + stream.set_discarding(discarding); + stream.set_stop_sending_sent(stop_sending_sent); + + reset_stream +} From cc2fcc8f91838734cd427948bfe37796b7ae87ad Mon Sep 17 00:00:00 2001 From: Mygod Date: Fri, 23 Jan 2026 14:05:35 -0800 Subject: [PATCH 9/9] More refactor --- crates/slipstream-client/src/streams.rs | 159 +++++++-------------- crates/slipstream-core/src/flow_control.rs | 119 +++++++++++++-- crates/slipstream-server/src/streams.rs | 152 ++++++-------------- 3 files changed, 206 insertions(+), 224 deletions(-) diff --git a/crates/slipstream-client/src/streams.rs b/crates/slipstream-client/src/streams.rs index a44a74e8..f9b2b3e2 100644 --- a/crates/slipstream-client/src/streams.rs +++ b/crates/slipstream-client/src/streams.rs @@ -1,7 +1,7 @@ use slipstream_core::flow_control::{ - conn_reserve_bytes, consume_stream_data, handle_stream_receive, promote_streams, - reserve_target_offset, stream_queue_max_bytes, FlowControlStream, PromoteEntry, - StreamReceiveConfig, StreamReceiveOps, + conn_reserve_bytes, consume_error_log_message, consume_stream_data, handle_stream_receive, + overflow_log_message, promote_error_log_message, promote_streams, reserve_target_offset, + FlowControlState, HasFlowControlState, PromoteEntry, StreamReceiveConfig, StreamReceiveOps, }; use slipstream_core::tcp::{stream_read_limit_chunks, tcp_send_buffer_bytes}; use slipstream_ffi::picoquic::{ @@ -99,59 +99,18 @@ struct ClientStream { write_tx: mpsc::UnboundedSender, read_abort_tx: Option>, data_rx: Option>>, - queued_bytes: usize, - rx_bytes: u64, tx_bytes: u64, - consumed_offset: u64, - fin_offset: Option, fin_enqueued: bool, - discarding: bool, - stop_sending_sent: bool, + flow: FlowControlState, } -impl FlowControlStream for ClientStream { - fn queued_bytes(&self) -> usize { - self.queued_bytes +impl HasFlowControlState for ClientStream { + fn flow_control(&self) -> &FlowControlState { + &self.flow } - fn set_queued_bytes(&mut self, value: usize) { - self.queued_bytes = value; - } - - fn rx_bytes(&self) -> u64 { - self.rx_bytes - } - - fn set_rx_bytes(&mut self, value: u64) { - self.rx_bytes = value; - } - - fn consumed_offset(&self) -> u64 { - self.consumed_offset - } - - fn set_consumed_offset(&mut self, value: u64) { - self.consumed_offset = value; - } - - fn fin_offset(&self) -> Option { - self.fin_offset - } - - fn discarding(&self) -> bool { - self.discarding - } - - fn set_discarding(&mut self, value: bool) { - self.discarding = value; - } - - fn stop_sending_sent(&self) -> bool { - self.stop_sending_sent - } - - fn set_stop_sending_sent(&mut self, value: bool) { - self.stop_sending_sent = value; + fn flow_control_mut(&mut self) -> &mut FlowControlState { + &mut self.flow } } @@ -227,11 +186,11 @@ pub(crate) unsafe extern "C" fn client_callback( "stream {}: reset event={} rx_bytes={} tx_bytes={} queued={} consumed_offset={} fin_offset={:?} fin_enqueued={}", stream_id, reason, - stream.rx_bytes, + stream.flow.rx_bytes, stream.tx_bytes, - stream.queued_bytes, - stream.consumed_offset, - stream.fin_offset, + stream.flow.queued_bytes, + stream.flow.consumed_offset, + stream.flow.fin_offset, stream.fin_enqueued ); } else { @@ -319,19 +278,10 @@ fn handle_stream_data( return; }; - let max_queue = if multi_stream { - stream_queue_max_bytes() - } else { - 0 - }; if handle_stream_receive( stream, data.len(), - StreamReceiveConfig { - multi_stream, - reserve_bytes, - max_queue, - }, + StreamReceiveConfig::new(multi_stream, reserve_bytes), StreamReceiveOps { enqueue: |stream: &mut ClientStream| { if stream @@ -341,7 +291,10 @@ fn handle_stream_data( { warn!( "stream {}: tcp write channel closed queued={} rx_bytes={} tx_bytes={}", - stream_id, stream.queued_bytes, stream.rx_bytes, stream.tx_bytes + stream_id, + stream.flow.queued_bytes, + stream.flow.rx_bytes, + stream.tx_bytes ); Err(()) } else { @@ -360,15 +313,12 @@ fn handle_stream_data( unsafe { picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) }; }, log_overflow: |queued, incoming, max| { - warn!( - "stream {}: queued_bytes {} + {} exceeds limit {}; stopping", - stream_id, queued, incoming, max - ); + warn!("{}", overflow_log_message(stream_id, queued, incoming, max)); }, on_consume_error: |ret, current, target| { warn!( - "stream {}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", - stream_id, "", ret, current, target + "{}", + consume_error_log_message(stream_id, "", ret, current, target) ); }, }, @@ -377,11 +327,11 @@ fn handle_stream_data( } if fin { - if stream.discarding { + if stream.flow.discarding { remove_stream = true; } else { - if stream.fin_offset.is_none() { - stream.fin_offset = Some(stream.rx_bytes); + if stream.flow.fin_offset.is_none() { + stream.flow.fin_offset = Some(stream.flow.rx_bytes); } stream.data_rx = None; if !stream.fin_enqueued { @@ -389,8 +339,8 @@ fn handle_stream_data( warn!( "stream {}: tcp write channel closed on fin queued={} rx_bytes={} tx_bytes={}", stream_id, - stream.queued_bytes, - stream.rx_bytes, + stream.flow.queued_bytes, + stream.flow.rx_bytes, stream.tx_bytes ); reset_stream = true; @@ -401,7 +351,11 @@ fn handle_stream_data( } } - if !reset_stream && !stream.discarding && stream.fin_enqueued && stream.queued_bytes == 0 { + if !reset_stream + && !stream.flow.discarding + && stream.fin_enqueued + && stream.flow.queued_bytes == 0 + { remove_stream = true; } } @@ -511,14 +465,9 @@ pub(crate) fn handle_command( write_tx, read_abort_tx: Some(read_abort_tx), data_rx: Some(data_rx), - queued_bytes: 0, - rx_bytes: 0, tx_bytes: 0, - consumed_offset: 0, - fin_offset: None, fin_enqueued: false, - discarding: false, - stop_sending_sent: false, + flow: FlowControlState::default(), }, ); if !state.multi_stream_mode && state.streams.len() > 1 { @@ -529,17 +478,17 @@ pub(crate) fn handle_command( .iter_mut() .map(|(stream_id, stream)| PromoteEntry { stream_id: *stream_id, - rx_bytes: stream.rx_bytes, - consumed_offset: &mut stream.consumed_offset, - discarding: stream.discarding, + rx_bytes: stream.flow.rx_bytes, + consumed_offset: &mut stream.flow.consumed_offset, + discarding: stream.flow.discarding, }), |stream_id, new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, |stream_id, ret, consumed_offset, rx_bytes| { warn!( - "stream {}: stream_data_consumed failed during promote ret={} consumed_offset={} target={}", - stream_id, ret, consumed_offset, rx_bytes + "{}", + promote_error_log_message(stream_id, ret, consumed_offset, rx_bytes) ); }, ); @@ -585,11 +534,11 @@ pub(crate) fn handle_command( warn!( "stream {}: tcp read error rx_bytes={} tx_bytes={} queued={} consumed_offset={} fin_offset={:?}", stream_id, - stream.rx_bytes, + stream.flow.rx_bytes, stream.tx_bytes, - stream.queued_bytes, - stream.consumed_offset, - stream.fin_offset + stream.flow.queued_bytes, + stream.flow.consumed_offset, + stream.flow.fin_offset ); } else { warn!("stream {}: tcp read error (unknown stream)", stream_id); @@ -601,11 +550,11 @@ pub(crate) fn handle_command( warn!( "stream {}: tcp write error rx_bytes={} tx_bytes={} queued={} consumed_offset={} fin_offset={:?}", stream_id, - stream.rx_bytes, + stream.flow.rx_bytes, stream.tx_bytes, - stream.queued_bytes, - stream.consumed_offset, - stream.fin_offset + stream.flow.queued_bytes, + stream.flow.consumed_offset, + stream.flow.fin_offset ); } else { warn!("stream {}: tcp write error (unknown stream)", stream_id); @@ -615,27 +564,27 @@ pub(crate) fn handle_command( Command::StreamWriteDrained { stream_id, bytes } => { let mut remove_stream = false; if let Some(stream) = state.streams.get_mut(&stream_id) { - if stream.discarding { + if stream.flow.discarding { return; } - stream.queued_bytes = stream.queued_bytes.saturating_sub(bytes); + stream.flow.queued_bytes = stream.flow.queued_bytes.saturating_sub(bytes); if !state.multi_stream_mode { let new_offset = reserve_target_offset( - stream.rx_bytes, - stream.queued_bytes, - stream.fin_offset, + stream.flow.rx_bytes, + stream.flow.queued_bytes, + stream.flow.fin_offset, conn_reserve_bytes(), ); if !consume_stream_data( - &mut stream.consumed_offset, + &mut stream.flow.consumed_offset, new_offset, |new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, |ret, current, target| { warn!( - "stream {}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", - stream_id, "", ret, current, target + "{}", + consume_error_log_message(stream_id, "", ret, current, target) ); }, ) { @@ -646,7 +595,7 @@ pub(crate) fn handle_command( return; } } - if stream.fin_enqueued && stream.queued_bytes == 0 { + if stream.fin_enqueued && stream.flow.queued_bytes == 0 { remove_stream = true; } } diff --git a/crates/slipstream-core/src/flow_control.rs b/crates/slipstream-core/src/flow_control.rs index 18c7f7ed..40687c12 100644 --- a/crates/slipstream-core/src/flow_control.rs +++ b/crates/slipstream-core/src/flow_control.rs @@ -3,20 +3,69 @@ use std::sync::OnceLock; const DEFAULT_STREAM_QUEUE_MAX_BYTES: usize = 2 * 1024 * 1024; const DEFAULT_CONN_RESERVE_BYTES: usize = 64 * 1024; -pub trait FlowControlStream { - fn queued_bytes(&self) -> usize; - fn set_queued_bytes(&mut self, value: usize); - fn rx_bytes(&self) -> u64; - fn set_rx_bytes(&mut self, value: u64); - fn consumed_offset(&self) -> u64; - fn set_consumed_offset(&mut self, value: u64); - fn fin_offset(&self) -> Option; - fn discarding(&self) -> bool; - fn set_discarding(&mut self, value: bool); - fn stop_sending_sent(&self) -> bool; - fn set_stop_sending_sent(&mut self, value: bool); +#[derive(Debug, Default)] +pub struct FlowControlState { + pub queued_bytes: usize, + pub rx_bytes: u64, + pub consumed_offset: u64, + pub fin_offset: Option, + pub discarding: bool, + pub stop_sending_sent: bool, +} + +pub trait HasFlowControlState { + fn flow_control(&self) -> &FlowControlState; + fn flow_control_mut(&mut self) -> &mut FlowControlState; } +pub trait FlowControlStream: HasFlowControlState { + fn queued_bytes(&self) -> usize { + self.flow_control().queued_bytes + } + + fn set_queued_bytes(&mut self, value: usize) { + self.flow_control_mut().queued_bytes = value; + } + + fn rx_bytes(&self) -> u64 { + self.flow_control().rx_bytes + } + + fn set_rx_bytes(&mut self, value: u64) { + self.flow_control_mut().rx_bytes = value; + } + + fn consumed_offset(&self) -> u64 { + self.flow_control().consumed_offset + } + + fn set_consumed_offset(&mut self, value: u64) { + self.flow_control_mut().consumed_offset = value; + } + + fn fin_offset(&self) -> Option { + self.flow_control().fin_offset + } + + fn discarding(&self) -> bool { + self.flow_control().discarding + } + + fn set_discarding(&mut self, value: bool) { + self.flow_control_mut().discarding = value; + } + + fn stop_sending_sent(&self) -> bool { + self.flow_control().stop_sending_sent + } + + fn set_stop_sending_sent(&mut self, value: bool) { + self.flow_control_mut().stop_sending_sent = value; + } +} + +impl FlowControlStream for T {} + pub struct QueueOverflowOps { pub log_overflow: Log, pub consume: Consume, @@ -30,6 +79,21 @@ pub struct StreamReceiveConfig { pub max_queue: usize, } +impl StreamReceiveConfig { + pub fn new(multi_stream: bool, reserve_bytes: usize) -> Self { + let max_queue = if multi_stream { + stream_queue_max_bytes() + } else { + 0 + }; + Self { + multi_stream, + reserve_bytes, + max_queue, + } + } +} + pub struct StreamReceiveOps { pub enqueue: Enqueue, pub on_overflow: Overflow, @@ -39,6 +103,35 @@ pub struct StreamReceiveOps { pub on_consume_error: Err, } +pub fn overflow_log_message( + stream_id: u64, + queued_bytes: usize, + incoming_len: usize, + max_queue: usize, +) -> String { + format!( + "stream {}: queued_bytes {} + {} exceeds limit {}; stopping", + stream_id, queued_bytes, incoming_len, max_queue + ) +} + +pub fn consume_error_log_message( + stream_id: u64, + context: &str, + ret: i32, + current: u64, + target: u64, +) -> String { + format!( + "stream {}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", + stream_id, context, ret, current, target + ) +} + +pub fn promote_error_log_message(stream_id: u64, ret: i32, current: u64, target: u64) -> String { + consume_error_log_message(stream_id, " during promote", ret, current, target) +} + pub fn stream_queue_max_bytes() -> usize { static MAX_BYTES: OnceLock = OnceLock::new(); *MAX_BYTES.get_or_init(|| { @@ -115,7 +208,6 @@ where apply_consumed_offset(consumed_offset, target, consume_fn, on_error) } -#[allow(clippy::too_many_arguments)] pub fn handle_queue_overflow( queued_bytes: usize, incoming_len: usize, @@ -199,7 +291,6 @@ where } } -#[allow(clippy::too_many_arguments)] pub fn handle_stream_receive( stream: &mut S, incoming_len: usize, diff --git a/crates/slipstream-server/src/streams.rs b/crates/slipstream-server/src/streams.rs index 9b692fb3..44834908 100644 --- a/crates/slipstream-server/src/streams.rs +++ b/crates/slipstream-server/src/streams.rs @@ -1,9 +1,9 @@ use crate::server::{Command, StreamKey, StreamWrite}; use crate::target::spawn_target_connector; use slipstream_core::flow_control::{ - conn_reserve_bytes, consume_stream_data, handle_stream_receive, promote_streams, - reserve_target_offset, stream_queue_max_bytes, FlowControlStream, PromoteEntry, - StreamReceiveConfig, StreamReceiveOps, + conn_reserve_bytes, consume_error_log_message, consume_stream_data, handle_stream_receive, + overflow_log_message, promote_error_log_message, promote_streams, reserve_target_offset, + FlowControlState, HasFlowControlState, PromoteEntry, StreamReceiveConfig, StreamReceiveOps, }; use slipstream_ffi::picoquic::{ picoquic_call_back_event_t, picoquic_close, picoquic_close_immediate, picoquic_cnx_t, @@ -95,64 +95,23 @@ struct ServerStream { data_rx: Option>>, send_pending: Option>, send_stash: Option>, - queued_bytes: usize, shutdown_tx: watch::Sender, - rx_bytes: u64, - consumed_offset: u64, - fin_offset: Option, tx_bytes: u64, target_fin_pending: bool, close_after_flush: bool, pending_data: VecDeque>, pending_fin: bool, fin_enqueued: bool, - discarding: bool, - stop_sending_sent: bool, + flow: FlowControlState, } -impl FlowControlStream for ServerStream { - fn queued_bytes(&self) -> usize { - self.queued_bytes +impl HasFlowControlState for ServerStream { + fn flow_control(&self) -> &FlowControlState { + &self.flow } - fn set_queued_bytes(&mut self, value: usize) { - self.queued_bytes = value; - } - - fn rx_bytes(&self) -> u64 { - self.rx_bytes - } - - fn set_rx_bytes(&mut self, value: u64) { - self.rx_bytes = value; - } - - fn consumed_offset(&self) -> u64 { - self.consumed_offset - } - - fn set_consumed_offset(&mut self, value: u64) { - self.consumed_offset = value; - } - - fn fin_offset(&self) -> Option { - self.fin_offset - } - - fn discarding(&self) -> bool { - self.discarding - } - - fn set_discarding(&mut self, value: bool) { - self.discarding = value; - } - - fn stop_sending_sent(&self) -> bool { - self.stop_sending_sent - } - - fn set_stop_sending_sent(&mut self, value: bool) { - self.stop_sending_sent = value; + fn flow_control_mut(&mut self) -> &mut FlowControlState { + &mut self.flow } } @@ -214,13 +173,13 @@ pub(crate) unsafe extern "C" fn server_callback( key.stream_id, reason, stream.tx_bytes, - stream.rx_bytes, - stream.consumed_offset, - stream.queued_bytes, + stream.flow.rx_bytes, + stream.flow.consumed_offset, + stream.flow.queued_bytes, stream.pending_data.len(), stream.pending_fin, stream.fin_enqueued, - stream.fin_offset, + stream.flow.fin_offset, stream.target_fin_pending, stream.close_after_flush ); @@ -310,7 +269,7 @@ pub(crate) unsafe extern "C" fn server_callback( "stream {:?}: provide_stream_data_buffer returned null send_len={} queued={} pending_chunks={} tx_bytes={}", key.stream_id, send_len, - stream.queued_bytes, + stream.flow.queued_bytes, stream.pending_data.len(), stream.tx_bytes ); @@ -390,19 +349,14 @@ fn handle_stream_data( data_rx: None, send_pending: None, send_stash: None, - queued_bytes: 0, shutdown_tx, - rx_bytes: 0, - consumed_offset: 0, - fin_offset: None, tx_bytes: 0, target_fin_pending: false, close_after_flush: false, pending_data: VecDeque::new(), pending_fin: false, fin_enqueued: false, - discarding: false, - stop_sending_sent: false, + flow: FlowControlState::default(), }, ); } @@ -415,17 +369,17 @@ fn handle_stream_data( .filter(|(entry_key, _)| entry_key.cnx == key.cnx) .map(|(entry_key, stream)| PromoteEntry { stream_id: entry_key.stream_id, - rx_bytes: stream.rx_bytes, - consumed_offset: &mut stream.consumed_offset, - discarding: stream.discarding, + rx_bytes: stream.flow.rx_bytes, + consumed_offset: &mut stream.flow.consumed_offset, + discarding: stream.flow.discarding, }), |stream_id, new_offset| unsafe { picoquic_stream_data_consumed(cnx, stream_id, new_offset) }, |stream_id, ret, consumed_offset, rx_bytes| { warn!( - "stream {:?}: stream_data_consumed failed during promote ret={} consumed_offset={} target={}", - stream_id, ret, consumed_offset, rx_bytes + "{}", + promote_error_log_message(stream_id, ret, consumed_offset, rx_bytes) ); }, ); @@ -443,19 +397,10 @@ fn handle_stream_data( None => return, }; - let max_queue = if multi_stream { - stream_queue_max_bytes() - } else { - 0 - }; if handle_stream_receive( stream, data.len(), - StreamReceiveConfig { - multi_stream, - reserve_bytes, - max_queue, - }, + StreamReceiveConfig::new(multi_stream, reserve_bytes), StreamReceiveOps { enqueue: |stream: &mut ServerStream| { if let Some(write_tx) = stream.write_tx.as_ref() { @@ -487,15 +432,12 @@ fn handle_stream_data( unsafe { picoquic_stop_sending(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) }; }, log_overflow: |queued, incoming, max| { - warn!( - "stream {:?}: queued_bytes {} + {} exceeds limit {}; stopping", - stream_id, queued, incoming, max - ); + warn!("{}", overflow_log_message(stream_id, queued, incoming, max)); }, on_consume_error: |ret, current, target| { warn!( - "stream {:?}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", - stream_id, "", ret, current, target + "{}", + consume_error_log_message(stream_id, "", ret, current, target) ); }, }, @@ -504,13 +446,13 @@ fn handle_stream_data( } if fin { - if stream.discarding { + if stream.flow.discarding { if !reset_stream { remove_stream = true; } } else { - if stream.fin_offset.is_none() { - stream.fin_offset = Some(stream.rx_bytes); + if stream.flow.fin_offset.is_none() { + stream.flow.fin_offset = Some(stream.flow.rx_bytes); } if !stream.fin_enqueued { if stream.write_tx.is_some() && stream.pending_data.is_empty() { @@ -542,7 +484,7 @@ fn handle_stream_data( if !state .streams .get(&key) - .map(|stream| stream.discarding) + .map(|stream| stream.flow.discarding) .unwrap_or(false) { shutdown_stream(state, key); @@ -608,7 +550,7 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { if state.debug_streams { debug!("stream {:?}: target connected", stream_id); } - if stream.discarding { + if stream.flow.discarding { stream.pending_data.clear(); stream.pending_fin = false; stream.fin_enqueued = false; @@ -624,7 +566,7 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { warn!( "stream {:?}: pending write flush failed queued={} pending_chunks={} tx_bytes={}", stream_id, - stream.queued_bytes, + stream.flow.queued_bytes, stream.pending_data.len(), stream.tx_bytes ); @@ -637,7 +579,7 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { warn!( "stream {:?}: pending fin flush failed queued={} pending_chunks={} tx_bytes={}", stream_id, - stream.queued_bytes, + stream.flow.queued_bytes, stream.pending_data.len(), stream.tx_bytes ); @@ -726,10 +668,10 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { "stream {:?}: target read error tx_bytes={} rx_bytes={} consumed_offset={} queued={} fin_offset={:?}", stream_id, stream.tx_bytes, - stream.rx_bytes, - stream.consumed_offset, - stream.queued_bytes, - stream.fin_offset + stream.flow.rx_bytes, + stream.flow.consumed_offset, + stream.flow.queued_bytes, + stream.flow.fin_offset ); let _ = unsafe { picoquic_reset_stream(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) }; } @@ -745,10 +687,10 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { "stream {:?}: target write failed tx_bytes={} rx_bytes={} consumed_offset={} queued={} fin_offset={:?}", stream_id, stream.tx_bytes, - stream.rx_bytes, - stream.consumed_offset, - stream.queued_bytes, - stream.fin_offset + stream.flow.rx_bytes, + stream.flow.consumed_offset, + stream.flow.queued_bytes, + stream.flow.fin_offset ); let _ = unsafe { picoquic_reset_stream(cnx, stream_id, SLIPSTREAM_INTERNAL_ERROR) }; } @@ -764,19 +706,19 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { }; let mut reset_stream = false; if let Some(stream) = state.streams.get_mut(&key) { - if stream.discarding { + if stream.flow.discarding { return; } - stream.queued_bytes = stream.queued_bytes.saturating_sub(bytes); + stream.flow.queued_bytes = stream.flow.queued_bytes.saturating_sub(bytes); if !state.multi_streams.contains(&cnx_id) { let new_offset = reserve_target_offset( - stream.rx_bytes, - stream.queued_bytes, - stream.fin_offset, + stream.flow.rx_bytes, + stream.flow.queued_bytes, + stream.flow.fin_offset, conn_reserve_bytes(), ); if !consume_stream_data( - &mut stream.consumed_offset, + &mut stream.flow.consumed_offset, new_offset, |new_offset| unsafe { picoquic_stream_data_consumed( @@ -787,8 +729,8 @@ pub(crate) fn handle_command(state_ptr: *mut ServerState, command: Command) { }, |ret, current, target| { warn!( - "stream {:?}: stream_data_consumed failed{} ret={} consumed_offset={} target={}", - stream_id, "", ret, current, target + "{}", + consume_error_log_message(stream_id, "", ret, current, target) ); }, ) {