diff --git a/crates/mq-integration/src/shard_pair.rs b/crates/mq-integration/src/shard_pair.rs index c3a00f3c..a0db07e2 100644 --- a/crates/mq-integration/src/shard_pair.rs +++ b/crates/mq-integration/src/shard_pair.rs @@ -540,8 +540,14 @@ pub fn spawn_server(cfg: ServerConfig, max_conns: u32) -> Side { /// The real `Client` at `client_addr()` with SOCKS5 and transparent listeners. pub fn spawn_client(cfg: ClientConfig) -> Side { + spawn_client_cc(cfg, CongestionControl::Bbr) +} + +/// `spawn_client` with another congestion control. +pub fn spawn_client_cc(cfg: ClientConfig, cc: CongestionControl) -> Side { let mut tc = transport_cfg(Role::Client, 0); tc.scheduler = cfg.scheduler; + tc.cc = cc; Side::spawn(tc, client_addr(), vec![SOCKS5, TRANSPARENT], move || { Client::new(cfg) }) diff --git a/crates/mq-integration/tests/pair_relay.rs b/crates/mq-integration/tests/pair_relay.rs index e5275038..410666a9 100644 --- a/crates/mq-integration/tests/pair_relay.rs +++ b/crates/mq-integration/tests/pair_relay.rs @@ -8,7 +8,7 @@ use mq_proxy::config::ClientConfig; use mq_proxy::server::Server; use mq_runtime::testing::Op; use mq_transport_api::fabric::Rule; -use mq_transport_api::{ConnId, PathId, Time, TransportOps}; +use mq_transport_api::{CongestionControl, ConnId, PathId, Time, TransportOps}; /// spec §8.1: authentication, then a SOCKS5 request relayed to an echo origin. #[test] @@ -36,12 +36,12 @@ fn pair_auth_and_relay() { const ACTIVE: u32 = 2; // XQC_PATH_STATE_ACTIVE /// A client with a second `--path`, authenticated, both paths active. -fn two_paths() -> (Pair, ConnId) { +fn two_paths(cc: CongestionControl) -> (Pair, ConnId) { let cfg = ClientConfig { paths: vec![client_addr().ip(), CLIENT_IP2], ..client_cfg() }; - let mut p = Pair::new(spawn_server(server_cfg(), 0), spawn_client(cfg)); + let mut p = Pair::new(spawn_server(server_cfg(), 0), spawn_client_cc(cfg, cc)); assert!(p.run_until(5 * SEC, |p| p.auth_attempts() == 1)); let c = p.client_conns()[0]; let up = move |p: &mut Pair| { @@ -59,7 +59,7 @@ fn two_paths() -> (Pair, ConnId) { /// socket and both paths carry the relayed bytes. #[test] fn pair_two_paths_carry_traffic() { - let (mut p, c) = two_paths(); + let (mut p, c) = two_paths(CongestionControl::Bbr); const N: usize = 1 << 20; p.origin(OriginMode::Send(bulk(N))); let s = p.socks_open(origin_addr(), b""); @@ -83,10 +83,11 @@ fn pair_two_paths_carry_traffic() { /// window has a bandwidth-delay product to grow into. #[test] fn pair_blocked_path_quota_and_resume() { - const QUOTA: usize = 256 * 1024; // spec §4.4 + const QUOTA: usize = 1024 * 1024; // spec §4.4 const WARM: usize = 8 << 20; const N: usize = 8 << 20; - let (mut p, c) = two_paths(); + // Cubic: in the fabric BBR's window on path 1 settles below one queue quota. + let (mut p, c) = two_paths(CongestionControl::Cubic); p.fabric.add_rule(Rule::DelayRange { min: 5 * MS, max: 5 * MS, @@ -175,7 +176,7 @@ fn pair_blocked_path_quota_and_resume() { /// the primary) and both paths are active again. fn blackhole_one_path_then_recover(ip: std::net::IpAddr, dead: u64) { const CLOSED: u32 = 4; // XQC_PATH_STATE_CLOSED - let (mut p, c) = two_paths(); + let (mut p, c) = two_paths(CongestionControl::Bbr); let paths = move |p: &Pair| { let st = p.with_client(move |n| n.transport().conn_stats(c)).unwrap(); st.paths.iter().map(|x| (x.id, x.state)).collect::>() diff --git a/crates/mq-transport/src/txq.rs b/crates/mq-transport/src/txq.rs index 8c39fee5..7c6b7843 100644 --- a/crates/mq-transport/src/txq.rs +++ b/crates/mq-transport/src/txq.rs @@ -3,10 +3,12 @@ use std::net::SocketAddr; use mq_transport_api::{ConnId, Transmit, TxKey}; -// spec §4.4 -pub const QUEUE_QUOTA: usize = 256 * 1024; -pub const QUEUE_LOW: usize = 128 * 1024; -pub const TOTAL_HIGH: usize = 2 * 1024 * 1024; +// spec §4.4. One drive can hand xquic's output for a path over 256 KiB at ~1 Gbps; each +// refusal below makes xquic retry on every wakeup, so the quota sits well above that. +// Queues grow on use, so an idle (conn, path) costs nothing. +pub const QUEUE_QUOTA: usize = 1024 * 1024; +pub const QUEUE_LOW: usize = 512 * 1024; +pub const TOTAL_HIGH: usize = 4 * 1024 * 1024; #[derive(Clone, Copy, PartialEq, Eq, Debug)] pub enum Refusal { @@ -37,7 +39,7 @@ struct Queue { impl Queue { fn new() -> Self { Self { - buf: Vec::with_capacity(QUEUE_QUOTA + 1500), + buf: Vec::new(), head: 0, tail: 0, pkts: VecDeque::new(), @@ -47,7 +49,7 @@ impl Queue { fn append(&mut self, dst: SocketAddr, pkt: &[u8]) { if self.tail + pkt.len() > self.buf.capacity() { - self.compact(); // memmove only here + self.compact(); // memmove only here; the extend below grows the buffer if still short } self.buf.truncate(self.tail); self.buf.extend_from_slice(pkt); @@ -233,6 +235,11 @@ mod tests { fn addr(n: u8) -> SocketAddr { SocketAddr::from(([10, 0, 0, n], 4433)) } + /// Keys that fill to exactly TOTAL_HIGH; key(N) is the next one. + const N: u32 = (TOTAL_HIGH / QUEUE_QUOTA) as u32; + /// 1024-byte packets to drain a full queue to exactly QUEUE_LOW. + const TO_LOW: usize = (QUEUE_QUOTA - QUEUE_LOW) / 1024; + /// Fill `k` to exactly QUEUE_QUOTA with 1024-byte packets. fn fill(t: &mut TxQueues, k: TxKey) { for _ in 0..QUEUE_QUOTA / 1024 { @@ -301,9 +308,18 @@ mod tests { assert_eq!((q.head, q.tail, q.bytes, t.total), (0, 0, 0, 0)); } + #[test] + fn fresh_queue_reserves_nothing_and_grows_with_use() { + let mut q = Queue::new(); + assert_eq!(q.buf.capacity(), 0); + q.append(addr(1), &[0; 1200]); + assert!((1200..QUEUE_QUOTA).contains(&q.buf.capacity())); + } + #[test] fn compaction_only_when_capacity_reached() { let mut q = Queue::new(); + q.buf.reserve_exact(64 * 1024); let cap = q.buf.capacity(); let p = [9u8; 1000]; while q.tail + 1000 * 2 <= cap { @@ -335,7 +351,7 @@ mod tests { let k = key(0); fill(&mut t, k); t.record_blocked(conn(0), k, Refusal::Quota); - t.done(k, 128); // exactly at QUEUE_LOW: not below + t.done(k, TO_LOW); // exactly at QUEUE_LOW: not below assert!(!t.resume_pending()); t.done(k, 1); assert!(t.resume_pending()); @@ -346,24 +362,24 @@ mod tests { #[test] fn total_refuses_at_high_water() { let mut t = TxQueues::default(); - for i in 0..8 { + for i in 0..N { fill(&mut t, key(i)); } assert_eq!(t.total, TOTAL_HIGH); - assert_eq!(t.push(key(8), addr(1), &[0; 1]), Err(Refusal::Total)); + assert_eq!(t.push(key(N), addr(1), &[0; 1]), Err(Refusal::Total)); } #[test] fn total_resumes_on_any_commit_below_high_even_with_two_keys_at_quota() { let mut t = TxQueues::default(); - for i in 0..8 { + for i in 0..N { fill(&mut t, key(i)); } t.record_blocked(conn(0), key(0), Refusal::Quota); t.record_blocked(conn(1), key(1), Refusal::Quota); - t.record_blocked(conn(8), key(8), Refusal::Total); + t.record_blocked(conn(N), key(N), Refusal::Total); t.done(key(3), 1); // unrelated key; total now below high - assert_eq!(t.take_resumable(), vec![conn(8)]); + assert_eq!(t.take_resumable(), vec![conn(N)]); assert!(t.blocked.contains_key(&conn(0)) && t.blocked.contains_key(&conn(1))); } @@ -375,7 +391,7 @@ mod tests { // blocked on two quota keys: draining only A resumes it t.record_blocked(conn(0), key(0), Refusal::Quota); t.record_blocked(conn(0), key(1), Refusal::Quota); - t.done(key(0), 129); + t.done(key(0), TO_LOW + 1); assert_eq!(t.take_resumable(), vec![conn(0)]); assert!(!t.blocked.contains_key(&conn(0))); // quota + total: a commit below high-water clears via total while the quota key stays full @@ -399,17 +415,17 @@ mod tests { #[test] fn drop_conn_counts_as_commit_and_forgets_block() { let mut t = TxQueues::default(); - for i in 0..8 { + for i in 0..N { fill(&mut t, key(i)); } t.push_or_drop((Some(conn(0)), PathId(1)), addr(1), &[0; 1]); t.record_blocked(conn(0), key(0), Refusal::Quota); - t.record_blocked(conn(8), key(8), Refusal::Total); + t.record_blocked(conn(N), key(N), Refusal::Total); t.drop_conn(conn(0)); - assert_eq!(t.total, 7 * QUEUE_QUOTA); + assert_eq!(t.total, (N as usize - 1) * QUEUE_QUOTA); assert!(t.queues.keys().all(|k| k.0 != Some(conn(0)))); assert!(!t.blocked.contains_key(&conn(0))); - assert_eq!(t.take_resumable(), vec![conn(8)]); + assert_eq!(t.take_resumable(), vec![conn(N)]); } #[test] diff --git a/crates/mq-transport/tests/fabric_two_blocked_paths.rs b/crates/mq-transport/tests/fabric_two_blocked_paths.rs index abf01714..c65c4081 100644 --- a/crates/mq-transport/tests/fabric_two_blocked_paths.rs +++ b/crates/mq-transport/tests/fabric_two_blocked_paths.rs @@ -8,7 +8,7 @@ use common::xfer::Xfer; use mq_transport_api::{CongestionControl, Event, PathId, Role, TransportOps, TxKey}; /// spec §4.4 per-queue quota. -const QUEUE_QUOTA: usize = 256 * 1024; +const QUEUE_QUOTA: usize = 1024 * 1024; #[test] fn two_blocked_paths_one_drains_resumes_at_once() { diff --git a/crates/mq-transport/tests/fabric_txq.rs b/crates/mq-transport/tests/fabric_txq.rs index db1f43a4..1b026192 100644 --- a/crates/mq-transport/tests/fabric_txq.rs +++ b/crates/mq-transport/tests/fabric_txq.rs @@ -1,18 +1,21 @@ //! spec §4.4, §8.4 "Transmit queue full, then drained". mod common; -use common::pair::{MS, Pair}; +use common::pair::{MS, Opts, Pair}; use common::xfer::Xfer; -use mq_transport_api::{PathId, TransportOps}; +use mq_transport_api::{CongestionControl, PathId, TransportOps}; /// spec §4.4 per-queue quota. -const QUEUE_QUOTA: usize = 256 * 1024; +const QUEUE_QUOTA: usize = 1024 * 1024; #[test] fn txq_full_then_drained_resumes() { - let mut p = Pair::new(); + // Cubic: in the fabric BBR's window settles below one queue quota; cubic's keeps growing. + let mut o = Opts::default(); + o.client.cc = CongestionControl::Cubic; + let mut p = Pair::with(o); let key = (Some(p.conn), PathId(0)); - let mut x = Xfer::start(&p, 16 * 1024 * 1024); + let mut x = Xfer::start(&p, 64 * 1024 * 1024); // Grow the congestion window past the queue quota first, or the window, not the queue, // stops the sender. assert!(p.pump_until(MS, 20_000, |p| {