From 9811592f136472a6514973f3903ad6992a9c61f5 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Fri, 17 Jul 2026 02:07:02 +0700 Subject: [PATCH] feat(platform): introduce abstractions for cross-platform interface statistics and traffic control Decouple the rate controller from Linux-specific Netlink calls to support other operating systems. This adds a macOS backend for interface statistics and an "observe-only" mode for platforms where traffic shaping is not yet implemented. --- Cargo.toml | 4 +- src/log.rs | 2 +- src/main.rs | 53 +++++++---- src/netlink.rs | 32 ++++++- src/pinger_icmp.rs | 2 +- src/pinger_icmp_ts.rs | 2 +- src/platform.rs | 181 ++++++++++++++++++++++++++++++++++++++ src/platform/macos.rs | 199 ++++++++++++++++++++++++++++++++++++++++++ src/ratecontroller.rs | 194 +++++++++++++++++++++++----------------- src/time.rs | 3 +- 10 files changed, 570 insertions(+), 102 deletions(-) create mode 100644 src/platform.rs create mode 100644 src/platform/macos.rs diff --git a/Cargo.toml b/Cargo.toml index bb852ae..5ae9134 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,8 +21,6 @@ strip = "symbols" [dependencies] anyhow = "1.0" icmp-socket2 = "0.4.2" -netlink-bindings = { git = "https://github.com/one-d-wide/netlink-bindings.git", rev = "0b609e9380610e7da73f8c43cdf804a1d1bf5474", features = ["rt-link", "tc"] } -netlink-socket2 = { git = "https://github.com/one-d-wide/netlink-bindings.git", rev = "0b609e9380610e7da73f8c43cdf804a1d1bf5474" } fastrand = "2" rustix = { version = "1.1", features = ["fs", "system", "thread", "time"] } thiserror = "2.0" @@ -34,6 +32,8 @@ features = ["std"] version = "0.4" [target.'cfg(target_os = "linux")'.dependencies] +netlink-bindings = { git = "https://github.com/one-d-wide/netlink-bindings.git", rev = "0b609e9380610e7da73f8c43cdf804a1d1bf5474", features = ["rt-link", "tc"] } +netlink-socket2 = { git = "https://github.com/one-d-wide/netlink-bindings.git", rev = "0b609e9380610e7da73f8c43cdf804a1d1bf5474" } rust-uci = { git = "https://github.com/namib-project/rust-uci.git", rev = "3e7b6153b6df92b45428c32bb6461756563deecb", optional = true } [features] diff --git a/src/log.rs b/src/log.rs index fad74db..fc89067 100644 --- a/src/log.rs +++ b/src/log.rs @@ -3,7 +3,7 @@ // SPDX-License-Identifier: MPL-2.0 use log::{Level, LevelFilter, Metadata, Record, SetLoggerError}; -use rustix::thread::ClockId; +use rustix::time::ClockId; use crate::time::Time; diff --git a/src/main.rs b/src/main.rs index 8b33217..862413b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,10 +11,12 @@ mod baseliner; mod config; mod log; mod metrics; +#[cfg(target_os = "linux")] mod netlink; mod pinger; mod pinger_icmp; mod pinger_icmp_ts; +mod platform; mod ratecontroller; mod reflector_selector; mod time; @@ -34,10 +36,10 @@ use std::time::{Duration, Instant}; use std::{process, thread}; use crate::config::{Config, MeasurementType}; -use crate::netlink::{Netlink, Qdisc}; use crate::pinger::{InFlightProbeCache, PingListener, PingSender}; use crate::pinger_icmp::{PingerICMPEchoListener, PingerICMPEchoSender}; use crate::pinger_icmp_ts::{PingerICMPTimestampListener, PingerICMPTimestampSender}; +use crate::platform::{TrafficControlBackend, traffic_control_backend}; use crate::ratecontroller::{Ratecontroller, StatsDirection}; use crate::reflector_selector::ReflectorSelector; @@ -101,19 +103,28 @@ fn create_pinger(measurement_type: MeasurementType) -> (PingListenerBox, PingSen } } -fn initialize_shaper(config: &Config) -> anyhow::Result<(Qdisc, Qdisc)> { - let down = Netlink::qdisc_from_ifname(&config.download_interface)?; - let up = Netlink::qdisc_from_ifname(&config.upload_interface)?; +fn initialize_shaper( + config: &Config, + traffic_control: &mut T, +) -> anyhow::Result<(T::Handle, T::Handle)> { + initialize_shaper_with_settle_time(config, traffic_control, Duration::from_secs(2)) +} + +fn initialize_shaper_with_settle_time( + config: &Config, + traffic_control: &mut T, + settle_time: Duration, +) -> anyhow::Result<(T::Handle, T::Handle)> { + let down = traffic_control.find_shaper(&config.download_interface)?; + let up = traffic_control.find_shaper(&config.upload_interface)?; info!( - "Setting shaper rates to minimum (D/L): {} / {}", + "Requesting minimum shaper rates (D/U): {} / {}", config.download_min_kbits, config.upload_min_kbits, ); - Netlink::set_qdisc_rate(down, config.download_min_kbits as u64, config.dry_run)?; - Netlink::set_qdisc_rate(up, config.upload_min_kbits as u64, config.dry_run)?; - - let settle_time = Duration::from_secs(2); + traffic_control.set_rate(&down, config.download_min_kbits as u64, config.dry_run)?; + traffic_control.set_rate(&up, config.upload_min_kbits as u64, config.dry_run)?; info!( "Sleeping for {} seconds to give the shaper a chance to control existing bloat", @@ -163,19 +174,25 @@ fn probe_identifier() -> u16 { (process::id() & 0xffff) as u16 } -fn restore_shaper(config: &Config, down: Qdisc, up: Qdisc) { +fn restore_shaper( + config: &Config, + traffic_control: &mut T, + down: &T::Handle, + up: &T::Handle, +) { info!( - "Restoring base shaper rates (D/L): {} / {}", + "Requesting restoration to base shaper rates (D/U): {} / {}", config.download_base_kbits, config.upload_base_kbits, ); if let Err(error) = - Netlink::set_qdisc_rate(down, config.download_base_kbits as u64, config.dry_run) + traffic_control.set_rate(down, config.download_base_kbits as u64, config.dry_run) { warn!("Failed to restore download shaper rate: {error}"); } - if let Err(error) = Netlink::set_qdisc_rate(up, config.upload_base_kbits as u64, config.dry_run) + if let Err(error) = + traffic_control.set_rate(up, config.upload_base_kbits as u64, config.dry_run) { warn!("Failed to restore upload shaper rate: {error}"); } @@ -319,7 +336,13 @@ fn run(config: Config) -> anyhow::Result<()> { event_metrics: event_metrics.clone(), }; - let (down_qdisc, up_qdisc) = initialize_shaper(&config)?; + let mut main_traffic_control = traffic_control_backend(); + if main_traffic_control.is_observe_only() { + warn!( + "Traffic control is observe-only on this platform; requested rate changes will not be applied." + ); + } + let (down_shaper, up_shaper) = initialize_shaper(&config, &mut main_traffic_control)?; let err_tx = error_tx.clone(); let listener_peers = Arc::clone(&reflector_peers); @@ -448,7 +471,7 @@ fn run(config: Config) -> anyhow::Result<()> { let _ = handle.join(); } - restore_shaper(&config, down_qdisc, up_qdisc); + restore_shaper(&config, &mut main_traffic_control, &down_shaper, &up_shaper); result } diff --git a/src/netlink.rs b/src/netlink.rs index 84e3df8..1efaf84 100644 --- a/src/netlink.rs +++ b/src/netlink.rs @@ -10,6 +10,8 @@ use netlink_bindings::{rt_link, tc}; use netlink_socket2::NetlinkSocket; use thiserror::Error; +use crate::platform::{InterfaceStats, InterfaceStatsProvider, TrafficControlBackend}; + #[derive(Debug, Error)] pub enum NetlinkError { #[error("Couldn't find interface `{0}`")] @@ -34,12 +36,13 @@ pub enum NetlinkError { Utf8Error(#[from] Utf8Error), } -#[derive(Clone, Copy, Debug)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct Qdisc { pub ifindex: i32, pub parent: u32, } +#[derive(Debug, Default)] pub struct Netlink {} impl Netlink { @@ -140,3 +143,30 @@ impl Netlink { 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/pinger_icmp.rs b/src/pinger_icmp.rs index 011088c..3a36144 100644 --- a/src/pinger_icmp.rs +++ b/src/pinger_icmp.rs @@ -14,7 +14,7 @@ use crate::time::Time; use icmp_socket2::Icmpv4Message; use icmp_socket2::Icmpv4Packet; use icmp_socket2::packet::WithEchoRequest; -use rustix::thread::ClockId; +use rustix::time::ClockId; pub struct PingerICMPEchoListener {} diff --git a/src/pinger_icmp_ts.rs b/src/pinger_icmp_ts.rs index 51d6012..de8c815 100644 --- a/src/pinger_icmp_ts.rs +++ b/src/pinger_icmp_ts.rs @@ -11,7 +11,7 @@ use crate::time::Time; use icmp_socket2::Icmpv4Message; use icmp_socket2::Icmpv4Packet; use icmp_socket2::packet::WithTimestampRequest; -use rustix::thread::ClockId; +use rustix::time::ClockId; use std::net::IpAddr; use std::time::Instant; diff --git a/src/platform.rs b/src/platform.rs new file mode 100644 index 0000000..37b18c6 --- /dev/null +++ b/src/platform.rs @@ -0,0 +1,181 @@ +// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) +// +// SPDX-License-Identifier: MPL-2.0 + +#[cfg(any(not(target_os = "linux"), test))] +use log::debug; +#[cfg(any(not(target_os = "linux"), test))] +use std::collections::HashMap; +#[cfg(any(not(target_os = "linux"), test))] +use std::convert::Infallible; +use std::error::Error; +#[cfg(any(not(any(target_os = "linux", target_os = "macos")), test))] +use thiserror::Error; + +#[cfg(target_os = "linux")] +use crate::netlink::Netlink; + +#[cfg(target_os = "macos")] +mod macos; +#[cfg(target_os = "macos")] +pub use macos::MacOsInterfaceStats; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct InterfaceStats { + pub rx_bytes: u64, + pub tx_bytes: u64, +} + +pub trait InterfaceStatsProvider: Send { + type Error: Error + Send + Sync + 'static; + + fn read_stats(&mut self, interface: &str) -> Result; +} + +pub 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>; + + fn is_observe_only(&self) -> bool { + false + } +} + +#[cfg(any(not(any(target_os = "linux", target_os = "macos")), test))] +#[derive(Debug, Default)] +pub struct UnsupportedInterfaceStats; + +#[cfg(any(not(any(target_os = "linux", target_os = "macos")), test))] +#[derive(Debug, Error, PartialEq, Eq)] +#[error("interface statistics are unsupported on target OS `{target_os}`")] +pub struct UnsupportedInterfaceStatsError { + target_os: &'static str, +} + +#[cfg(any(not(any(target_os = "linux", target_os = "macos")), test))] +impl InterfaceStatsProvider for UnsupportedInterfaceStats { + type Error = UnsupportedInterfaceStatsError; + + fn read_stats(&mut self, _: &str) -> Result { + Err(UnsupportedInterfaceStatsError { + target_os: std::env::consts::OS, + }) + } +} + +#[cfg(any(not(target_os = "linux"), test))] +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ObserveOnlyShaper { + interface: String, +} + +#[cfg(any(not(target_os = "linux"), test))] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct ObserveOnlyRequest { + bandwidth_kbit: u64, + dry_run: bool, +} + +#[cfg(any(not(target_os = "linux"), test))] +#[derive(Debug, Default)] +pub struct ObserveOnlyTrafficControl { + last_requests: HashMap, +} + +#[cfg(any(not(target_os = "linux"), test))] +impl TrafficControlBackend for ObserveOnlyTrafficControl { + type Error = Infallible; + type Handle = ObserveOnlyShaper; + + fn find_shaper(&mut self, interface: &str) -> Result { + Ok(ObserveOnlyShaper { + interface: interface.to_string(), + }) + } + + fn set_rate( + &mut self, + shaper: &Self::Handle, + bandwidth_kbit: u64, + dry_run: bool, + ) -> Result<(), Self::Error> { + debug!( + "Observe-only traffic control requested rate {} kbit/s for interface {} (dry_run={})", + bandwidth_kbit, shaper.interface, dry_run + ); + self.last_requests.insert( + shaper.interface.clone(), + ObserveOnlyRequest { + bandwidth_kbit, + dry_run, + }, + ); + Ok(()) + } + + fn is_observe_only(&self) -> bool { + true + } +} + +#[cfg(test)] +#[derive(Debug, Default)] +pub struct UnsupportedTrafficControl; + +#[cfg(test)] +#[derive(Debug, PartialEq, Eq)] +pub struct UnsupportedShaper; + +#[cfg(test)] +#[derive(Debug, Error, PartialEq, Eq)] +#[error("traffic control is unsupported on target OS `{target_os}`")] +pub struct UnsupportedTrafficControlError { + target_os: &'static str, +} + +#[cfg(test)] +impl TrafficControlBackend for UnsupportedTrafficControl { + type Error = UnsupportedTrafficControlError; + type Handle = UnsupportedShaper; + + fn find_shaper(&mut self, _: &str) -> Result { + Err(UnsupportedTrafficControlError { + target_os: std::env::consts::OS, + }) + } + + fn set_rate(&mut self, _: &UnsupportedShaper, _: u64, _: bool) -> Result<(), Self::Error> { + Err(UnsupportedTrafficControlError { + target_os: std::env::consts::OS, + }) + } +} + +#[cfg(target_os = "linux")] +pub type PlatformInterfaceStats = Netlink; +#[cfg(target_os = "macos")] +pub type PlatformInterfaceStats = MacOsInterfaceStats; +#[cfg(not(any(target_os = "linux", target_os = "macos")))] +pub type PlatformInterfaceStats = UnsupportedInterfaceStats; + +#[cfg(target_os = "linux")] +pub type PlatformTrafficControl = Netlink; +#[cfg(not(target_os = "linux"))] +pub type PlatformTrafficControl = ObserveOnlyTrafficControl; + +pub fn interface_stats_provider() -> PlatformInterfaceStats { + PlatformInterfaceStats::default() +} + +pub fn traffic_control_backend() -> PlatformTrafficControl { + PlatformTrafficControl::default() +} diff --git a/src/platform/macos.rs b/src/platform/macos.rs new file mode 100644 index 0000000..53bf94a --- /dev/null +++ b/src/platform/macos.rs @@ -0,0 +1,199 @@ +// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) +// +// SPDX-License-Identifier: MPL-2.0 + +use std::ffi::{CString, NulError}; +use std::io; +use std::mem::size_of; +use std::ptr; + +use thiserror::Error; + +use super::{InterfaceStats, InterfaceStatsProvider}; + +const ROUTE_MESSAGE_PREFIX_LEN: usize = size_of::() + 2; +const SYSCTL_FETCH_ATTEMPTS: usize = 3; + +#[derive(Debug, Error)] +pub enum MacOsInterfaceStatsError { + #[error("interface name contains a NUL byte")] + InterfaceNameContainsNul(#[source] NulError), + + #[error("interface `{interface}` was not found")] + InterfaceNotFound { + interface: String, + #[source] + source: io::Error, + }, + + #[error("sysctl failed while {operation}")] + Sysctl { + operation: &'static str, + #[source] + source: io::Error, + }, + + #[error("malformed route-message response: {0}")] + MalformedResponse(String), + + #[error("no RTM_IFINFO2 statistics returned for interface index {interface_index}")] + NoInterfaceStats { interface_index: libc::c_uint }, +} + +#[derive(Debug, Default)] +pub struct MacOsInterfaceStats; + +impl InterfaceStatsProvider for MacOsInterfaceStats { + type Error = MacOsInterfaceStatsError; + + fn read_stats(&mut 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. + let interface_index = unsafe { libc::if_nametoindex(interface_name.as_ptr()) }; + if interface_index == 0 { + return Err(MacOsInterfaceStatsError::InterfaceNotFound { + interface: interface.to_string(), + source: io::Error::last_os_error(), + }); + } + + let response = route_interface_list(interface_index)?; + parse_route_messages(&response, interface_index) + } +} + +fn route_interface_list( + interface_index: libc::c_uint, +) -> Result, MacOsInterfaceStatsError> { + let mut mib = [ + libc::CTL_NET, + libc::PF_ROUTE, + 0, + 0, + libc::NET_RT_IFLIST2, + interface_index as libc::c_int, + ]; + + for attempt in 0..SYSCTL_FETCH_ATTEMPTS { + let mut required_len = 0; + // SAFETY: `mib` has the declared six elements, `required_len` is writable, and the + // null data pointers request only the response size without reading or writing data. + let size_result = unsafe { + libc::sysctl( + mib.as_mut_ptr(), + mib.len() as libc::c_uint, + ptr::null_mut(), + &mut required_len, + ptr::null_mut(), + 0, + ) + }; + if size_result != 0 { + return Err(MacOsInterfaceStatsError::Sysctl { + operation: "querying the interface-list buffer size", + source: io::Error::last_os_error(), + }); + } + + let mut response = vec![0_u8; required_len]; + let mut returned_len = required_len; + // SAFETY: `response` owns `required_len` initialized bytes, `returned_len` describes + // that allocation, and all other pointers remain valid for the duration of the call. + let fetch_result = unsafe { + libc::sysctl( + mib.as_mut_ptr(), + mib.len() as libc::c_uint, + response.as_mut_ptr().cast(), + &mut returned_len, + ptr::null_mut(), + 0, + ) + }; + if fetch_result == 0 { + if returned_len > response.len() { + return Err(MacOsInterfaceStatsError::MalformedResponse(format!( + "sysctl returned length {returned_len} for a {}-byte buffer", + response.len() + ))); + } + response.truncate(returned_len); + return Ok(response); + } + + let source = io::Error::last_os_error(); + let buffer_too_small = matches!(source.raw_os_error(), Some(libc::ENOMEM | libc::ENOBUFS)); + if !buffer_too_small || attempt + 1 == SYSCTL_FETCH_ATTEMPTS { + return Err(MacOsInterfaceStatsError::Sysctl { + operation: "fetching the interface list", + source, + }); + } + } + + unreachable!("the bounded sysctl retry loop always returns") +} + +fn parse_route_messages( + response: &[u8], + requested_interface_index: libc::c_uint, +) -> Result { + let mut offset = 0; + + while offset < response.len() { + let remaining = response.len() - offset; + if remaining < ROUTE_MESSAGE_PREFIX_LEN { + return Err(MacOsInterfaceStatsError::MalformedResponse(format!( + "truncated route-message prefix at offset {offset}: {remaining} bytes remain" + ))); + } + + let message_len = usize::from(u16::from_ne_bytes([response[offset], response[offset + 1]])); + if message_len < ROUTE_MESSAGE_PREFIX_LEN { + return Err(MacOsInterfaceStatsError::MalformedResponse(format!( + "route message at offset {offset} has implausible length {message_len}" + ))); + } + + let message_end = offset.checked_add(message_len).ok_or_else(|| { + MacOsInterfaceStatsError::MalformedResponse(format!( + "route message length overflows at offset {offset}" + )) + })?; + if message_end > response.len() { + return Err(MacOsInterfaceStatsError::MalformedResponse(format!( + "route message at offset {offset} declares length {message_len}, exceeding the {}-byte response", + response.len() + ))); + } + + let message_type = response[offset + 3]; + if message_type == libc::RTM_IFINFO2 as u8 { + if message_len < size_of::() { + return Err(MacOsInterfaceStatsError::MalformedResponse(format!( + "RTM_IFINFO2 message at offset {offset} has length {message_len}, smaller than {}", + size_of::() + ))); + } + + // SAFETY: the bounds checks above prove that a complete `if_msghdr2` lies within + // `response`; route messages are not guaranteed to have Rust alignment, so the value + // is copied with `read_unaligned` instead of creating a reference into the byte slice. + let header = unsafe { + ptr::read_unaligned(response.as_ptr().add(offset).cast::()) + }; + if libc::c_uint::from(header.ifm_index) == requested_interface_index { + return Ok(InterfaceStats { + rx_bytes: header.ifm_data.ifi_ibytes, + tx_bytes: header.ifm_data.ifi_obytes, + }); + } + } + + offset = message_end; + } + + Err(MacOsInterfaceStatsError::NoInterfaceStats { + interface_index: requested_interface_index, + }) +} diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index 824afba..096d190 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -7,13 +7,16 @@ use crate::SHUTDOWN; use crate::metrics::{Metric, MetricsSender}; -use crate::netlink::{Netlink, NetlinkError, Qdisc}; +use crate::platform::{ + InterfaceStatsProvider, PlatformInterfaceStats, PlatformTrafficControl, TrafficControlBackend, + interface_stats_provider, traffic_control_backend, +}; use crate::time::Time; use crate::util::{ArcMutex, ArcRwLock, MutexExt, RwLockExt}; use crate::{Config, ReflectorStats}; use flume::Sender; use log::{debug, info, warn}; -use rustix::thread::ClockId; +use rustix::time::ClockId; use std::collections::HashMap; use std::fs::File; use std::io::Write; @@ -21,7 +24,6 @@ use std::net::IpAddr; use std::sync::atomic::Ordering; use std::thread::sleep; use std::time::{Duration, Instant}; -use thiserror::Error; #[derive(Copy, Clone, Debug, PartialEq)] enum Direction { @@ -29,12 +31,6 @@ enum Direction { Up, } -#[derive(Debug, Error)] -pub enum RatecontrolError { - #[error("Netlink error")] - Netlink(#[from] NetlinkError), -} - #[derive(Copy, Clone, Debug)] pub enum StatsDirection { RX, @@ -51,34 +47,36 @@ fn generate_initial_speeds(base_speed: f64, size: u32) -> Vec { rates } -fn get_interface_stats( - config: &Config, +fn get_interface_stats( + stats_provider: &mut S, + download_interface: &str, + upload_interface: &str, down_direction: StatsDirection, up_direction: StatsDirection, -) -> Result<(i128, i128), RatecontrolError> { - let (down_rx, down_tx) = Netlink::get_interface_stats(config.download_interface.as_str())?; - let (up_rx, up_tx) = Netlink::get_interface_stats(config.upload_interface.as_str())?; +) -> Result<(i128, i128), S::Error> { + let down = stats_provider.read_stats(download_interface)?; + let up = stats_provider.read_stats(upload_interface)?; let rx_bytes = match down_direction { - StatsDirection::RX => down_rx, - StatsDirection::TX => down_tx, + StatsDirection::RX => down.rx_bytes, + StatsDirection::TX => down.tx_bytes, }; let tx_bytes = match up_direction { - StatsDirection::RX => up_rx, - StatsDirection::TX => up_tx, + StatsDirection::RX => up.rx_bytes, + StatsDirection::TX => up.tx_bytes, }; Ok((rx_bytes.into(), tx_bytes.into())) } #[derive(Clone, Debug)] -struct State { +struct State { current_bytes: i128, current_rate: f64, delta_stat: f64, deltas: Vec, - qdisc: Qdisc, + shaper: H, load: f64, next_rate: f64, nrate: usize, @@ -88,8 +86,8 @@ struct State { utilisation: f64, } -impl State { - fn new(qdisc: Qdisc, previous_bytes: i128, safe_rates: Vec) -> Self { +impl State { + fn new(shaper: H, previous_bytes: i128, safe_rates: Vec) -> Self { State { current_bytes: 0, current_rate: 0.0, @@ -98,7 +96,7 @@ impl State { load: 0.0, next_rate: 0.0, nrate: 0, - qdisc, + shaper, previous_bytes, prev_t: Instant::now(), safe_rates, @@ -107,20 +105,84 @@ impl State { } } -pub struct Ratecontroller { +pub struct Ratecontroller { config: Config, down_direction: StatsDirection, owd_baseline: ArcMutex>, owd_recent: ArcMutex>, reflectors_lock: ArcRwLock>, reselect_trigger: Sender, - state_dl: State, - state_ul: State, + state_dl: State, + state_ul: State, + stats_provider: S, + traffic_control: T, up_direction: StatsDirection, metrics: MetricsSender, } -impl Ratecontroller { +impl Ratecontroller { + fn new_with_backends( + config: Config, + owd_baseline: ArcMutex>, + owd_recent: ArcMutex>, + reflectors_lock: ArcRwLock>, + reselect_trigger: Sender, + down_direction: StatsDirection, + up_direction: StatsDirection, + metrics: MetricsSender, + mut stats_provider: S, + mut traffic_control: T, + ) -> anyhow::Result { + let dl_shaper = traffic_control.find_shaper(config.download_interface.as_str())?; + let dl_safe_rates = + generate_initial_speeds(config.download_base_kbits, config.speed_hist_size); + let ul_shaper = traffic_control.find_shaper(config.upload_interface.as_str())?; + let ul_safe_rates = + generate_initial_speeds(config.upload_base_kbits, config.speed_hist_size); + + let (cur_rx, cur_tx) = get_interface_stats( + &mut stats_provider, + &config.download_interface, + &config.upload_interface, + down_direction, + up_direction, + )?; + + Ok(Self { + config, + down_direction, + owd_baseline, + owd_recent, + 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), + stats_provider, + traffic_control, + up_direction, + metrics, + }) + } + + fn request_initial_rates(&mut self) -> anyhow::Result<()> { + // Set rates to 60% of base rate to make sure we start with sane baselines. + self.state_dl.current_rate = self.config.download_base_kbits * 0.6; + self.state_ul.current_rate = self.config.upload_base_kbits * 0.6; + + self.traffic_control.set_rate( + &self.state_dl.shaper, + self.state_dl.current_rate.round() as u64, + self.config.dry_run, + )?; + self.traffic_control.set_rate( + &self.state_ul.shaper, + self.state_ul.current_rate.round() as u64, + self.config.dry_run, + )?; + + Ok(()) + } + fn calculate_rate(&mut self, direction: Direction) -> anyhow::Result<()> { let (base_rate, delay_ms, min_rate, state) = if direction == Direction::Down { ( @@ -251,7 +313,9 @@ impl Ratecontroller { Ok(()) } +} +impl Ratecontroller { pub fn new( config: Config, owd_baseline: ArcMutex>, @@ -262,49 +326,29 @@ impl Ratecontroller { up_direction: StatsDirection, metrics: MetricsSender, ) -> anyhow::Result { - let dl_qdisc = Netlink::qdisc_from_ifname(config.download_interface.as_str())?; - let dl_safe_rates = - generate_initial_speeds(config.download_base_kbits, config.speed_hist_size); - let ul_qdisc = Netlink::qdisc_from_ifname(config.upload_interface.as_str())?; - let ul_safe_rates = - generate_initial_speeds(config.upload_base_kbits, config.speed_hist_size); - - let (cur_rx, cur_tx) = get_interface_stats(&config, down_direction, up_direction)?; - - Ok(Self { + Self::new_with_backends( config, - down_direction, owd_baseline, owd_recent, reflectors_lock, reselect_trigger, - state_dl: State::new(dl_qdisc, cur_rx, dl_safe_rates), - state_ul: State::new(ul_qdisc, cur_tx, ul_safe_rates), + down_direction, up_direction, metrics, - }) + interface_stats_provider(), + traffic_control_backend(), + ) } +} +impl Ratecontroller { pub fn run(&mut self) -> anyhow::Result<()> { let sleep_time = Duration::from_secs_f64(self.config.min_change_interval); let mut lastchg_t = Instant::now(); let mut lastdump_t = Instant::now(); - // set qdisc rates to 60% of base rate to make sure we start with sane baselines - self.state_dl.current_rate = self.config.download_base_kbits * 0.6; - self.state_ul.current_rate = self.config.upload_base_kbits * 0.6; - - Netlink::set_qdisc_rate( - self.state_dl.qdisc, - self.state_dl.current_rate.round() as u64, - self.config.dry_run, - )?; - Netlink::set_qdisc_rate( - self.state_ul.qdisc, - self.state_ul.current_rate.round() as u64, - self.config.dry_run, - )?; + self.request_initial_rates()?; let mut speed_hist_fd: Option = None; let mut speed_hist_fd_inner: File; @@ -349,21 +393,13 @@ 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(&self.config, self.down_direction, self.up_direction)?; - if self.state_dl.current_bytes == -1 || self.state_ul.current_bytes == -1 { - warn!( - "One or both Netlink stats could not be read. Skipping rate control algorithm" - ); - self.metrics.send(Metric::Event { - name: "netlink_read_failure", - reason: "", - reflector: None, - tags: &[], - }); - continue; - } - + (self.state_dl.current_bytes, self.state_ul.current_bytes) = get_interface_stats( + &mut self.stats_provider, + &self.config.download_interface, + &self.config.upload_interface, + self.down_direction, + self.up_direction, + )?; self.update_deltas()?; if self.state_dl.deltas.is_empty() || self.state_ul.deltas.is_empty() { @@ -377,13 +413,13 @@ impl Ratecontroller { self.state_dl.next_rate = self.config.download_min_kbits; self.state_ul.next_rate = self.config.upload_min_kbits; - Netlink::set_qdisc_rate( - self.state_dl.qdisc, + self.traffic_control.set_rate( + &self.state_dl.shaper, self.state_dl.next_rate as u64, self.config.dry_run, )?; - Netlink::set_qdisc_rate( - self.state_ul.qdisc, + self.traffic_control.set_rate( + &self.state_ul.shaper, self.state_ul.next_rate as u64, self.config.dry_run, )?; @@ -400,22 +436,22 @@ impl Ratecontroller { || self.state_ul.next_rate != self.state_ul.current_rate { info!( - "Adjusting rates (D/U): {} / {} kbit/s", + "Requesting rates (D/U): {} / {} kbit/s", self.state_dl.next_rate as u64, self.state_ul.next_rate as u64 ); } if self.state_dl.next_rate != self.state_dl.current_rate { - Netlink::set_qdisc_rate( - self.state_dl.qdisc, + self.traffic_control.set_rate( + &self.state_dl.shaper, self.state_dl.next_rate as u64, self.config.dry_run, )?; } if self.state_ul.next_rate != self.state_ul.current_rate { - Netlink::set_qdisc_rate( - self.state_ul.qdisc, + self.traffic_control.set_rate( + &self.state_ul.shaper, self.state_ul.next_rate as u64, self.config.dry_run, )?; diff --git a/src/time.rs b/src/time.rs index ed68c34..4ef9b96 100644 --- a/src/time.rs +++ b/src/time.rs @@ -3,8 +3,7 @@ // SPDX-License-Identifier: MPL-2.0 use rustix::fs::Timespec; -use rustix::thread::ClockId; -use rustix::time::clock_gettime; +use rustix::time::{ClockId, clock_gettime}; pub struct Time { time_s: u64,