classic: per-link weights (Moblin connection priorities) via the IPs file

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 `<ip>[ <weight>]`: 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
 

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 (`<ip> <weight>`), 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 {

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<usize
         if c.is_timed_out(now_ms) || !c.is_schedulable() || c.stall_gated {
             continue;
         }
-        let score = c.get_score();
+        let score = weighted_score(c);
         if score > best_score {
             best_score = score;
             best_idx = Some(i);
@@ -39,4 +43,16 @@ pub fn select_connection(conns: &[SrtlaConnection], now_ms: u64) -> Option<usize
     best_idx
 }
 
+/// `get_score()` scaled by the link's weight multiplier. Truncated back to an
+/// integer like Moblin's `Int(Float(score) * priority)`. A weight of 1 returns
+/// `get_score()` unchanged (including the `-1` of a disconnected link).
+#[inline(always)]
+pub fn weighted_score(c: &SrtlaConnection) -> 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

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

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,

src/sender/connections.rs

@@ -17,6 +17,9 @@ use crate::net::{
 
 pub struct PendingConnectionChanges {
     pub new_ips: Option<SmallVec<IpAddr, 4>>,
+    /// Normalised operator weight per entry of `new_ips` (same order). Empty =
+    /// every link weight 1.
+    pub new_weights: SmallVec<u8, 4>,
     pub receiver_host: String,
     pub receiver_port: u16,
 }
@@ -26,6 +29,7 @@ pub async fn apply_connection_changes(
     connections: &mut SmallVec<SrtlaConnection, 4>,
     conn_io: &mut ConnIoMap,
     new_ips: &[IpAddr],
+    new_weights: &[u8],
     receiver_host: &str,
     receiver_port: u16,
     last_selected_idx: &mut Option<usize>,
@@ -62,21 +66,26 @@ 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::<IpAddr>::new();
-    let new_ips_needed: SmallVec<IpAddr, 4> = new_ips
+    let (new_ips_needed, new_weights_needed): (SmallVec<IpAddr, 4>, SmallVec<u8, 4>) = new_ips
         .iter()
         .copied()
         .filter(|ip| seen.insert(*ip))
         .filter(|ip| {
             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<dyn UplinkBinder>,
@@ -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);
             }

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,24 +104,24 @@ 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::<SmallVec<_, 4>>()
-            .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));
     }
 
     // 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<SmallVec<IpAddr, 4>> {
+    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::<SmallVec<_, 4>>()
+        .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<IpAddr, 4>, SmallVec<u8, 4>)> {
     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<SmallVec<IpAddr, 4>> {
     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())),
     }
 }