Multipath: add Tempo, an arrival-time scheduler

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/multipath_scheduler.rs

@@ -20,6 +20,7 @@ use std::time::Instant;
 use self::scheduler_minrtt::*;
 use self::scheduler_redundant::*;
 use self::scheduler_rr::*;
+use self::scheduler_tempo::*;
 use crate::connection::path::PathMap;
 use crate::connection::space::PacketNumSpaceMap;
 use crate::connection::space::SentPacket;
@@ -83,6 +84,12 @@ pub enum MultipathAlgorithm {
     /// distribution across all path is equal. It is only used for testing
     /// purposes.
     RoundRobin,
+
+    /// TEMPO: arrival-time (earliest-delivery) scheduler. Picks the path where
+    /// the next packet ARRIVES earliest (bytes_in_flight/pacing_rate + srtt/2),
+    /// distributing load by capacity so a single stream aggregates across
+    /// heterogeneous paths without the reorder-collapse of naive round-robin.
+    Tempo,
 }
 
 impl FromStr for MultipathAlgorithm {
@@ -95,6 +102,8 @@ impl FromStr for MultipathAlgorithm {
             Ok(MultipathAlgorithm::Redundant)
         } else if algor.eq_ignore_ascii_case("roundrobin") {
             Ok(MultipathAlgorithm::RoundRobin)
+        } else if algor.eq_ignore_ascii_case("tempo") {
+            Ok(MultipathAlgorithm::Tempo)
         } else {
             Err(Error::InvalidConfig("unknown".into()))
         }
@@ -107,6 +116,7 @@ pub(crate) fn build_multipath_scheduler(conf: &MultipathConfig) -> Box<dyn Multi
         MultipathAlgorithm::MinRtt => Box::new(MinRttScheduler::new(conf)),
         MultipathAlgorithm::Redundant => Box::new(RedundantScheduler::new(conf)),
         MultipathAlgorithm::RoundRobin => Box::new(RoundRobinScheduler::new(conf)),
+        MultipathAlgorithm::Tempo => Box::new(TempoScheduler::new(conf)),
     }
 }
 
@@ -115,6 +125,7 @@ pub(crate) fn buffer_required(algor: MultipathAlgorithm) -> bool {
         MultipathAlgorithm::MinRtt => false,
         MultipathAlgorithm::Redundant => true,
         MultipathAlgorithm::RoundRobin => false,
+        MultipathAlgorithm::Tempo => false,
     }
 }
 
@@ -206,5 +217,6 @@ pub(crate) mod tests {
 }
 
 mod scheduler_minrtt;
+mod scheduler_tempo;
 mod scheduler_redundant;
 mod scheduler_rr;

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)
     }

src/multipath_scheduler/scheduler_tempo.rs

@@ -0,0 +1,83 @@
+// Copyright (c) 2024 bondq — TEMPO multipath scheduler (added to the TQUIC fork).
+//
+// TEMPO is an arrival-time (earliest-delivery-path-first) scheduler. For each
+// packet it predicts the time the packet would ARRIVE at the receiver on every
+// eligible path and picks the earliest:
+//
+//     arrival_i = bytes_in_flight_i / pacing_rate_i + smoothed_rtt_i / 2
+//
+// The first term is how long the path's current backlog takes to drain at its
+// (BBR-estimated) rate; the second is the one-way propagation delay. Selecting
+// the minimum distributes load in proportion to each path's capacity AND keeps
+// packets arriving near-ordered, so a SINGLE stream aggregates across
+// heterogeneous paths — unlike MinRtt, which concentrates on the lowest-RTT
+// path and never spills while BBR keeps it un-full.
+
+use crate::connection::path::PathMap;
+use crate::connection::space::PacketNumSpaceMap;
+use crate::connection::stream::StreamMap;
+use crate::multipath_scheduler::MultipathScheduler;
+use crate::Error;
+use crate::MultipathConfig;
+use crate::Result;
+
+pub struct TempoScheduler {}
+
+impl TempoScheduler {
+    pub fn new(_conf: &MultipathConfig) -> TempoScheduler {
+        TempoScheduler {}
+    }
+}
+
+impl MultipathScheduler for TempoScheduler {
+    fn on_select(
+        &mut self,
+        paths: &mut PathMap,
+        _spaces: &mut PacketNumSpaceMap,
+        _streams: &mut StreamMap,
+    ) -> Result<usize> {
+        let max_srtt = paths.max_srtt;
+        let mut best: Option<(usize, f64)> = None;
+        let mut best_unhealthy: Option<(usize, f64)> = None;
+        for (pid, path) in paths.iter_mut() {
+            if !path.active() || !path.recovery.can_send() {
+                continue;
+            }
+            let srtt = path.recovery.rtt.smoothed_rtt().as_secs_f64();
+            let owd = srtt / 2.0;
+            let inflight = path.recovery.bytes_in_flight as f64;
+            // rate = BBR pacing-rate estimate (bytes/sec); fall back to
+            // cwnd/srtt when pacing rate is unavailable.
+            let rate = match path.recovery.congestion.pacing_rate() {
+                Some(r) if r > 0 => r as f64,
+                _ => {
+                    let cwnd = path.recovery.congestion.congestion_window() as f64;
+                    if srtt > 0.0 {
+                        (cwnd / srtt).max(1.0e6)
+                    } else {
+                        1.0e8
+                    }
+                }
+            };
+            let arrival = inflight / rate + owd;
+            // Blackhole-suspect paths are only used when no healthy path can.
+            let slot = if path.unhealthy_with(max_srtt) {
+                &mut best_unhealthy
+            } else {
+                &mut best
+            };
+            match slot {
+                None => *slot = Some((pid, arrival)),
+                Some((_, a)) => {
+                    if arrival < *a {
+                        *slot = Some((pid, arrival));
+                    }
+                }
+            }
+        }
+        match best.or(best_unhealthy) {
+            Some((i, _)) => Ok(i),
+            None => Err(Error::Done),
+        }
+    }
+}