diff --git a/src/connection/connection.rs b/src/connection/connection.rs index 1b4190bc5..a80b03e70 100644 --- a/src/connection/connection.rs +++ b/src/connection/connection.rs @@ -3721,6 +3721,41 @@ impl Connection { self.paths.mark_ping(path_addr) } + /// Switch the multipath scheduling algorithm live. Takes effect on the + /// next packet-send decision; a no-op scheduler rebuild if multipath was + /// never negotiated (the setting still applies if it is later enabled). + pub fn set_multipath_algorithm(&mut self, alg: crate::MultipathAlgorithm) { + self.multipath_conf.multipath_algorithm = alg; + if self.flags.contains(EnableMultipath) { + self.multipath_scheduler = Some(build_multipath_scheduler(&self.multipath_conf)); + } + } + + /// Set (or clear, with None) the zombie-bufferbloat cutoff: paths whose + /// smoothed RTT exceeds `max_ms` are avoided by the multipath schedulers + /// unless no healthy path can send. + pub fn set_scheduler_max_rtt(&mut self, max_ms: Option) { + self.paths.max_srtt = max_ms.map(time::Duration::from_millis); + } + + /// Per-path health snapshot: + /// (local, remote, srtt_ms, consecutive_ptos, active, unhealthy). + pub fn path_health(&self) -> Vec<(SocketAddr, SocketAddr, u64, usize, bool, bool)> { + self.paths + .iter() + .map(|(_, p)| { + ( + p.local_addr(), + p.remote_addr(), + p.recovery.rtt.smoothed_rtt().as_millis() as u64, + p.recovery.consecutive_pto_count(), + p.active(), + p.unhealthy_with(self.paths.max_srtt), + ) + }) + .collect() + } + /// Client add a new path on the connection. pub fn add_path(&mut self, local_addr: SocketAddr, remote_addr: SocketAddr) -> Result { if self.is_server { diff --git a/src/connection/path.rs b/src/connection/path.rs index 71054b5dc..32ec6a5e8 100644 --- a/src/connection/path.rs +++ b/src/connection/path.rs @@ -337,6 +337,25 @@ impl Path { self.active && self.dcid_seq.is_some() } + /// Whether the path looks blackholed: two or more consecutive PTOs fired + /// without any acknowledgment. Schedulers prefer healthy paths and only + /// fall back to unhealthy ones when no healthy path can send — otherwise + /// a dead-but-uncongested path (sparse traffic never fills its cwnd, so + /// srtt stays frozen at last-known-good) gets picked forever and traffic + /// blackholes despite live alternatives. Self-reviving: the first ACK + /// after link recovery resets the PTO count. + pub fn unhealthy(&self) -> bool { + self.recovery.consecutive_pto_count() >= 2 + } + + /// unhealthy(), plus the optional zombie-bufferbloat srtt cutoff (see + /// PathMap::max_srtt). Schedulers use this so a link parking packets for + /// seconds is avoided even though it technically delivers. + pub fn unhealthy_with(&self, max_srtt: Option) -> bool { + self.unhealthy() + || max_srtt.is_some_and(|m| self.recovery.rtt.smoothed_rtt() > m) + } + /// Set the active state of the path pub(crate) fn set_active(&mut self, v: bool) { self.active = v; @@ -418,6 +437,12 @@ pub(crate) struct PathMap { /// Whether it serves as a server. is_server: bool, + + /// Zombie-bufferbloat cutoff: paths whose smoothed RTT exceeds this are + /// treated as unhealthy by the multipath schedulers (a link can be "up" + /// yet park packets for seconds — worse than dead for interactive + /// traffic). None disables the cutoff. + pub(crate) max_srtt: Option, } impl PathMap { @@ -449,6 +474,7 @@ impl PathMap { anti_ampl_factor, is_multipath: false, is_server, + max_srtt: None, } } diff --git a/src/connection/recovery.rs b/src/connection/recovery.rs index 007ba2c8a..f5b256086 100644 --- a/src/connection/recovery.rs +++ b/src/connection/recovery.rs @@ -846,6 +846,14 @@ impl Recovery { self.max_datagram_size = max_datagram_size; } + /// The number of consecutive PTOs fired without any acknowledgment. + /// Resets to zero on every ACK, so it doubles as a self-reviving + /// path-health signal: a path that keeps losing probe packets accumulates + /// PTOs; the first ACK after recovery clears it. + pub(crate) fn consecutive_pto_count(&self) -> usize { + self.pto_count + } + /// Check whether this path can still send packets. pub(crate) fn can_send(&mut self) -> bool { // Check congestion controller diff --git a/src/multipath_scheduler/scheduler_minrtt.rs b/src/multipath_scheduler/scheduler_minrtt.rs index 31fea2cf2..2b6c68be6 100644 --- a/src/multipath_scheduler/scheduler_minrtt.rs +++ b/src/multipath_scheduler/scheduler_minrtt.rs @@ -43,7 +43,9 @@ impl MultipathScheduler for MinRttScheduler { spaces: &mut PacketNumSpaceMap, streams: &mut StreamMap, ) -> Result { + let max_srtt = paths.max_srtt; let mut best = None; + let mut best_unhealthy = None; for (pid, path) in paths.iter_mut() { // Skip the path that is not ready for sending non-probing packets. @@ -51,19 +53,27 @@ impl MultipathScheduler for MinRttScheduler { continue; } - // Select the path with the minimum srtt + // Blackhole-suspect paths (consecutive PTOs without ACKs) are + // tracked separately and only used when no healthy path can send. let srtt = path.recovery.rtt.smoothed_rtt(); - match best { - None => best = Some((pid, srtt)), + let slot = if path.unhealthy_with(max_srtt) { + &mut best_unhealthy + } else { + &mut best + }; + + // Select the path with the minimum srtt + match slot { + None => *slot = Some((pid, srtt)), Some((_, rtt)) => { - if srtt < rtt { - best = Some((pid, srtt)); + if srtt < *rtt { + *slot = Some((pid, srtt)); } } } } - match best { + match best.or(best_unhealthy) { Some((i, _)) => Ok(i), None => Err(Error::Done), } diff --git a/src/multipath_scheduler/scheduler_rr.rs b/src/multipath_scheduler/scheduler_rr.rs index 92c0a883f..8cdf18cc7 100644 --- a/src/multipath_scheduler/scheduler_rr.rs +++ b/src/multipath_scheduler/scheduler_rr.rs @@ -50,12 +50,21 @@ impl RoundRobinScheduler { } /// Try to select an available path - fn select(&mut self, iter: &mut slab::IterMut) -> Option { + fn select( + &mut self, + iter: &mut slab::IterMut, + healthy_only: bool, + max_srtt: Option, + ) -> Option { for (pid, path) in iter.by_ref() { // Skip the path that is not ready for sending non-probing packets. if !path.active() || !path.recovery.can_send() { continue; } + // Blackhole-suspect paths are only used when no healthy path can. + if healthy_only && path.unhealthy_with(max_srtt) { + continue; + } self.last = Some(pid); return Some(pid); @@ -72,30 +81,33 @@ impl MultipathScheduler for RoundRobinScheduler { spaces: &mut PacketNumSpaceMap, streams: &mut StreamMap, ) -> Result { - let mut iter = paths.iter_mut(); - let mut exist_last = false; - - // Iterate and find the last used path - if let Some(last) = self.last { - if self.find_last(&mut iter, last) { - exist_last = true; - } else { - // The last path has been abandoned - iter = paths.iter_mut(); + // First pass considers only healthy paths; the fallback pass accepts + // blackhole-suspect ones so a fully-degraded path set still sends. + let max_srtt = paths.max_srtt; + for healthy_only in [true, false] { + let mut iter = paths.iter_mut(); + let mut exist_last = false; + + // Iterate and find the last used path + if let Some(last) = self.last { + if self.find_last(&mut iter, last) { + exist_last = true; + } else { + // The last path has been abandoned + iter = paths.iter_mut(); + } } - } - // Find the next available path - if let Some(pid) = self.select(&mut iter) { - return Ok(pid); - } - if !exist_last { - return Err(Error::Done); - } - - let mut iter = paths.iter_mut(); - if let Some(pid) = self.select(&mut iter) { - return Ok(pid); + // Find the next available path + if let Some(pid) = self.select(&mut iter, healthy_only, max_srtt) { + return Ok(pid); + } + if exist_last { + let mut iter = paths.iter_mut(); + if let Some(pid) = self.select(&mut iter, healthy_only, max_srtt) { + return Ok(pid); + } + } } Err(Error::Done) }