Multipath: path-health aware scheduling and control knobs

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<u64>) {
+        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<u64> {
         if self.is_server {

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<Duration>) -> 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<time::Duration>,
 }
 
 impl PathMap {
@@ -449,6 +474,7 @@ impl PathMap {
             anti_ampl_factor,
             is_multipath: false,
             is_server,
+            max_srtt: None,
         }
     }
 

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

src/multipath_scheduler/scheduler_minrtt.rs

@@ -43,27 +43,37 @@ impl MultipathScheduler for MinRttScheduler {
         spaces: &mut PacketNumSpaceMap,
         streams: &mut StreamMap,
     ) -> Result<usize> {
+        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.
             if !path.active() || !path.recovery.can_send() {
                 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),
         }

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<Path>) -> Option<usize> {
+    fn select(
+        &mut self,
+        iter: &mut slab::IterMut<Path>,
+        healthy_only: bool,
+        max_srtt: Option<std::time::Duration>,
+    ) -> Option<usize> {
         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<usize> {
-        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)
     }