From ca2d033a9a42394af1bac75a43dc5319d46a5650 Mon Sep 17 00:00:00 2001 From: Simon Date: Thu, 1 Oct 2026 04:34:09 +0800 Subject: [PATCH] feat(classic): per-link weights from the IPs file (Moblin connection priorities) Each IPs-file line is `[ ]`: integer 1..10, missing = 1, larger values clamped to 10, a 0 or unparsable weight is warned about and read as 1 (the IP still counts, so the SIGHUP reload guard is unchanged). Weights are normalised on every load so the lowest link is 1 (priority - lowest + 1, as Moblin's updateConnectionPriorities). Classic selection multiplies get_score() by the link's weight while its window is above 20 000, fades the multiplier linearly to 1 between 20 000 and 10 000, and ignores it below (Moblin's RemoteConnection.score()). A weight of 1 is exactly neutral, so an unweighted file schedules as before. get_score() and enhanced mode are untouched. Observability: srtla_send_link_weight gauge per link, weight in the stats topic, in the 'added uplink' line, on reload re-weights and in the SIGHUP queued line. Co-Authored-By: Claude Fable 5.1 --- README.md | 8 +- crates/srtla-core/src/connection/mod.rs | 68 ++++++ crates/srtla-core/src/selection/classic.rs | 18 +- crates/srtla-core/src/test_helpers.rs | 1 + src/metrics.rs | 16 ++ src/sender/connections.rs | 44 +++- src/sender/mod.rs | 60 +++-- src/sender/reload.rs | 140 +++++++++++- src/stats.rs | 4 + src/tests/link_weight_tests.rs | 242 +++++++++++++++++++++ src/tests/mod.rs | 3 + src/tests/sender_tests.rs | 4 +- 12 files changed, 581 insertions(+), 27 deletions(-) create mode 100644 src/tests/link_weight_tests.rs diff --git a/README.md b/README.md index 848cc70a..b2502f56 100644 --- a/README.md +++ b/README.md @@ -127,7 +127,13 @@ srtla_send [OPTIONS] SRT_LISTEN_PORT SRTLA_HOST SRTLA_PORT BIND_IPS_FILE - `SRT_LISTEN_PORT`: UDP port on which to receive SRT packets locally - `SRTLA_HOST`: hostname or IP of the SRTLA receiver (e.g., srtla_rec) - `SRTLA_PORT`: UDP port of the SRTLA receiver -- `BIND_IPS_FILE`: path to a file with newline-separated local source IPs (uplinks) +- `BIND_IPS_FILE`: path to a file with newline-separated local source IPs (uplinks). Each line is `[ ]`: the optional + weight (integer 1..10, default 1) is a link priority in the style of Moblin's + "connection priorities". Weights are normalised so the lowest link is 1; in + classic mode a link's score is multiplied by its weight while its window is + above 20 000, the multiplier fades linearly to 1 between 20 000 and 10 000, + and is ignored below. Enhanced mode ignores weights. Exported per link as + `srtla_send_link_weight`. ### Options diff --git a/crates/srtla-core/src/connection/mod.rs b/crates/srtla-core/src/connection/mod.rs index 20cd5fb2..b3399773 100644 --- a/crates/srtla-core/src/connection/mod.rs +++ b/crates/srtla-core/src/connection/mod.rs @@ -125,6 +125,61 @@ impl Default for CachedQuality { } } +/// Lowest operator link weight (the IPs file's optional second column). +/// Weights are Moblin's "connection priorities": an integer 1..10. +pub const LINK_WEIGHT_MIN: u8 = 1; +/// Highest operator link weight. +pub const LINK_WEIGHT_MAX: u8 = 10; +/// Above this window a weighted link gets its full weight (Moblin's +/// `windowStableMaximum * windowMultiply`, = `WINDOW_DEF * WINDOW_MULT`). +pub const LINK_WEIGHT_FULL_ABOVE_WINDOW: i32 = 20 * WINDOW_MULT; +/// At or below this window the weight is ignored (Moblin's +/// `windowStableMinimum * windowMultiply`). +pub const LINK_WEIGHT_NONE_AT_OR_BELOW_WINDOW: i32 = 10 * WINDOW_MULT; + +/// Score multiplier for a link with operator weight `weight` and congestion +/// window `window`, copied from Moblin's `RemoteConnection.score()`: +/// +/// - `window > 20 000` (healthy, room to spare): the full weight; +/// - `10 000 < window <= 20 000` (the link is shedding): fades linearly from +/// the weight down to 1; +/// - `window <= 10 000`: 1 (the weight is ignored, plain capacity balancing). +/// +/// A weight of 1 (the default for every link) is exactly 1.0 in every band, so +/// an unweighted IPs file schedules byte-for-byte as before. Used by classic +/// selection only. +#[inline] +pub fn link_weight_multiplier(window: i32, weight: u8) -> f32 { + let weight = f32::from(weight.clamp(LINK_WEIGHT_MIN, LINK_WEIGHT_MAX)); + if window > LINK_WEIGHT_FULL_ABOVE_WINDOW { + weight + } else if window > LINK_WEIGHT_NONE_AT_OR_BELOW_WINDOW { + let factor = (window - LINK_WEIGHT_NONE_AT_OR_BELOW_WINDOW) as f32 + / (LINK_WEIGHT_FULL_ABOVE_WINDOW - LINK_WEIGHT_NONE_AT_OR_BELOW_WINDOW) as f32; + 1.0 + (weight - 1.0) * factor + } else { + 1.0 + } +} + +/// Normalise operator weights so the lowest one is 1, keeping the differences +/// (Moblin's `updateConnectionPriorities`: `priority - lowest + 1`). Only the +/// ratio between links matters to the scheduler, and this keeps a file that +/// says `5`/`5` from behaving any differently from one that says nothing. +/// Inputs are clamped to `LINK_WEIGHT_MIN..=LINK_WEIGHT_MAX` first, so the +/// result stays in that range. +pub fn normalise_link_weights(weights: &mut [u8]) { + for w in weights.iter_mut() { + *w = (*w).clamp(LINK_WEIGHT_MIN, LINK_WEIGHT_MAX); + } + let Some(lowest) = weights.iter().copied().min() else { + return; + }; + for w in weights.iter_mut() { + *w = *w - lowest + 1; + } +} + /// Share of its natural score a link should compete with while ramping back /// in after a stall gate, rising linearly from /// [`crate::config_snapshot::STALL_REJOIN_RAMP_FLOOR`] to `1.0` over `ramp_ms`. @@ -447,6 +502,11 @@ pub struct SrtlaConnection { /// `is_timed_out`/`CONN_TIMEOUT`); a gated link keeps a trickle of /// traffic so the loss EWMA can recover and clear the latch. pub loss_degraded: bool, + /// Operator link weight from the IPs file (` `), normalised so + /// the lowest link is 1. Scales this link's classic score through + /// [`link_weight_multiplier`]. Survives recovery/reconnect resets: it is + /// configuration, not link state; only a reload changes it. + pub link_weight: u8, } impl SrtlaConnection { @@ -505,9 +565,17 @@ impl SrtlaConnection { cc_backing_off: false, cc_target_bps: 0, loss_degraded: false, + link_weight: LINK_WEIGHT_MIN, } } + /// Classic-mode score multiplier from this link's operator weight at its + /// current window (see [`link_weight_multiplier`]). + #[inline(always)] + pub fn weight_multiplier(&self) -> f32 { + link_weight_multiplier(self.window, self.link_weight) + } + #[inline(always)] pub fn get_score(&self) -> i32 { if !self.connected { diff --git a/crates/srtla-core/src/selection/classic.rs b/crates/srtla-core/src/selection/classic.rs index ccdc2501..ad6223d5 100644 --- a/crates/srtla-core/src/selection/classic.rs +++ b/crates/srtla-core/src/selection/classic.rs @@ -8,6 +8,10 @@ //! - No quality awareness (no NAK penalties) //! - No RTT consideration //! - Simple "pick highest score" algorithm +//! - Optional operator link weights (Moblin's connection priorities): the score +//! is multiplied by [`SrtlaConnection::weight_multiplier`], which is exactly +//! 1.0 for an unweighted link, so an IPs file without weights schedules as +//! the C implementation does. use crate::connection::SrtlaConnection; @@ -29,7 +33,7 @@ pub fn select_connection(conns: &[SrtlaConnection], now_ms: u64) -> Option best_score { best_score = score; best_idx = Some(i); @@ -39,4 +43,16 @@ pub fn select_connection(conns: &[SrtlaConnection], now_ms: u64) -> Option i32 { + let score = c.get_score(); + if c.link_weight <= 1 || score <= 0 { + return score; + } + (score as f32 * c.weight_multiplier()) as i32 +} + // Tests are in src/tests/sender_tests.rs diff --git a/crates/srtla-core/src/test_helpers.rs b/crates/srtla-core/src/test_helpers.rs index 69b5e0f5..d03058e0 100644 --- a/crates/srtla-core/src/test_helpers.rs +++ b/crates/srtla-core/src/test_helpers.rs @@ -75,6 +75,7 @@ fn build_connection(local_ip: IpAddr, label: String) -> SrtlaConnection { cc_backing_off: false, cc_target_bps: 0, loss_degraded: false, + link_weight: 1, } } diff --git a/src/metrics.rs b/src/metrics.rs index 343e285a..93fb0d37 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -238,6 +238,22 @@ pub fn render(stats: &SharedStats, config: &DynamicConfig, cw: &CriticalWindow) .ok(); } + writeln!( + out, + "# HELP srtla_send_link_weight operator link weight from the ips file, normalised so the \ + lowest link is 1 (classic mode only)" + ) + .ok(); + writeln!(out, "# TYPE srtla_send_link_weight gauge").ok(); + for link in &snap.links { + writeln!( + out, + r#"srtla_send_link_weight{{ip="{}"}} {}"#, + link.ip, link.weight + ) + .ok(); + } + // Aggregate gauges. writeln!( out, diff --git a/src/sender/connections.rs b/src/sender/connections.rs index d5a43b0a..a941a30d 100644 --- a/src/sender/connections.rs +++ b/src/sender/connections.rs @@ -17,6 +17,9 @@ use crate::net::{ pub struct PendingConnectionChanges { pub new_ips: Option>, + /// Normalised operator weight per entry of `new_ips` (same order). Empty = + /// every link weight 1. + pub new_weights: SmallVec, pub receiver_host: String, pub receiver_port: u16, } @@ -26,6 +29,7 @@ pub async fn apply_connection_changes( connections: &mut SmallVec, conn_io: &mut ConnIoMap, new_ips: &[IpAddr], + new_weights: &[u8], receiver_host: &str, receiver_port: u16, last_selected_idx: &mut Option, @@ -62,9 +66,12 @@ pub async fn apply_connection_changes( } } + // Weight changes on the links that stay (a reload may only re-weight). + apply_link_weights(connections, new_ips, new_weights); + // Add new connections let mut seen = HashSet::::new(); - let new_ips_needed: SmallVec = new_ips + let (new_ips_needed, new_weights_needed): (SmallVec, SmallVec) = new_ips .iter() .copied() .filter(|ip| seen.insert(*ip)) @@ -72,11 +79,13 @@ pub async fn apply_connection_changes( let label = format!("{}:{} via {}", receiver_host, receiver_port, ip); !current_labels.contains(&label) }) - .collect(); + .map(|ip| (ip, weight_for(new_ips, new_weights, ip))) + .unzip(); if !new_ips_needed.is_empty() { let mut new_connections = create_connections_from_ips( &new_ips_needed, + &new_weights_needed, receiver_host, receiver_port, binder, @@ -97,8 +106,34 @@ pub async fn apply_connection_changes( } } +/// Operator weight for `ip` from the parallel `ips`/`weights` lists (first +/// match wins); 1 when the list carries no weight for it. +pub fn weight_for(ips: &[IpAddr], weights: &[u8], ip: IpAddr) -> u8 { + ips.iter() + .position(|candidate| *candidate == ip) + .and_then(|i| weights.get(i).copied()) + .unwrap_or(srtla_core::connection::LINK_WEIGHT_MIN) +} + +/// Set every existing connection's operator weight from the parallel +/// `ips`/`weights` lists, logging each change. A link the lists do not name +/// goes back to 1. +pub fn apply_link_weights(connections: &mut [SrtlaConnection], ips: &[IpAddr], weights: &[u8]) { + for conn in connections.iter_mut() { + let weight = weight_for(ips, weights, conn.local_ip); + if conn.link_weight != weight { + info!( + "uplink {} weight {} -> {}", + conn.label, conn.link_weight, weight + ); + conn.link_weight = weight; + } + } +} + pub async fn create_connections_from_ips( ips: &[IpAddr], + weights: &[u8], receiver_host: &str, receiver_port: u16, binder: &Arc, @@ -107,8 +142,9 @@ pub async fn create_connections_from_ips( let mut connections = SmallVec::new(); for ip in ips { match connect_uplink(*ip, receiver_host, receiver_port, binder).await { - Ok((conn, io)) => { - info!("added uplink {}", conn.label); + Ok((mut conn, io)) => { + conn.link_weight = weight_for(ips, weights, *ip); + info!("added uplink {} (weight {})", conn.label, conn.link_weight); conn_io.insert(conn.conn_id, io); connections.push(conn); } diff --git a/src/sender/mod.rs b/src/sender/mod.rs index 087f08f2..7515637b 100644 --- a/src/sender/mod.rs +++ b/src/sender/mod.rs @@ -19,8 +19,8 @@ pub(crate) use client_dedup::ClientDedup; // Re-export connection management functions for tests #[allow(unused_imports)] pub use connections::{ - PendingConnectionChanges, apply_connection_changes, create_connections_from_ips, - recover_connection, + PendingConnectionChanges, apply_connection_changes, apply_link_weights, + create_connections_from_ips, recover_connection, }; // Re-export public items used by tests #[allow(unused_imports)] @@ -104,14 +104,8 @@ pub async fn run_sender_with_config( .context("bind local SRT UDP listener")?; info!("listening for SRT on [::]:{}", local_srt_port); - let ips = read_ip_list(ips_file).await?; - debug!( - "uplink IPs loaded: {}", - ips.iter() - .map(|i| i.to_string()) - .collect::>() - .join(", ") - ); + let (ips, weights) = read_weighted_ip_list(ips_file).await?; + debug!("uplink IPs loaded: {}", describe_weights(&ips, &weights)); if ips.is_empty() { return Err(anyhow!("no IPs in list: {}", ips_file)); } @@ -119,9 +113,15 @@ pub async fn run_sender_with_config( // Shell-owned I/O half of every connection, keyed by conn_id (never by // index — so no lockstep with the connections vec through add/remove). let mut conn_io: ConnIoMap = std::collections::HashMap::new(); - let mut connections = - create_connections_from_ips(&ips, receiver_host, receiver_port, &binder, &mut conn_io) - .await; + let mut connections = create_connections_from_ips( + &ips, + &weights, + receiver_host, + receiver_port, + &binder, + &mut conn_io, + ) + .await; if connections.is_empty() { return Err(anyhow!("no uplinks available")); } @@ -365,6 +365,7 @@ pub async fn run_sender_with_config( &mut connections, &mut conn_io, &new_ips, + &changes.new_weights, &changes.receiver_host, changes.receiver_port, &mut last_selected_idx, @@ -416,18 +417,22 @@ pub async fn run_sender_with_config( // up rather than queuing an empty list, which would tear down every // connection in apply_connection_changes. Mirrors the C sender. match reload::analyze_ip_reload(ips_file) { - reload::IpReload::Apply { ips, first_invalid_line } => { + reload::IpReload::Apply { ips, weights, first_invalid_line } => { if let Some(line) = first_invalid_line { warn!( "ips file has an invalid entry starting at line {line}; applying valid IPs only" ); } + info!( + "uplink IP changes queued for next processing cycle: {}", + describe_weights(&ips, &weights) + ); pending_changes = Some(PendingConnectionChanges { new_ips: Some(ips), + new_weights: weights, receiver_host: receiver_host.to_string(), receiver_port, }); - info!("uplink IP changes queued for next processing cycle"); } reload::IpReload::Refuse(reason) => { warn!( @@ -466,6 +471,26 @@ pub async fn run_sender_with_config( } pub async fn read_ip_list(path: &str) -> Result> { + Ok(read_weighted_ip_list(path).await?.0) +} + +/// `ip weight` pairs for a log line, e.g. `192.168.0.15 (weight 10), 192.168.0.2 +/// (weight 1)`. +fn describe_weights(ips: &[IpAddr], weights: &[u8]) -> String { + ips.iter() + .map(|ip| { + format!( + "{ip} (weight {})", + connections::weight_for(ips, weights, *ip) + ) + }) + .collect::>() + .join(", ") +} + +/// [`read_ip_list`] plus the normalised operator weight of each IP (same +/// order). An unweighted file yields all 1s. +pub async fn read_weighted_ip_list(path: &str) -> Result<(SmallVec, SmallVec)> { let text = std::fs::read_to_string(Path::new(path)).context("read IPs file")?; // Shares the SIGHUP reload guard's parser so startup and reload agree on what // counts as a valid IP. At startup an empty or all-invalid file is tolerated @@ -474,13 +499,14 @@ pub async fn read_ip_list(path: &str) -> Result> { match reload::analyze_ip_reload_text(&text) { reload::IpReload::Apply { ips, + weights, first_invalid_line, } => { if let Some(line) = first_invalid_line { warn!("ips file has an invalid entry starting at line {line}; skipping it"); } - Ok(ips) + Ok((ips, weights)) } - reload::IpReload::Refuse(_) => Ok(SmallVec::new()), + reload::IpReload::Refuse(_) => Ok((SmallVec::new(), SmallVec::new())), } } diff --git a/src/sender/reload.rs b/src/sender/reload.rs index e7500f6b..da525140 100644 --- a/src/sender/reload.rs +++ b/src/sender/reload.rs @@ -6,11 +6,21 @@ //! all-garbage file — is REFUSED so the stream keeps running on the existing //! links instead of tearing every connection down. A file mixing valid and //! invalid lines still applies; the bad lines are skipped with a warning. +//! +//! Each line is `[ ]`. The optional weight is an operator link +//! weight (Moblin's "connection priorities"): an integer 1..10, missing = 1, +//! larger values clamped to 10, and a `0` or unparsable weight is warned about +//! and read as 1 — the IP on that line still counts, so a weight can never make +//! the reload guard refuse a file it used to accept. Weights are normalised so +//! the lowest one is 1 on every load (see +//! [`srtla_core::connection::normalise_link_weights`]). use std::net::IpAddr; use std::str::FromStr; use smallvec::SmallVec; +use srtla_core::connection::{LINK_WEIGHT_MAX, LINK_WEIGHT_MIN, normalise_link_weights}; +use tracing::warn; /// Why a SIGHUP reload was refused. In every case the existing connections are /// kept and the stream keeps running. @@ -36,6 +46,9 @@ pub enum IpReload { /// valid+invalid file), otherwise `None`. Apply { ips: SmallVec, + /// Normalised operator weight per entry of `ips` (same order, same + /// length). All 1 when the file carries no weights. + weights: SmallVec, first_invalid_line: Option, }, /// Refuse the reload and keep the current connections. @@ -48,6 +61,7 @@ pub enum IpReload { /// the file read on top. pub fn analyze_ip_reload_text(text: &str) -> IpReload { let mut ips: SmallVec = SmallVec::new(); + let mut weights: SmallVec = SmallVec::new(); let mut first_invalid_line: Option = None; let mut saw_content = false; @@ -57,15 +71,19 @@ pub fn analyze_ip_reload_text(text: &str) -> IpReload { continue; } saw_content = true; - match IpAddr::from_str(trimmed) { - Ok(ip) => ips.push(ip), - Err(_) => { + match parse_ip_line(trimmed, idx + 1) { + Some((ip, weight)) => { + ips.push(ip); + weights.push(weight); + } + None => { if first_invalid_line.is_none() { first_invalid_line = Some(idx + 1); } } } } + normalise_link_weights(&mut weights); if ips.is_empty() { return if saw_content { @@ -79,10 +97,43 @@ pub fn analyze_ip_reload_text(text: &str) -> IpReload { IpReload::Apply { ips, + weights, first_invalid_line, } } +/// Parse one non-blank line as `[ ]`. `None` = the line has no +/// valid IP (or more than two fields) and is skipped like any invalid line. A +/// bad weight never invalidates the line: it is warned about and read as 1. +fn parse_ip_line(line: &str, line_no: usize) -> Option<(IpAddr, u8)> { + let mut fields = line.split_whitespace(); + let ip = IpAddr::from_str(fields.next()?).ok()?; + let weight = match fields.next() { + None => LINK_WEIGHT_MIN, + Some(raw) => match raw.parse::() { + Ok(0) | Err(_) => { + warn!( + "ips file line {line_no}: weight {raw:?} for {ip} is not an integer \ + {LINK_WEIGHT_MIN}..{LINK_WEIGHT_MAX}; using {LINK_WEIGHT_MIN}" + ); + LINK_WEIGHT_MIN + } + Ok(w) if w > u64::from(LINK_WEIGHT_MAX) => { + warn!( + "ips file line {line_no}: weight {w} for {ip} is above {LINK_WEIGHT_MAX}; \ + clamping to {LINK_WEIGHT_MAX}" + ); + LINK_WEIGHT_MAX + } + Ok(w) => w as u8, + }, + }; + if fields.next().is_some() { + return None; + } + Some((ip, weight)) +} + /// Read `path` and analyze it for a SIGHUP reload. A read error maps to /// [`ReloadRefusal::NotFound`] — the C guard treats an unreadable file as zero /// valid IPs and refuses the reload. @@ -122,6 +173,7 @@ mod tests { IpReload::Apply { ips, first_invalid_line, + .. } => { assert_eq!(ips.as_slice(), [ip("10.0.0.1"), ip("10.0.0.2")]); assert_eq!(first_invalid_line, None); @@ -136,6 +188,7 @@ mod tests { IpReload::Apply { ips, first_invalid_line, + .. } => { assert_eq!(ips.as_slice(), [ip("10.0.0.1"), ip("10.0.0.2")]); assert_eq!(first_invalid_line, None); @@ -151,6 +204,7 @@ mod tests { IpReload::Apply { ips, first_invalid_line, + .. } => { assert_eq!(ips.as_slice(), [ip("10.0.0.1"), ip("10.0.0.2")]); assert_eq!(first_invalid_line, Some(2)); @@ -225,4 +279,84 @@ mod tests { other => panic!("expected Apply, got {other:?}"), } } + + fn apply(text: &str) -> (SmallVec, SmallVec, Option) { + match analyze_ip_reload_text(text) { + IpReload::Apply { + ips, + weights, + first_invalid_line, + } => (ips, weights, first_invalid_line), + other => panic!("expected Apply, got {other:?}"), + } + } + + #[test] + fn unweighted_file_gets_weight_one_everywhere() { + let (ips, weights, bad) = apply("10.0.0.1\n10.0.0.2\n"); + assert_eq!(ips.as_slice(), [ip("10.0.0.1"), ip("10.0.0.2")]); + assert_eq!(weights.as_slice(), [1, 1]); + assert_eq!(bad, None); + } + + #[test] + fn weight_column_is_parsed_and_missing_is_one() { + let (ips, weights, bad) = apply("192.168.0.15 10\n192.168.0.2\n"); + assert_eq!(ips.as_slice(), [ip("192.168.0.15"), ip("192.168.0.2")]); + assert_eq!(weights.as_slice(), [10, 1]); + assert_eq!(bad, None); + } + + #[test] + fn weights_are_normalised_so_the_lowest_is_one() { + // Moblin: priority - lowest + 1. + let (_, weights, _) = apply("10.0.0.1 10\n10.0.0.2 5\n10.0.0.3 7\n"); + assert_eq!(weights.as_slice(), [6, 1, 3]); + // Equal weights are no preference at all. + let (_, weights, _) = apply("10.0.0.1 5\n10.0.0.2 5\n"); + assert_eq!(weights.as_slice(), [1, 1]); + // A single link is always 1. + let (_, weights, _) = apply("10.0.0.1 9\n"); + assert_eq!(weights.as_slice(), [1]); + } + + #[test] + fn out_of_range_and_bad_weights_keep_the_ip() { + // 42 clamps to 10; 0, -3 and "x" read as 1 but the IPs still count. + let (ips, weights, bad) = apply("10.0.0.1 42\n10.0.0.2 0\n10.0.0.3 -3\n10.0.0.4 x\n"); + assert_eq!(ips.len(), 4); + assert_eq!(weights.as_slice(), [10, 1, 1, 1]); + assert_eq!(bad, None); + } + + #[test] + fn reload_guard_counts_weighted_lines_as_valid() { + // A file whose only line carries a (bad) weight is still a valid reload. + let (ips, weights, _) = apply("10.0.0.1 0\n"); + assert_eq!(ips.as_slice(), [ip("10.0.0.1")]); + assert_eq!(weights.as_slice(), [1]); + // Garbage IP with a weight is still garbage, and three fields are invalid. + assert_eq!( + analyze_ip_reload_text("not-an-ip 10\n10.0.0.1 10 extra\n"), + IpReload::Refuse(ReloadRefusal::NoValidIps { + first_invalid_line: 1, + }) + ); + } + + #[cfg(unix)] + #[test] + fn reads_weights_from_a_real_file() { + let mut f = NamedTempFile::new().unwrap(); + writeln!(f, "127.0.0.1 10").unwrap(); + writeln!(f, "127.0.0.2").unwrap(); + f.flush().unwrap(); + match analyze_ip_reload(f.path().to_str().unwrap()) { + IpReload::Apply { ips, weights, .. } => { + assert_eq!(ips.len(), 2); + assert_eq!(weights.as_slice(), [10, 1]); + } + other => panic!("expected Apply, got {other:?}"), + } + } } diff --git a/src/stats.rs b/src/stats.rs index 69522849..8408fd6b 100644 --- a/src/stats.rs +++ b/src/stats.rs @@ -158,6 +158,9 @@ pub struct LinkStats { pub stall_gated: bool, /// Cumulative stall-latch engagements since the link was created. pub stall_gate_events: u64, + /// Normalised operator link weight from the IPs file (1 = unweighted). + /// Applies in classic mode only. + pub weight: u8, /// Cumulative fast silence-pull engagements (a loaded link heard nothing /// for ~2 RTTs and was transiently skipped). Ticks on routine cellular /// HARQ stalls — rate telemetry, not an alarm; compare against @@ -383,6 +386,7 @@ impl SharedStats { stall_gated: conn.stall_latched(), stall_gate_events: conn.stall_gate_events(), silence_pulls: conn.silence_pulls(), + weight: conn.link_weight, sole_carrier: conn.is_sole_carrier(), sole_carrier_excluded: conn.is_sole_carrier_excluded(), sole_carrier_elections: conn.sole_carrier_elections(), diff --git a/src/tests/link_weight_tests.rs b/src/tests/link_weight_tests.rs new file mode 100644 index 00000000..0a25a77d --- /dev/null +++ b/src/tests/link_weight_tests.rs @@ -0,0 +1,242 @@ +//! Tests for operator link weights (Moblin's "connection priorities"): the +//! optional second column of the IPs file, normalised so the lowest link is 1, +//! applied by classic selection only through a three-band window multiplier. +//! Parsing and the reload guard are covered in `sender::reload`'s own tests. + +#[cfg(test)] +mod tests { + use std::net::{IpAddr, Ipv4Addr}; + + use srtla_core::connection::{ + LINK_WEIGHT_FULL_ABOVE_WINDOW, LINK_WEIGHT_NONE_AT_OR_BELOW_WINDOW, link_weight_multiplier, + normalise_link_weights, + }; + use srtla_core::mode::SchedulingMode; + use srtla_core::selection::select_connection_idx; + use srtla_core::utils::now_ms; + + use crate::config::{ConfigSnapshot, STALL_ACK_STALE_MS, STALL_MIN_IN_FLIGHT_PACKETS}; + use crate::sender::apply_link_weights; + use crate::stats::SharedStats; + use crate::test_helpers::create_test_connections; + + fn classic() -> ConfigSnapshot { + ConfigSnapshot { + mode: SchedulingMode::Classic, + quality_enabled: false, + ..ConfigSnapshot::default() + } + } + + /// Feed `packets` packets through classic selection, counting each pick as + /// one more in-flight packet on the chosen link (no ACKs): the share each + /// link ends up with is the scheduler's steady preference at these windows. + fn picks(conns: &mut [srtla_core::connection::SrtlaConnection], packets: usize) -> Vec { + let config = classic(); + let mut counts = vec![0usize; conns.len()]; + for _ in 0..packets { + let now = now_ms(); + let idx = select_connection_idx(conns, None, now, &config).expect("a link"); + conns[idx].in_flight_packets += 1; + counts[idx] += 1; + } + counts + } + + #[test] + fn multiplier_full_weight_above_20k() { + assert_eq!(link_weight_multiplier(60_000, 10), 10.0); + assert_eq!(link_weight_multiplier(20_001, 10), 10.0); + assert_eq!( + link_weight_multiplier(LINK_WEIGHT_FULL_ABOVE_WINDOW, 10), + 10.0 + ); + } + + #[test] + fn multiplier_fades_linearly_between_10k_and_20k() { + assert_eq!(link_weight_multiplier(15_000, 10), 5.5); + assert_eq!(link_weight_multiplier(12_500, 5), 2.0); + let just_above = link_weight_multiplier(LINK_WEIGHT_NONE_AT_OR_BELOW_WINDOW + 1, 10); + assert!(just_above > 1.0 && just_above < 1.01, "{just_above}"); + } + + #[test] + fn multiplier_ignores_weight_at_or_below_10k() { + assert_eq!( + link_weight_multiplier(LINK_WEIGHT_NONE_AT_OR_BELOW_WINDOW, 10), + 1.0 + ); + assert_eq!(link_weight_multiplier(1_000, 10), 1.0); + } + + #[test] + fn weight_one_is_neutral_in_every_band() { + for window in [60_000, 20_000, 15_000, 10_000, 1] { + assert_eq!(link_weight_multiplier(window, 1), 1.0); + } + } + + #[test] + fn normalisation_makes_the_lowest_weight_one() { + let mut w = [10u8, 1]; + normalise_link_weights(&mut w); + assert_eq!(w, [10, 1]); + let mut w = [10u8, 5, 7]; + normalise_link_weights(&mut w); + assert_eq!(w, [6, 1, 3]); + let mut w = [4u8, 4]; + normalise_link_weights(&mut w); + assert_eq!(w, [1, 1]); + let mut w = [0u8, 200]; + normalise_link_weights(&mut w); + assert_eq!(w, [1, 10], "clamped to 1..10 before normalising"); + let mut w: [u8; 0] = []; + normalise_link_weights(&mut w); + } + + #[test] + fn unweighted_links_share_evenly() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut conns = rt.block_on(create_test_connections(2)); + let counts = picks(&mut conns, 200); + assert_eq!(counts, vec![100, 100]); + } + + #[test] + fn classic_prefers_the_weighted_link_while_healthy() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut conns = rt.block_on(create_test_connections(2)); + for c in conns.iter_mut() { + c.window = 40_000; // healthy band + } + conns[0].link_weight = 10; + let counts = picks(&mut conns, 220); + let share_b = counts[1] as f64 / 220.0; + assert!(counts[0] > counts[1], "{counts:?}"); + assert!( + share_b > 0.0 && share_b < 0.12, + "weight 10 leaves roughly one packet in ten to the other link: {counts:?}" + ); + } + + #[test] + fn preference_fades_as_the_weighted_links_window_shrinks() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let share_b_at = |window_a: i32| { + let mut conns = rt.block_on(create_test_connections(2)); + conns[0].window = window_a; + conns[1].window = window_a; + conns[0].link_weight = 10; + let counts = picks(&mut conns, 400); + counts[1] as f64 / 400.0 + }; + let healthy = share_b_at(40_000); + let shedding = share_b_at(15_000); + let congested = share_b_at(10_000); + assert!(healthy < shedding, "{healthy} {shedding}"); + assert!(shedding < congested, "{shedding} {congested}"); + assert!( + (congested - 0.5).abs() < 0.01, + "at or below 10k the weight is ignored: {congested}" + ); + } + + #[test] + fn a_small_window_on_the_weighted_link_hands_traffic_to_the_other() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut conns = rt.block_on(create_test_connections(2)); + conns[0].link_weight = 10; + conns[0].window = 5_000; // A is congested: no multiplier + conns[1].window = 40_000; // B healthy, weight 1 + let counts = picks(&mut conns, 90); + assert!(counts[1] > counts[0] * 4, "{counts:?}"); + } + + #[test] + fn timed_out_weighted_link_is_still_skipped() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut conns = rt.block_on(create_test_connections(2)); + let now = now_ms(); + conns[0].link_weight = 10; + conns[0].window = 60_000; + conns[0].last_received = Some(now.saturating_sub(60_000)); + conns[1].in_flight_packets = 30; + let selected = select_connection_idx(&mut conns, None, now, &classic()); + assert_eq!(selected, Some(1)); + } + + #[test] + fn stall_gated_weighted_link_is_still_skipped() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut conns = rt.block_on(create_test_connections(2)); + let now = now_ms(); + conns[0].link_weight = 10; + conns[0].in_flight_packets = STALL_MIN_IN_FLIGHT_PACKETS; + conns[0].last_ack_or_rtt_sample_ms = now.saturating_sub(STALL_ACK_STALE_MS + 1000); + conns[1].in_flight_packets = STALL_MIN_IN_FLIGHT_PACKETS * 2; + conns[1].last_ack_or_rtt_sample_ms = now; + let selected = select_connection_idx(&mut conns, None, now, &classic()); + assert_eq!(selected, Some(1)); + assert!(conns[0].stall_gated); + } + + #[test] + fn enhanced_mode_ignores_weights() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut conns = rt.block_on(create_test_connections(2)); + conns[0].link_weight = 10; + conns[0].in_flight_packets = 5; + conns[1].in_flight_packets = 0; + let config = ConfigSnapshot { + mode: SchedulingMode::Enhanced, + quality_enabled: false, + ..ConfigSnapshot::default() + }; + let selected = select_connection_idx(&mut conns, None, now_ms(), &config); + assert_eq!(selected, Some(1)); + } + + #[test] + fn reload_reweights_existing_links_and_unnamed_links_fall_back_to_one() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut conns = rt.block_on(create_test_connections(2)); + let a = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 10)); + let b = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 11)); + assert_eq!(conns[0].local_ip, a); + + apply_link_weights(&mut conns, &[a, b], &[10, 1]); + assert_eq!((conns[0].link_weight, conns[1].link_weight), (10, 1)); + + // A reload without weights puts everyone back to 1. + apply_link_weights(&mut conns, &[a, b], &[1, 1]); + assert_eq!((conns[0].link_weight, conns[1].link_weight), (1, 1)); + + // Weight moves to the other link. + apply_link_weights(&mut conns, &[a, b], &[1, 7]); + assert_eq!((conns[0].link_weight, conns[1].link_weight), (1, 7)); + + // An empty weight list (legacy caller) means 1. + apply_link_weights(&mut conns, &[a, b], &[]); + assert_eq!((conns[0].link_weight, conns[1].link_weight), (1, 1)); + } + + #[test] + fn weight_is_exported_per_link_in_metrics() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let mut conns = rt.block_on(create_test_connections(2)); + conns[0].link_weight = 10; + let stats = SharedStats::new(); + stats.update(&conns, &classic(), None, None); + let text = crate::metrics::render( + &stats, + &crate::config::DynamicConfig::new(), + &srtla_core::priority::CriticalWindow::new(), + ); + assert!( + text.contains(r#"srtla_send_link_weight{ip="192.168.1.10"} 10"#), + "{text}" + ); + assert!(text.contains(r#"srtla_send_link_weight{ip="192.168.1.11"} 1"#)); + } +} diff --git a/src/tests/mod.rs b/src/tests/mod.rs index 00924020..4fedf88b 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -33,3 +33,6 @@ pub mod recovery_limbo_tests; #[cfg(test)] pub mod client_forward_dedup_tests; + +#[cfg(test)] +pub mod link_weight_tests; diff --git a/src/tests/sender_tests.rs b/src/tests/sender_tests.rs index 978db4e1..9d65cd63 100644 --- a/src/tests/sender_tests.rs +++ b/src/tests/sender_tests.rs @@ -503,6 +503,7 @@ mod tests { &mut connections, &mut conn_io, &new_ips, + &[], "127.0.0.1", 8080, &mut last_selected_idx, @@ -566,6 +567,7 @@ mod tests { new_ips: Some(SmallVec::from_vec(vec![IpAddr::V4(Ipv4Addr::new( 192, 168, 1, 100, ))])), + new_weights: SmallVec::new(), receiver_host: "test-host".to_string(), receiver_port: 9090, }; @@ -598,7 +600,7 @@ mod tests { std::sync::Arc::new(crate::net::SourceIpBinder); let mut conn_io = ConnIoMap::new(); let connections = - create_connections_from_ips(&ips, "127.0.0.1", 9999, &binder, &mut conn_io).await; + create_connections_from_ips(&ips, &[], "127.0.0.1", 9999, &binder, &mut conn_io).await; // Connections may be empty due to connection failures, which is OK for testing assert!(connections.len() <= ips.len());