From 43b3aca8976a83518b630a82f414551c355abf92 Mon Sep 17 00:00:00 2001 From: Aswanth K Date: Sun, 19 Jul 2026 18:41:45 +0530 Subject: [PATCH 1/2] Multipath: path-health aware scheduling and control knobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A path can blackhole while remaining schedulable: with sparse traffic the congestion window never fills and smoothed_rtt stays frozen at its last-known-good value, so RTT-based schedulers keep picking a dead path forever (measured: 10s link blackout = 100% loss with live alternatives available). A path can also be 'zombie': up and delivering, but parking packets in oversized carrier queues for seconds (measured on a real cellular link: srtt up to 37s at only 7% loss) — worse than dead for interactive traffic, and invisible to PTO-based health because ACKs do eventually arrive. - recovery: expose consecutive_pto_count() (resets on any ACK — a self-reviving health signal) - path: Path::unhealthy() (>= 2 consecutive PTOs) and unhealthy_with() adding an optional smoothed-RTT ceiling; PathMap::max_srtt holds the ceiling - schedulers (minrtt, round-robin): two-pass selection — prefer healthy paths, fall back to blackhole suspects only when no healthy path can send, so a fully-degraded path set still transmits - connection: set_multipath_algorithm() switches the scheduler live, set_scheduler_max_rtt() sets/clears the RTT ceiling, path_health() returns per-path (addrs, srtt, consecutive PTOs, active, unhealthy) for external control planes With the keepalive PING API an application can probe all paths periodically; a dead path accumulates PTOs and is avoided within ~1s, and the first ACK after link recovery restores it automatically. (cherry picked from commit 1f9f12349dd62160d076f10e6f1f478f8a9dee85) --- src/connection/connection.rs | 35 +++++++++++++ src/connection/path.rs | 26 +++++++++ src/connection/recovery.rs | 8 +++ src/multipath_scheduler/scheduler_minrtt.rs | 22 +++++--- src/multipath_scheduler/scheduler_rr.rs | 58 +++++++++++++-------- 5 files changed, 120 insertions(+), 29 deletions(-) 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) } From 848506ffa9acf19e880d920cda0e8ed2e39e929c Mon Sep 17 00:00:00 2001 From: Aswanth K Date: Sun, 19 Jul 2026 18:42:31 +0530 Subject: [PATCH 2/2] Multipath: add Tempo, an arrival-time scheduler Tempo selects the path where the next packet would ARRIVE earliest: bytes_in_flight / pacing_rate + srtt/2. Compared to MinRtt it distributes load by available capacity rather than piling onto the lowest-RTT path until its cwnd fills, which reduces the reordering that per-packet striping inflicts on heterogeneous paths while still using every path's capacity. Falls back to cwnd/srtt when the congestion controller does not expose a pacing rate. Honors the same path-health rules as the other schedulers (healthy-first, blackhole suspects only as a last resort). (cherry picked from commit 905210667a3f3c4cf6cbdfe2416abdf51cfe66eb) --- .../multipath_scheduler.rs | 12 +++ src/multipath_scheduler/scheduler_tempo.rs | 83 +++++++++++++++++++ 2 files changed, 95 insertions(+) create mode 100644 src/multipath_scheduler/scheduler_tempo.rs diff --git a/src/multipath_scheduler/multipath_scheduler.rs b/src/multipath_scheduler/multipath_scheduler.rs index ab658a545..3e15895fe 100644 --- a/src/multipath_scheduler/multipath_scheduler.rs +++ b/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 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; diff --git a/src/multipath_scheduler/scheduler_tempo.rs b/src/multipath_scheduler/scheduler_tempo.rs new file mode 100644 index 000000000..67594f567 --- /dev/null +++ b/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 { + 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), + } + } +}