From d9363389a33f19779807ec03ebfbf1a750365588 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Thu, 29 Oct 2026 12:15:59 +0700 Subject: [PATCH] refactor(platform): use concrete platform with safe CAKE updates --- build.rs | 5 +- src/main.rs | 63 ++++--------- src/platform/freebsd.rs | 25 ++--- src/platform/linux.rs | 157 +++++++++++++++++++++++++++++-- src/platform/linux/netlink.rs | 172 ---------------------------------- src/platform/macos.rs | 25 ++--- src/platform/mod.rs | 40 +------- src/platform/noop.rs | 31 ++---- src/platform/unsupported.rs | 100 -------------------- src/ratecontroller.rs | 110 ++++++++-------------- 10 files changed, 237 insertions(+), 491 deletions(-) delete mode 100644 src/platform/linux/netlink.rs delete mode 100644 src/platform/unsupported.rs diff --git a/build.rs b/build.rs index dba8fb6..230218b 100644 --- a/build.rs +++ b/build.rs @@ -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"); diff --git a/src/main.rs b/src/main.rs index 39ffe4f..2f05eb9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -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( - 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( - 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( 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( - 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 } diff --git a/src/platform/freebsd.rs b/src/platform/freebsd.rs index fee7e81..02c693e 100644 --- a/src/platform/freebsd.rs +++ b/src/platform/freebsd.rs @@ -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 { +impl Platform { + pub(crate) fn read_stats( + &self, + interface: &str, + ) -> Result { 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." diff --git a/src/platform/linux.rs b/src/platform/linux.rs index a5050fa..d47b8fa 100644 --- a/src/platform/linux.rs +++ b/src/platform/linux.rs @@ -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, + shaper_socket: Mutex, +} + +impl Platform { + pub(crate) fn read_stats(&self, interface: &str) -> Result { + 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 { + 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 { + 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() {} diff --git a/src/platform/linux/netlink.rs b/src/platform/linux/netlink.rs deleted file mode 100644 index 5742979..0000000 --- a/src/platform/linux/netlink.rs +++ /dev/null @@ -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 { - 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 { - 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 { - 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 { - 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::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) - } -} diff --git a/src/platform/macos.rs b/src/platform/macos.rs index c4bf5e8..c7d2345 100644 --- a/src/platform/macos.rs +++ b/src/platform/macos.rs @@ -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::() + 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 { +impl Platform { + pub(crate) fn read_stats( + &self, + interface: &str, + ) -> Result { 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." diff --git a/src/platform/mod.rs b/src/platform/mod.rs index 170d41a..f307004 100644 --- a/src/platform/mod.rs +++ b/src/platform/mod.rs @@ -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; -} - -pub(crate) trait TrafficControlBackend: Send { - type Error: Error + Send + Sync + 'static; - type Handle: Send + 'static; - - fn find_shaper(&mut self, interface: &str) -> Result; - - 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}; diff --git a/src/platform/noop.rs b/src/platform/noop.rs index 05c5e5b..0b3377d 100644 --- a/src/platform/noop.rs +++ b/src/platform/noop.rs @@ -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 { - 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(()) } diff --git a/src/platform/unsupported.rs b/src/platform/unsupported.rs deleted file mode 100644 index 8119fd9..0000000 --- a/src/platform/unsupported.rs +++ /dev/null @@ -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 = 500..=25_000; -const SYNTHETIC_TX_KBIT_RANGE: std::ops::RangeInclusive = 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, -} - -impl InterfaceStatsProvider for SyntheticInterfaceStats { - type Error = Infallible; - - fn read_stats(&mut self, interface: &str) -> Result { - 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." - ); -} diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index e6363ca..dbef7e0 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -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( - 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 { +struct State { current_bytes: i128, current_rate: f64, delta_stat: f64, deltas: Vec, - shaper: H, load: f64, next_rate: f64, nrate: usize, @@ -65,8 +62,8 @@ struct State { utilisation: f64, } -impl State { - fn new(shaper: H, previous_bytes: i128, safe_rates: Vec) -> Self { +impl State { + fn new(previous_bytes: i128, safe_rates: Vec) -> Self { State { current_bytes: 0, current_rate: 0.0, @@ -75,7 +72,6 @@ impl State { 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 { +pub struct Ratecontroller { settings: Settings, control_snapshot_rx: tokio::sync::watch::Receiver>, reflectors_lock: ArcRwLock>, reselect_trigger: mpsc::Sender, - state_dl: State, - state_ul: State, + state_dl: State, + state_ul: State, rng: fastrand::Rng, - stats_provider: S, - traffic_control: T, + platform: Arc, metrics: MetricsSender, } -impl Ratecontroller { - #[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>, reflectors_lock: ArcRwLock>, reselect_trigger: mpsc::Sender, metrics: MetricsSender, - mut stats_provider: S, - mut traffic_control: T, - mut rng: fastrand::Rng, + platform: Arc, ) -> anyhow::Result { - 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 Ratecontroller { 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 Ratecontroller { // 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 Ratecontroller { 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 Ratecontroller { } 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 Ratecontroller { Ok(()) } } - -impl Ratecontroller { - pub fn new( - settings: Settings, - control_snapshot_rx: tokio::sync::watch::Receiver>, - reflectors_lock: ArcRwLock>, - reselect_trigger: mpsc::Sender, - metrics: MetricsSender, - ) -> anyhow::Result { - Self::new_with_backends( - settings, - control_snapshot_rx, - reflectors_lock, - reselect_trigger, - metrics, - interface_stats_provider(), - traffic_control_backend(), - fastrand::Rng::new(), - ) - } -}