refactor(platform): use concrete platform with safe CAKE updates
Rust CI / cargo fmt (push) Successful in 17s
Rust CI / cargo clippy (push) Successful in 54s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 43s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Successful in 54s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m0s
Build binaries / Build (aarch64-musl) (push) Successful in 2m2s
Build binaries / Build (arm-musleabi) (push) Successful in 2m2s
Rust CI / cargo doc (push) Successful in 59s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Successful in 1m6s
Supply chain / actionlint (push) Successful in 17s
Supply chain / cargo-deny advisories (push) Successful in 1m36s
Supply chain / cargo-deny bans licenses sources (push) Successful in 1m36s
Build binaries / Build (arm-musleabihf) (push) Successful in 1m37s
Build binaries / Build (armv7-musleabi) (push) Successful in 1m38s
Build binaries / Build (armv7-musleabihf) (push) Successful in 1m34s
Build binaries / Build (i686-musl) (push) Successful in 1m37s
Build binaries / Build (mips-musl) (push) Successful in 1m54s
Build binaries / Build (mips64-muslabi64) (push) Successful in 1m53s
Build binaries / Build (mips64el-muslabi64) (push) Successful in 1m53s
Build binaries / Build (mipsel-musl) (push) Successful in 1m50s
Build binaries / Build (x86_64-musl) (push) Successful in 1m34s
Build binaries / Build (riscv64gc-musl) (push) Successful in 1m40s

This commit is contained in:
Nils Andreas Svee committed 2026-10-29 12:15:59 +07:00
1 parent 2ba2f966a6
commit d9363389a3
10 files changed
+237 -491

No files matched your search

+3 -2
View File
@@ -17,9 +17,10 @@ fn main() {
"macos" => println!(
"cargo::warning=traffic control is unavailable on macOS; requested rate changes will be calculated but not applied"
),
_ => println!(
"cargo::warning=target `{target}` has no native interface-statistics or traffic-control backend; synthetic interface load will be used and requested rate changes will not be applied"
"freebsd" => println!(
"cargo::warning=traffic control is unavailable on FreeBSD; requested rate changes will be calculated but not applied"
),
_ => {}
}
println!("cargo::rerun-if-changed=build.rs");
+20 -43
View File
@@ -26,7 +26,7 @@ use crate::measurement::icmp::IcmpEngine;
use crate::measurement::irtt::IrttEngine;
use crate::measurement::model::{MeasurementSource, MeasurementStream};
use crate::metrics::{Metric, Metrics, MetricsSender};
use crate::platform::{TrafficControlBackend, traffic_control_backend, warn_platform_limitations};
use crate::platform::{Platform, warn_platform_limitations};
use crate::ratecontroller::Ratecontroller;
use crate::reflector_selector::ReflectorSelector;
use crate::settings::{ReflectorDefinition, ReflectorProtocol, Settings};
@@ -58,20 +58,9 @@ extern "C" fn signal_handler(_: libc::c_int) {
const VERSION: &str = env!("CARGO_PKG_VERSION");
fn initialize_shaper<T: TrafficControlBackend>(
settings: &Settings,
traffic_control: &mut T,
) -> anyhow::Result<(T::Handle, T::Handle)> {
initialize_shaper_with_settle_time(settings, traffic_control, Duration::from_secs(2))
}
fn initialize_shaper_with_settle_time<T: TrafficControlBackend>(
settings: &Settings,
traffic_control: &mut T,
settle_time: Duration,
) -> anyhow::Result<(T::Handle, T::Handle)> {
let down = traffic_control.find_shaper(&settings.network.download_interface)?;
let up = traffic_control.find_shaper(&settings.network.upload_interface)?;
fn initialize_shaper(settings: &Settings, platform: &Platform) -> anyhow::Result<()> {
platform.validate_shaper(&settings.network.download_interface)?;
platform.validate_shaper(&settings.network.upload_interface)?;
info!(
"Requesting minimum shaper rates (D/U): {} / {}",
@@ -79,25 +68,22 @@ fn initialize_shaper_with_settle_time<T: TrafficControlBackend>(
settings.network.upload_min_kbits(),
);
traffic_control.set_rate(
&down,
platform.set_rate(
&settings.network.download_interface,
settings.network.download_min_kbits() as u64,
settings.advanced_settings.dry_run,
)?;
traffic_control.set_rate(
&up,
platform.set_rate(
&settings.network.upload_interface,
settings.network.upload_min_kbits() as u64,
settings.advanced_settings.dry_run,
)?;
info!(
"Sleeping for {} seconds to give the shaper a chance to control existing bloat",
settle_time.as_secs_f64(),
);
info!("Sleeping for 2 seconds to give the shaper a chance to control existing bloat");
sleep(settle_time);
sleep(Duration::from_secs(2));
Ok((down, up))
Ok(())
}
fn install_signal_handlers() {
@@ -113,27 +99,22 @@ fn install_signal_handlers() {
}
}
fn restore_shaper<T: TrafficControlBackend>(
settings: &Settings,
traffic_control: &mut T,
down: &T::Handle,
up: &T::Handle,
) {
fn restore_shaper(settings: &Settings, platform: &Platform) {
info!(
"Requesting restoration to base shaper rates (D/U): {} / {}",
settings.network.download_base_kbits, settings.network.upload_base_kbits,
);
if let Err(error) = traffic_control.set_rate(
down,
if let Err(error) = platform.set_rate(
&settings.network.download_interface,
settings.network.download_base_kbits as u64,
settings.advanced_settings.dry_run,
) {
warn!("Failed to restore download shaper rate: {error}");
}
if let Err(error) = traffic_control.set_rate(
up,
if let Err(error) = platform.set_rate(
&settings.network.upload_interface,
settings.network.upload_base_kbits as u64,
settings.advanced_settings.dry_run,
) {
@@ -445,8 +426,8 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
ping_metrics,
};
let mut main_traffic_control = traffic_control_backend();
let (down_shaper, up_shaper) = initialize_shaper(settings, &mut main_traffic_control)?;
let platform = Arc::new(Platform::default());
initialize_shaper(settings, &platform)?;
spawn_worker("baseliner", &error_tx, move || baseliner.run())?;
spawn_task(
@@ -480,6 +461,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
active_streams,
reselect_tx,
rate_metrics,
Arc::clone(&platform),
)?;
spawn_worker("ratecontroller", &error_tx, move || ratecontroller.run())?;
@@ -509,12 +491,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
let _ = rt.block_on(handle);
}
restore_shaper(
settings,
&mut main_traffic_control,
&down_shaper,
&up_shaper,
);
restore_shaper(settings, &platform);
result
}
+7 -18
View File
@@ -9,8 +9,7 @@ use std::ptr;
use log::warn;
use super::noop::NoopTrafficControl;
use crate::platform::{InterfaceStats, InterfaceStatsProvider};
use crate::platform::InterfaceStats;
#[derive(Debug, thiserror::Error)]
pub(crate) enum FreeBsdInterfaceStatsError {
@@ -43,12 +42,13 @@ pub(crate) enum FreeBsdInterfaceStatsError {
}
#[derive(Debug, Default)]
pub(crate) struct FreeBsdInterfaceStats;
pub(crate) struct Platform {}
impl InterfaceStatsProvider for FreeBsdInterfaceStats {
type Error = FreeBsdInterfaceStatsError;
fn read_stats(&mut self, interface: &str) -> Result<InterfaceStats, Self::Error> {
impl Platform {
pub(crate) fn read_stats(
&self,
interface: &str,
) -> Result<InterfaceStats, FreeBsdInterfaceStatsError> {
let interface_name = CString::new(interface)
.map_err(FreeBsdInterfaceStatsError::InterfaceNameContainsNul)?;
let interface_index = unsafe { libc::if_nametoindex(interface_name.as_ptr()) };
@@ -107,17 +107,6 @@ impl InterfaceStatsProvider for FreeBsdInterfaceStats {
}
}
pub(crate) type PlatformInterfaceStats = FreeBsdInterfaceStats;
pub(crate) type PlatformTrafficControl = NoopTrafficControl;
pub(crate) fn interface_stats_provider() -> PlatformInterfaceStats {
FreeBsdInterfaceStats
}
pub(crate) fn traffic_control_backend() -> PlatformTrafficControl {
NoopTrafficControl
}
pub(crate) fn warn_platform_limitations() {
warn!(
"Traffic control is unavailable on FreeBSD; requested rate changes will be calculated but not applied."
+149 -8
View File
@@ -1,18 +1,159 @@
// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
// SPDX-FileCopyrightText: 2022-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
//
// SPDX-License-Identifier: MPL-2.0
mod netlink;
use std::io;
use std::sync::Mutex;
pub(crate) type PlatformInterfaceStats = netlink::Netlink;
pub(crate) type PlatformTrafficControl = netlink::Netlink;
use log::info;
use netlink_bindings::{consts, rt_link, tc};
use netlink_socket2::NetlinkSocket;
use thiserror::Error;
pub(crate) fn interface_stats_provider() -> PlatformInterfaceStats {
netlink::Netlink::default()
use crate::platform::InterfaceStats;
const TC_H_ROOT: u32 = 0xffff_ffff;
#[derive(Debug, Error)]
pub(crate) enum NetlinkError {
#[error("Couldn't find interface `{0}`")]
InterfaceNotFound(String),
#[error("Netlink error: {0}")]
Netlink(#[from] io::Error),
#[error("Netlink reply error: {0}")]
Reply(#[from] netlink_socket2::ReplyError),
#[error("Couldn't find root CAKE qdisc on interface `{0}`")]
NoQdiscFound(String),
#[error("Root CAKE qdisc on interface `{0}` has a zero handle; refusing to modify it")]
ZeroQdiscHandle(String),
#[error("Couldn't find interface statistics: `{0}`")]
NoInterfaceStatsFound(String),
#[error("Netlink socket lock poisoned")]
SocketPoisoned,
}
pub(crate) fn traffic_control_backend() -> PlatformTrafficControl {
netlink::Netlink::default()
#[derive(Default)]
pub(crate) struct Platform {
stats_socket: Mutex<NetlinkSocket>,
shaper_socket: Mutex<NetlinkSocket>,
}
impl Platform {
pub(crate) fn read_stats(&self, interface: &str) -> Result<InterfaceStats, NetlinkError> {
let mut socket = self
.stats_socket
.lock()
.map_err(|_| NetlinkError::SocketPoisoned)?;
let mut request = rt_link::Request::new().op_getlink_do(&Default::default());
request
.encode()
.push_ifname_bytes(interface.as_bytes())
.push_ext_mask(1 /* RTEXT_FILTER_VF */);
let mut iter = socket.request(&request)?;
while let Some(reply) = iter.recv() {
let (_, attrs) = reply?;
if let Ok(stats) = attrs.get_stats64() {
return Ok(InterfaceStats {
rx_bytes: stats.rx_bytes,
tx_bytes: stats.tx_bytes,
});
}
}
Err(NetlinkError::NoInterfaceStatsFound(interface.to_string()))
}
pub(crate) fn validate_shaper(&self, interface: &str) -> Result<(), NetlinkError> {
let mut socket = self
.shaper_socket
.lock()
.map_err(|_| NetlinkError::SocketPoisoned)?;
Self::root_qdisc(&mut socket, interface)?;
Ok(())
}
pub(crate) fn set_rate(
&self,
interface: &str,
bandwidth_kbit: u64,
dry_run: bool,
) -> Result<(), NetlinkError> {
// Hold the lock through lookup and acknowledgement so requests sharing this socket
// cannot consume each other's replies.
let mut socket = self
.shaper_socket
.lock()
.map_err(|_| NetlinkError::SocketPoisoned)?;
let qdisc = Self::root_qdisc(&mut socket, interface)?;
if dry_run {
info!(
"dry-run: would set qdisc ifindex={} to {} kbit/s",
qdisc.ifindex, bandwidth_kbit
);
return Ok(());
}
let bandwidth = bandwidth_kbit * 1000 / 8;
let mut header = tc::Tcmsg::new();
header.ifindex = qdisc.ifindex;
header.parent = TC_H_ROOT;
header.handle = qdisc.handle;
// Modify only: CREATE/REPLACE could recreate a qdisc removed after lookup.
let mut request = tc::Request::new().op_newqdisc_do(&header);
request
.encode()
.push_kind(c"cake")
.nested_options_cake()
.push_base_rate64(bandwidth)
.end_nested();
let mut iter = socket.request(&request)?;
iter.recv_ack()?;
Ok(())
}
fn root_qdisc(socket: &mut NetlinkSocket, interface: &str) -> Result<tc::Tcmsg, NetlinkError> {
let ifindex = Self::find_interface(socket, interface)?;
let mut header = tc::Tcmsg::new();
header.ifindex = ifindex;
header.parent = TC_H_ROOT;
// Qdisc GET uses the notification path; ECHO requests a unicast reply.
let request = tc::Request::new()
.set_flags(consts::NLM_F_ECHO as u16)
.op_getqdisc_do(&header);
let mut iter = socket.request(&request)?;
let (header, attrs) = iter.recv_one()?;
if attrs.get_kind().is_ok_and(|kind| kind == c"cake") {
// A zero handle can let even a change-only request replace another kind.
if header.handle == 0 {
return Err(NetlinkError::ZeroQdiscHandle(interface.to_string()));
}
Ok(header)
} else {
Err(NetlinkError::NoQdiscFound(interface.to_string()))
}
}
fn find_interface(socket: &mut NetlinkSocket, interface: &str) -> Result<i32, NetlinkError> {
let mut request = rt_link::Request::new().op_getlink_do(&Default::default());
request.encode().push_ifname_bytes(interface.as_bytes());
let mut iter = socket.request(&request)?;
let (header, _) = iter
.recv_one()
.map_err(|_| NetlinkError::InterfaceNotFound(interface.to_string()))?;
Ok(header.ifi_index)
}
}
pub(crate) fn warn_platform_limitations() {}
-172
View File
@@ -1,172 +0,0 @@
// SPDX-FileCopyrightText: 2022-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
//
// SPDX-License-Identifier: MPL-2.0
use std::io;
use std::str::Utf8Error;
use log::info;
use netlink_bindings::{rt_link, tc};
use netlink_socket2::NetlinkSocket;
use thiserror::Error;
use crate::platform::{InterfaceStats, InterfaceStatsProvider, TrafficControlBackend};
#[derive(Debug, Error)]
pub(crate) enum NetlinkError {
#[error("Couldn't find interface `{0}`")]
InterfaceNotFound(String),
#[error("Netlink error: {0}")]
Netlink(#[from] io::Error),
#[error("Netlink reply error: {0}")]
Reply(#[from] netlink_socket2::ReplyError),
#[error("Something went wrong while finding qdisc")]
NlQdiscError(String),
#[error("Couldn't find CAKE qdisc on interface `{0}`")]
NoQdiscFound(String),
#[error("Couldn't find interface statistics: `{0}`")]
NoInterfaceStatsFound(String),
#[error("Error happened while parsing UTF-8 string")]
Utf8Error(#[from] Utf8Error),
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct Qdisc {
ifindex: i32,
parent: u32,
}
#[derive(Debug, Default)]
pub(crate) struct Netlink {}
impl Netlink {
fn find_interface(ifname: &str) -> Result<i32, NetlinkError> {
let mut socket = NetlinkSocket::new();
let mut request = rt_link::Request::new().op_getlink_do(&Default::default());
request.encode().push_ifname_bytes(ifname.as_bytes());
let mut iter = socket.request(&request)?;
let (header, _) = iter
.recv_one()
.map_err(|_| NetlinkError::InterfaceNotFound(ifname.to_string()))?;
Ok(header.ifi_index)
}
fn get_interface_stats(ifname: &str) -> Result<(u64, u64), NetlinkError> {
let mut socket = NetlinkSocket::new();
let mut request = rt_link::Request::new().op_getlink_do(&Default::default());
request
.encode()
.push_ifname_bytes(ifname.as_bytes())
.push_ext_mask(1 /* RTEXT_FILTER_VF */);
let mut iter = socket.request(&request)?;
while let Some(reply) = iter.recv() {
let (_, attrs) = reply?;
if let Ok(stats) = attrs.get_stats64() {
return Ok((stats.rx_bytes, stats.tx_bytes));
}
}
Err(NetlinkError::NoInterfaceStatsFound(ifname.to_string()))
}
fn qdisc_from_ifindex(ifindex: i32, ifname: &str) -> Result<Qdisc, NetlinkError> {
let mut socket = NetlinkSocket::new();
let header = tc::Tcmsg::new();
let request = tc::Request::new().op_getqdisc_dump(&header);
let mut iter = socket.request(&request)?;
while let Some(reply) = iter.recv() {
let (header, attrs) = reply?;
if header.ifindex == ifindex
&& let Ok(kind) = attrs.get_kind()
&& kind
.to_str()
.map_err(|_| NetlinkError::NlQdiscError("Invalid UTF-8".to_string()))?
== "cake"
{
return Ok(Qdisc {
ifindex,
parent: header.parent,
});
}
}
Err(NetlinkError::NoQdiscFound(ifname.to_string()))
}
fn qdisc_from_ifname(ifname: &str) -> Result<Qdisc, NetlinkError> {
let ifindex = Netlink::find_interface(ifname)?;
Netlink::qdisc_from_ifindex(ifindex, ifname)
}
fn set_qdisc_rate(
qdisc: Qdisc,
bandwidth_kbit: u64,
dry_run: bool,
) -> Result<(), NetlinkError> {
if dry_run {
info!(
"dry-run: would set qdisc ifindex={} to {} kbit/s",
qdisc.ifindex, bandwidth_kbit
);
return Ok(());
}
let mut socket = NetlinkSocket::new();
let bandwidth = bandwidth_kbit * 1000 / 8;
let mut header = tc::Tcmsg::new();
header.ifindex = qdisc.ifindex;
header.parent = qdisc.parent;
let mut request = tc::Request::new().set_change().op_newqdisc_do(&header);
request
.encode()
.push_kind(c"cake")
.nested_options_cake()
.push_base_rate64(bandwidth)
.end_nested();
let mut iter = socket.request(&request)?;
iter.recv_ack()?;
Ok(())
}
}
impl InterfaceStatsProvider for Netlink {
type Error = NetlinkError;
fn read_stats(&mut self, interface: &str) -> Result<InterfaceStats, Self::Error> {
let (rx_bytes, tx_bytes) = Self::get_interface_stats(interface)?;
Ok(InterfaceStats { rx_bytes, tx_bytes })
}
}
impl TrafficControlBackend for Netlink {
type Error = NetlinkError;
type Handle = Qdisc;
fn find_shaper(&mut self, interface: &str) -> Result<Self::Handle, Self::Error> {
Self::qdisc_from_ifname(interface)
}
fn set_rate(
&mut self,
shaper: &Self::Handle,
bandwidth_kbit: u64,
dry_run: bool,
) -> Result<(), Self::Error> {
Self::set_qdisc_rate(*shaper, bandwidth_kbit, dry_run)
}
}
+7 -18
View File
@@ -10,8 +10,7 @@ use std::ptr;
use log::warn;
use thiserror::Error;
use super::noop::NoopTrafficControl;
use crate::platform::{InterfaceStats, InterfaceStatsProvider};
use crate::platform::InterfaceStats;
const ROUTE_MESSAGE_PREFIX_LEN: usize = size_of::<libc::c_ushort>() + 2;
const SYSCTL_FETCH_ATTEMPTS: usize = 3;
@@ -43,12 +42,13 @@ pub(crate) enum MacOsInterfaceStatsError {
}
#[derive(Debug, Default)]
pub(crate) struct MacOsInterfaceStats;
pub(crate) struct Platform {}
impl InterfaceStatsProvider for MacOsInterfaceStats {
type Error = MacOsInterfaceStatsError;
fn read_stats(&mut self, interface: &str) -> Result<InterfaceStats, Self::Error> {
impl Platform {
pub(crate) fn read_stats(
&self,
interface: &str,
) -> Result<InterfaceStats, MacOsInterfaceStatsError> {
let interface_name =
CString::new(interface).map_err(MacOsInterfaceStatsError::InterfaceNameContainsNul)?;
// SAFETY: `interface_name` is a live, NUL-terminated C string for the duration of the call.
@@ -200,17 +200,6 @@ fn parse_route_messages(
})
}
pub(crate) type PlatformInterfaceStats = MacOsInterfaceStats;
pub(crate) type PlatformTrafficControl = NoopTrafficControl;
pub(crate) fn interface_stats_provider() -> PlatformInterfaceStats {
MacOsInterfaceStats
}
pub(crate) fn traffic_control_backend() -> PlatformTrafficControl {
NoopTrafficControl
}
pub(crate) fn warn_platform_limitations() {
warn!(
"Traffic control is unavailable on macOS; requested rate changes will be calculated but not applied."
+5 -35
View File
@@ -2,35 +2,13 @@
//
// SPDX-License-Identifier: MPL-2.0
use std::error::Error;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) struct InterfaceStats {
pub(crate) rx_bytes: u64,
pub(crate) tx_bytes: u64,
}
pub(crate) trait InterfaceStatsProvider: Send {
type Error: Error + Send + Sync + 'static;
fn read_stats(&mut self, interface: &str) -> Result<InterfaceStats, Self::Error>;
}
pub(crate) trait TrafficControlBackend: Send {
type Error: Error + Send + Sync + 'static;
type Handle: Send + 'static;
fn find_shaper(&mut self, interface: &str) -> Result<Self::Handle, Self::Error>;
fn set_rate(
&mut self,
shaper: &Self::Handle,
bandwidth_kbit: u64,
dry_run: bool,
) -> Result<(), Self::Error>;
}
#[cfg(not(target_os = "linux"))]
#[cfg(any(target_os = "freebsd", target_os = "macos"))]
mod noop;
#[cfg(target_os = "freebsd")]
@@ -43,21 +21,13 @@ mod linux;
mod macos;
#[cfg(not(any(target_os = "freebsd", target_os = "linux", target_os = "macos")))]
mod unsupported;
compile_error!("RAVEL supports only Linux, FreeBSD, and macOS");
#[cfg(target_os = "freebsd")]
use freebsd as imp;
pub(crate) use freebsd::{Platform, warn_platform_limitations};
#[cfg(target_os = "linux")]
use linux as imp;
pub(crate) use linux::{Platform, warn_platform_limitations};
#[cfg(target_os = "macos")]
use macos as imp;
#[cfg(not(any(target_os = "freebsd", target_os = "linux", target_os = "macos")))]
use unsupported as imp;
pub(crate) use imp::{
PlatformInterfaceStats, PlatformTrafficControl, interface_stats_provider,
traffic_control_backend, warn_platform_limitations,
};
pub(crate) use macos::{Platform, warn_platform_limitations};
+9 -22
View File
@@ -2,37 +2,24 @@
//
// SPDX-License-Identifier: MPL-2.0
use crate::platform::TrafficControlBackend;
use crate::platform::Platform;
use log::debug;
use std::convert::Infallible;
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct NoopShaper {
interface: String,
}
#[derive(Debug, Default)]
pub(crate) struct NoopTrafficControl;
impl TrafficControlBackend for NoopTrafficControl {
type Error = Infallible;
type Handle = NoopShaper;
fn find_shaper(&mut self, interface: &str) -> Result<Self::Handle, Self::Error> {
Ok(NoopShaper {
interface: interface.to_string(),
})
impl Platform {
pub(crate) fn validate_shaper(&self, _interface: &str) -> Result<(), Infallible> {
Ok(())
}
fn set_rate(
&mut self,
shaper: &Self::Handle,
pub(crate) fn set_rate(
&self,
interface: &str,
bandwidth_kbit: u64,
dry_run: bool,
) -> Result<(), Self::Error> {
) -> Result<(), Infallible> {
debug!(
"Ignoring traffic-control rate request of {} kbit/s for interface {} (dry_run={})",
bandwidth_kbit, shaper.interface, dry_run
bandwidth_kbit, interface, dry_run
);
Ok(())
}
-100
View File
@@ -1,100 +0,0 @@
// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
//
// SPDX-License-Identifier: MPL-2.0
use std::collections::HashMap;
use std::convert::Infallible;
use std::time::{Duration, Instant};
use log::warn;
use crate::platform::noop::NoopTrafficControl;
use crate::platform::{InterfaceStats, InterfaceStatsProvider};
// Synthetic rates are fixed per interface and intentionally conservative. Counter increments are
// derived from elapsed monotonic time, so changing the polling frequency does not change the
// apparent traffic rate.
const SYNTHETIC_RX_KBIT_RANGE: std::ops::RangeInclusive<u64> = 500..=25_000;
const SYNTHETIC_TX_KBIT_RANGE: std::ops::RangeInclusive<u64> = 250..=10_000;
const INITIAL_COUNTER_BYTES: u64 = 1;
#[derive(Debug)]
struct SyntheticInterfaceState {
rx_bytes: u64,
tx_bytes: u64,
rx_kbit: u64,
tx_kbit: u64,
last_update: Instant,
}
impl SyntheticInterfaceState {
fn new(now: Instant) -> Self {
Self {
rx_bytes: INITIAL_COUNTER_BYTES,
tx_bytes: INITIAL_COUNTER_BYTES,
rx_kbit: fastrand::u64(SYNTHETIC_RX_KBIT_RANGE),
tx_kbit: fastrand::u64(SYNTHETIC_TX_KBIT_RANGE),
last_update: now,
}
}
fn update(&mut self, now: Instant) {
let elapsed = now.saturating_duration_since(self.last_update);
self.rx_bytes = self
.rx_bytes
.saturating_add(bytes_for_elapsed(elapsed, self.rx_kbit));
self.tx_bytes = self
.tx_bytes
.saturating_add(bytes_for_elapsed(elapsed, self.tx_kbit));
self.last_update = now;
}
}
fn bytes_for_elapsed(elapsed: Duration, bitrate_kbit: u64) -> u64 {
let bytes = elapsed
.as_nanos()
.saturating_mul(u128::from(bitrate_kbit))
.saturating_mul(1_000)
/ 8_000_000_000;
u64::try_from(bytes).unwrap_or(u64::MAX)
}
#[derive(Debug, Default)]
pub(crate) struct SyntheticInterfaceStats {
interfaces: HashMap<String, SyntheticInterfaceState>,
}
impl InterfaceStatsProvider for SyntheticInterfaceStats {
type Error = Infallible;
fn read_stats(&mut self, interface: &str) -> Result<InterfaceStats, Self::Error> {
let now = Instant::now();
let state = self
.interfaces
.entry(interface.to_string())
.or_insert_with(|| SyntheticInterfaceState::new(now));
state.update(now);
Ok(InterfaceStats {
rx_bytes: state.rx_bytes,
tx_bytes: state.tx_bytes,
})
}
}
pub(crate) type PlatformInterfaceStats = SyntheticInterfaceStats;
pub(crate) type PlatformTrafficControl = NoopTrafficControl;
pub(crate) fn interface_stats_provider() -> PlatformInterfaceStats {
SyntheticInterfaceStats::default()
}
pub(crate) fn traffic_control_backend() -> PlatformTrafficControl {
NoopTrafficControl
}
pub(crate) fn warn_platform_limitations() {
warn!(
"This operating system has no native interface-statistics or traffic-control backend; synthetic interface load will be used and requested rate changes will not be applied."
);
}
+37 -73
View File
@@ -9,16 +9,14 @@ use crate::SHUTDOWN;
use crate::baseliner::ControlSnapshot;
use crate::measurement::model::MeasurementStream;
use crate::metrics::{Metric, MetricsSender};
use crate::platform::{
InterfaceStatsProvider, PlatformInterfaceStats, PlatformTrafficControl, TrafficControlBackend,
interface_stats_provider, traffic_control_backend,
};
use crate::platform::Platform;
use crate::settings::Settings;
use crate::time::{ClockSample, Realtime};
use crate::util::{ArcRwLock, RwLockExt};
use log::{debug, info, warn};
use std::fs::File;
use std::io::Write;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::thread::sleep;
use std::time::{Duration, Instant};
@@ -40,22 +38,21 @@ fn generate_initial_speeds(rng: &mut fastrand::Rng, base_speed: f64, size: u32)
rates
}
fn get_interface_stats<S: InterfaceStatsProvider>(
stats_provider: &mut S,
fn get_interface_stats(
platform: &Platform,
upload_interface: &str,
) -> Result<(i128, i128), S::Error> {
let stats = stats_provider.read_stats(upload_interface)?;
) -> anyhow::Result<(i128, i128)> {
let stats = platform.read_stats(upload_interface)?;
Ok((stats.rx_bytes.into(), stats.tx_bytes.into()))
}
#[derive(Clone, Debug)]
struct State<H> {
struct State {
current_bytes: i128,
current_rate: f64,
delta_stat: f64,
deltas: Vec<f64>,
shaper: H,
load: f64,
next_rate: f64,
nrate: usize,
@@ -65,8 +62,8 @@ struct State<H> {
utilisation: f64,
}
impl<H> State<H> {
fn new(shaper: H, previous_bytes: i128, safe_rates: Vec<f64>) -> Self {
impl State {
fn new(previous_bytes: i128, safe_rates: Vec<f64>) -> Self {
State {
current_bytes: 0,
current_rate: 0.0,
@@ -75,7 +72,6 @@ impl<H> State<H> {
load: 0.0,
next_rate: 0.0,
nrate: 0,
shaper,
previous_bytes,
prev_t: Instant::now(),
safe_rates,
@@ -121,61 +117,52 @@ fn deltas_from_snapshot(
(down, up)
}
pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
pub struct Ratecontroller {
settings: Settings,
control_snapshot_rx: tokio::sync::watch::Receiver<Option<ControlSnapshot>>,
reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
reselect_trigger: mpsc::Sender<bool>,
state_dl: State<T::Handle>,
state_ul: State<T::Handle>,
state_dl: State,
state_ul: State,
rng: fastrand::Rng,
stats_provider: S,
traffic_control: T,
platform: Arc<Platform>,
metrics: MetricsSender,
}
impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
#[allow(
clippy::too_many_arguments,
reason = "Constructor injects independent runtime and test backends"
)]
fn new_with_backends(
impl Ratecontroller {
pub fn new(
settings: Settings,
control_snapshot_rx: tokio::sync::watch::Receiver<Option<ControlSnapshot>>,
reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
reselect_trigger: mpsc::Sender<bool>,
metrics: MetricsSender,
mut stats_provider: S,
mut traffic_control: T,
mut rng: fastrand::Rng,
platform: Arc<Platform>,
) -> anyhow::Result<Self> {
let dl_shaper =
traffic_control.find_shaper(settings.network.download_interface.as_str())?;
platform.validate_shaper(&settings.network.download_interface)?;
platform.validate_shaper(&settings.network.upload_interface)?;
let mut rng = fastrand::Rng::new();
let dl_safe_rates = generate_initial_speeds(
&mut rng,
settings.network.download_base_kbits,
settings.advanced_settings.speed_hist_size,
);
let ul_shaper = traffic_control.find_shaper(settings.network.upload_interface.as_str())?;
let ul_safe_rates = generate_initial_speeds(
&mut rng,
settings.network.upload_base_kbits,
settings.advanced_settings.speed_hist_size,
);
let (cur_rx, cur_tx) =
get_interface_stats(&mut stats_provider, &settings.network.upload_interface)?;
let (cur_rx, cur_tx) = get_interface_stats(&platform, &settings.network.upload_interface)?;
Ok(Self {
settings,
control_snapshot_rx,
reflectors_lock,
reselect_trigger,
state_dl: State::new(dl_shaper, cur_rx, dl_safe_rates),
state_ul: State::new(ul_shaper, cur_tx, ul_safe_rates),
state_dl: State::new(cur_rx, dl_safe_rates),
state_ul: State::new(cur_tx, ul_safe_rates),
rng,
stats_provider,
traffic_control,
platform,
metrics,
})
}
@@ -185,13 +172,13 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
self.state_dl.current_rate = self.settings.network.download_base_kbits * 0.6;
self.state_ul.current_rate = self.settings.network.upload_base_kbits * 0.6;
self.traffic_control.set_rate(
&self.state_dl.shaper,
self.platform.set_rate(
&self.settings.network.download_interface,
self.state_dl.current_rate.round() as u64,
self.settings.advanced_settings.dry_run,
)?;
self.traffic_control.set_rate(
&self.state_ul.shaper,
self.platform.set_rate(
&self.settings.network.upload_interface,
self.state_ul.current_rate.round() as u64,
self.settings.advanced_settings.dry_run,
)?;
@@ -332,10 +319,8 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
// if it's been long enough, and the stats indicate needing to change speeds
// change speeds here
(self.state_dl.current_bytes, self.state_ul.current_bytes) = get_interface_stats(
&mut self.stats_provider,
&self.settings.network.upload_interface,
)?;
(self.state_dl.current_bytes, self.state_ul.current_bytes) =
get_interface_stats(&self.platform, &self.settings.network.upload_interface)?;
self.update_deltas()?;
if self.state_dl.deltas.is_empty() || self.state_ul.deltas.is_empty() {
@@ -349,13 +334,13 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
self.state_dl.next_rate = self.settings.network.download_min_kbits();
self.state_ul.next_rate = self.settings.network.upload_min_kbits();
self.traffic_control.set_rate(
&self.state_dl.shaper,
self.platform.set_rate(
&self.settings.network.download_interface,
self.state_dl.next_rate as u64,
self.settings.advanced_settings.dry_run,
)?;
self.traffic_control.set_rate(
&self.state_ul.shaper,
self.platform.set_rate(
&self.settings.network.upload_interface,
self.state_ul.next_rate as u64,
self.settings.advanced_settings.dry_run,
)?;
@@ -378,16 +363,16 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
}
if self.state_dl.next_rate != self.state_dl.current_rate {
self.traffic_control.set_rate(
&self.state_dl.shaper,
self.platform.set_rate(
&self.settings.network.download_interface,
self.state_dl.next_rate as u64,
self.settings.advanced_settings.dry_run,
)?;
}
if self.state_ul.next_rate != self.state_ul.current_rate {
self.traffic_control.set_rate(
&self.state_ul.shaper,
self.platform.set_rate(
&self.settings.network.upload_interface,
self.state_ul.next_rate as u64,
self.settings.advanced_settings.dry_run,
)?;
@@ -499,24 +484,3 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
Ok(())
}
}
impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
pub fn new(
settings: Settings,
control_snapshot_rx: tokio::sync::watch::Receiver<Option<ControlSnapshot>>,
reflectors_lock: ArcRwLock<Vec<MeasurementStream>>,
reselect_trigger: mpsc::Sender<bool>,
metrics: MetricsSender,
) -> anyhow::Result<Self> {
Self::new_with_backends(
settings,
control_snapshot_rx,
reflectors_lock,
reselect_trigger,
metrics,
interface_stats_provider(),
traffic_control_backend(),
fastrand::Rng::new(),
)
}
}