diff --git a/ostp-client/src/bridge.rs b/ostp-client/src/bridge.rs index 3e4a821..bbe83b9 100644 --- a/ostp-client/src/bridge.rs +++ b/ostp-client/src/bridge.rs @@ -309,7 +309,15 @@ impl Bridge { // which on a mobile uplink is minutes of buffered queue rather // than throughput — the app kept handing over data long after // the path had stopped draining it. - s.iter().any(|ses| ses.machine.in_flight_count() < ses.machine.cwnd_packets().clamp(16, 1024)) + // Two independent gates. cwnd bounds how much may be in + // flight; pacing bounds how FAST it is released. Without the + // second, a full window goes out back-to-back and lands in + // the bottleneck's buffer as standing queue rather than + // throughput — the thing that produced multi-second RTT. + s.iter().any(|ses| { + ses.machine.in_flight_count() < ses.machine.cwnd_packets().clamp(16, 1024) + && ses.machine.can_pace_packet() + }) }).unwrap_or(true) => { self.handle_proxy_event(proxy_ev, &mut sessions_opt, &mut stream_map, &tx, &proxy_tx).await; } diff --git a/ostp-core/src/congestion.rs b/ostp-core/src/congestion.rs index 62955ae..89d2ca8 100644 --- a/ostp-core/src/congestion.rs +++ b/ostp-core/src/congestion.rs @@ -39,6 +39,9 @@ pub struct CongestionController { loss_count: u32, /// Pacing rate: bytes per second pacing_rate: u64, + /// Token-bucket allowance for pacing, in bytes. + pacing_tokens: f64, + pacing_last_refill: Instant, /// MTU estimate (used for cwnd → packet count conversion) mtu: u64, /// Min RTT expiry: re-probe after 10 seconds @@ -77,6 +80,8 @@ const MAX_CWND_PACKETS: u64 = 1024; const RTT_INFLATION_EXIT_SLOW_START: f64 = 2.0; /// SRTT/min_rtt ratio treated as a standing queue that must be actively drained. const RTT_INFLATION_BACKOFF: f64 = 4.0; +/// How much pacing allowance may accumulate, expressed as time-at-rate. +const PACING_BURST: Duration = Duration::from_millis(10); const RTO_MIN: Duration = Duration::from_millis(50); /// Maximum RTO const RTO_MAX: Duration = Duration::from_secs(16); @@ -125,9 +130,50 @@ impl CongestionController { min_rtt_stamp: now, slow_start_losses: 0, slow_start_loss_window_start: now, + pacing_tokens: (INITIAL_CWND_PACKETS * mtu) as f64, + pacing_last_refill: now, } } + /// Bytes of pacing allowance available right now, without consuming any. + /// + /// Read-only so the send path can use it as an admission check before it + /// commits to building a datagram. + pub fn pacing_available(&self) -> f64 { + let elapsed = self.pacing_last_refill.elapsed().as_secs_f64(); + (self.pacing_tokens + elapsed * self.pacing_rate as f64).min(self.pacing_burst()) + } + + /// Whether at least one full-size packet may be released right now. + pub fn can_pace_packet(&self) -> bool { + self.pacing_available() >= self.mtu as f64 + } + + /// Ceiling on accumulated allowance. + /// + /// Pacing intervals here are fractions of a millisecond, so releasing + /// strictly one packet at a time would need a sub-millisecond timer per + /// packet. Instead we allow a short burst — the same trade every real + /// pacing implementation makes — sized so the loop's existing ~10ms wakeups + /// can still saturate the configured rate, with a small floor so a + /// cold/low estimate can never wedge sending entirely. + fn pacing_burst(&self) -> f64 { + let by_rate = self.pacing_rate as f64 * PACING_BURST.as_secs_f64(); + by_rate.max((self.mtu * 4) as f64) + } + + /// Refill from elapsed time and deduct `bytes`. Called on the real send + /// path; allowance is permitted to go negative so an oversized packet still + /// pays for itself rather than being released for free. + fn consume_pacing(&mut self, bytes: u64) { + let now = Instant::now(); + let elapsed = now.duration_since(self.pacing_last_refill).as_secs_f64(); + self.pacing_last_refill = now; + self.pacing_tokens = + (self.pacing_tokens + elapsed * self.pacing_rate as f64).min(self.pacing_burst()) + - bytes as f64; + } + /// Returns the current congestion window in bytes. pub fn cwnd(&self) -> u64 { self.cwnd @@ -179,6 +225,11 @@ impl CongestionController { /// Record that we sent `bytes` of data. pub fn on_send(&mut self, bytes: u64) { self.bytes_in_flight = self.bytes_in_flight.saturating_add(bytes); + // Charge the pacing bucket here rather than at the admission check, so + // every byte that actually reaches the wire is paid for exactly once — + // including retransmits, which are precisely what must not be allowed + // to bypass the rate limit and pile into an already-full queue. + self.consume_pacing(bytes); } /// Record that `bytes` were acknowledged but WITHOUT a usable RTT sample @@ -425,6 +476,47 @@ mod tests { ); } + /// Pacing must actually bound the release rate: draining the bucket has to + /// deny the next packet. Without this the congestion window alone decides, + /// and a whole window leaves back-to-back. + #[test] + fn test_pacing_bucket_denies_once_drained() { + let mut cc = CongestionController::new(1200); + assert!(cc.can_pace_packet(), "a fresh controller must allow sending"); + + // Spend well beyond one burst allowance. + let burst_bytes = cc.pacing_available(); + let mut spent = 0.0; + while spent <= burst_bytes + 1200.0 { + cc.on_send(1200); + spent += 1200.0; + } + + assert!( + !cc.can_pace_packet(), + "pacing allowed unbounded sending: {} bytes still available after spending {}", + cc.pacing_available(), + spent + ); + } + + /// The allowance must refill over time, or sending would stall permanently + /// once the first burst is spent. + #[test] + fn test_pacing_bucket_refills_over_time() { + let mut cc = CongestionController::new(1200); + while cc.can_pace_packet() { + cc.on_send(1200); + } + assert!(!cc.can_pace_packet()); + + std::thread::sleep(Duration::from_millis(25)); + assert!( + cc.can_pace_packet(), + "pacing bucket never refilled; sending would be stuck forever" + ); + } + /// cwnd must never exceed the absolute ceiling, however long slow start /// runs unopposed — above it the window is buffered queue, not throughput. #[test] diff --git a/ostp-core/src/protocol.rs b/ostp-core/src/protocol.rs index 8520a03..c4eee46 100644 --- a/ostp-core/src/protocol.rs +++ b/ostp-core/src/protocol.rs @@ -188,6 +188,16 @@ impl ProtocolMachine { self.cc.cwnd_packets() as usize } + /// Whether the pacing bucket currently allows releasing another packet. + /// + /// The congestion window bounds how much may be UNACKNOWLEDGED; it says + /// nothing about how fast that window is emptied onto the wire. Sending a + /// whole window back-to-back is what drives a deep buffer into standing + /// queue, so admission is gated on both. + pub fn can_pace_packet(&self) -> bool { + self.cc.can_pace_packet() + } + pub fn on_send(&mut self, bytes: u64) { self.cc.on_send(bytes); } diff --git a/ostp-server/src/dispatcher.rs b/ostp-server/src/dispatcher.rs index 22a03ca..35d7a1b 100644 --- a/ostp-server/src/dispatcher.rs +++ b/ostp-server/src/dispatcher.rs @@ -263,8 +263,19 @@ impl Dispatcher { self.peer_machines .iter() .map(|(&sid, ps)| { - let cwnd = (ps.machine.cwnd_packets() as i64).clamp(16, 16384); + // Ceiling matches MAX_CWND_PACKETS in ostp-core. The old 16384 + // allowed ~20 MB outstanding toward one client — on a mobile + // downlink that is standing queue, not throughput, and it is the + // download direction that carries video. + let cwnd = (ps.machine.cwnd_packets() as i64).clamp(16, 1024); let in_flight = ps.machine.in_flight_count() as i64; + // Pacing gates the RATE, cwnd only the outstanding amount. With + // the pacing bucket empty, report no headroom so the relay + // reader pauses instead of handing over another chunk that would + // leave back-to-back. + if !ps.machine.can_pace_packet() { + return (sid, 0); + } (sid, cwnd - in_flight) }) .collect()