Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
09980743c6 | ||
|
|
ea11d9c995 |
No files matched your search
@@ -5,7 +5,6 @@ on:
|
||||
branches: [ "master" ]
|
||||
pull_request:
|
||||
branches: [ "master" ]
|
||||
workflow_dispatch:
|
||||
|
||||
env:
|
||||
CARGO_TERM_COLOR: always
|
||||
@@ -19,8 +18,5 @@ jobs:
|
||||
- uses: actions/checkout@v7
|
||||
- name: Install Mold
|
||||
run: sudo apt install mold
|
||||
- name: Install Rust toolchain
|
||||
uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
|
||||
|
||||
- name: Build
|
||||
run: cargo build --verbose
|
||||
@@ -34,7 +34,7 @@ jobs:
|
||||
actions: read
|
||||
if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }}
|
||||
# If you want to copy this config, highly suggest pinning this version to a release rather than tracking the main branch
|
||||
uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@v2.5.1"
|
||||
uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@v2.5.0"
|
||||
with:
|
||||
# Just scan the root directory and docs, since everything else is fixtures
|
||||
scan-args: |-
|
||||
@@ -47,7 +47,7 @@ jobs:
|
||||
actions: read
|
||||
if: ${{ github.event_name == 'pull_request' || github.event_name == 'merge_group' }}
|
||||
# If you want to copy this config, highly suggest pinning this version to a release rather than tracking the main branch
|
||||
uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@v2.5.1"
|
||||
uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@v2.5.0"
|
||||
with:
|
||||
# Just scan the root directory and docs, since everything else is fixtures
|
||||
scan-args: |-
|
||||
|
||||
Generated
+18
-17
@@ -96,9 +96,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "config"
|
||||
version = "0.15.26"
|
||||
version = "0.15.25"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "790374c727a4cb20877dd2cc99ce7011cbe03b0b8df083acccd37df0d161304d"
|
||||
checksum = "b85f248a4de22d204ceabc6299d89d2c70fbd7f09fea53c06c852369652d8139"
|
||||
dependencies = [
|
||||
"pathdiff",
|
||||
"serde_core",
|
||||
@@ -286,9 +286,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "log"
|
||||
version = "0.4.34"
|
||||
version = "0.4.33"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
|
||||
checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad"
|
||||
|
||||
[[package]]
|
||||
name = "maybe-async"
|
||||
@@ -315,15 +315,15 @@ checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
|
||||
|
||||
[[package]]
|
||||
name = "netlink-bindings"
|
||||
version = "0.3.4"
|
||||
version = "0.3.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2a2a0dd8beef757878b977fd3d662c82a93dac4bf7632f42577ca8f8f4d701a0"
|
||||
checksum = "114816d03a633cef7cfa951721dafbda16bb1ff6af83393ee568ec2d4e4ded2d"
|
||||
|
||||
[[package]]
|
||||
name = "netlink-socket2"
|
||||
version = "0.3.4"
|
||||
version = "0.3.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5c7ddb0ea7aa692947d5105160acb7c75c17bc5ec6d5a6263440ca7db4be9a34"
|
||||
checksum = "11193969f046efafdc9d8670d395a6ddcf9b806169f504e520a35dc95cab3241"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"maybe-async",
|
||||
@@ -490,7 +490,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.4",
|
||||
"syn 3.0.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -563,6 +563,7 @@ dependencies = [
|
||||
"rust-uci",
|
||||
"rustix",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"thiserror",
|
||||
"uci-derive",
|
||||
]
|
||||
@@ -580,9 +581,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "syn"
|
||||
version = "3.0.4"
|
||||
version = "3.0.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e6275cddf4610d1775e6d1fe9469b2e77d0f39fd98fb7450901b821e0c53649f"
|
||||
checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -606,22 +607,22 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "thiserror"
|
||||
version = "2.0.20"
|
||||
version = "2.0.19"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f"
|
||||
checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9"
|
||||
dependencies = [
|
||||
"thiserror-impl",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "thiserror-impl"
|
||||
version = "2.0.20"
|
||||
version = "2.0.19"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af"
|
||||
checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.4",
|
||||
"syn 3.0.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -684,7 +685,7 @@ version = "0.1.0"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 3.0.4",
|
||||
"syn 3.0.3",
|
||||
"trybuild",
|
||||
]
|
||||
|
||||
|
||||
+3
-1
@@ -29,6 +29,7 @@ flume = { version = "0.12.0", default-features = false, features = ["eventual-f
|
||||
config = { version = "0.15.25", default-features = false }
|
||||
csv = "1.4.0"
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = { version = "1", optional = true }
|
||||
uci-derive = { path = "crates/uci-derive" }
|
||||
|
||||
[dependencies.log]
|
||||
@@ -43,10 +44,11 @@ netlink-socket2 = { version = "0.3.3" }
|
||||
rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "fcfedd204fb76928ef617bf9631b041c816e03d4", optional = true, default-features = false, }
|
||||
|
||||
[features]
|
||||
default = ["toml", "uci-vendored"]
|
||||
default = ["toml", "trace", "uci-vendored"]
|
||||
toml = ["config/toml"]
|
||||
uci = ["dep:rust-uci"]
|
||||
uci-vendored = ["uci", "rust-uci/vendored"]
|
||||
trace = ["dep:serde_json"]
|
||||
|
||||
[workspace]
|
||||
members = ["crates/uci-derive"]
|
||||
|
||||
@@ -10,6 +10,14 @@
|
||||
* **Configurable via environment or UCI**: reads settings from environment variables or OpenWRT UCI, with sensible defaults and type‑safe parsing .
|
||||
* **Minimal dependencies**: relies on Rust crates only (anyhow, socket2, neli, rustix), no external scripts .
|
||||
|
||||
## Legacy control trace v1
|
||||
|
||||
The additive legacy characterization recorder is excluded from default builds. Build with
|
||||
`--features trace` and set `SQMA_TRACE_V1_PATH` to a JSON Lines output path to enable it at
|
||||
runtime. The format is versioned as trace v1 and is intended only for replaying the current
|
||||
baseliner and rate-controller boundary during the `SqmControlLaw` extraction; it is not the
|
||||
final runtime event schema. Existing CSV and metrics outputs remain unchanged.
|
||||
|
||||
## Installation
|
||||
|
||||
You can install **sqm‑autorate‑rust** in two ways:
|
||||
|
||||
@@ -2,18 +2,5 @@
|
||||
"$schema": "https://docs.renovatebot.com/renovate-schema.json",
|
||||
"extends": [
|
||||
"config:recommended"
|
||||
],
|
||||
"packageRules": [
|
||||
{
|
||||
"description": "Keep netlink-bindings and netlink-socket2 updates together",
|
||||
"matchManagers": [
|
||||
"cargo"
|
||||
],
|
||||
"matchPackageNames": [
|
||||
"netlink-bindings",
|
||||
"netlink-socket2"
|
||||
],
|
||||
"groupName": "netlink dependencies"
|
||||
}
|
||||
]
|
||||
}
|
||||
+16
-1
@@ -9,6 +9,7 @@ use crate::SHUTDOWN;
|
||||
use crate::metrics::{Metric, MetricsSender};
|
||||
use crate::pinger::PingReply;
|
||||
use crate::settings::Settings;
|
||||
use crate::trace::{Recorder, v1};
|
||||
use flume::{Receiver, Sender};
|
||||
use log::{debug, info};
|
||||
use std::collections::HashMap;
|
||||
@@ -78,6 +79,7 @@ pub struct Baseliner {
|
||||
pub selection_tx: Option<Sender<ControlSnapshot>>,
|
||||
pub baseline_metrics: MetricsSender,
|
||||
pub event_metrics: MetricsSender,
|
||||
pub trace: Recorder,
|
||||
}
|
||||
|
||||
fn ewma_factor(tick: f64, dur: f64) -> f64 {
|
||||
@@ -207,10 +209,23 @@ impl Baseliner {
|
||||
);
|
||||
|
||||
let snapshot = control_snapshot(&reflectors, Instant::now());
|
||||
self.trace.linearize(|trace| {
|
||||
if trace.is_enabled() {
|
||||
trace.record_at(
|
||||
reply.last_receive_time_s,
|
||||
v1::Event::PingReply {
|
||||
reflector: reply.reflector,
|
||||
down_time_ms: reply.down_time,
|
||||
up_time_ms: reply.up_time,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
if let Some(selection_tx) = &self.selection_tx {
|
||||
let _ = selection_tx.send(snapshot.clone());
|
||||
}
|
||||
self.control_tx.send(snapshot)?;
|
||||
self.control_tx.send(snapshot)
|
||||
})?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+7
-1
@@ -18,6 +18,7 @@ mod ratecontroller;
|
||||
mod reflector_selector;
|
||||
mod settings;
|
||||
mod time;
|
||||
mod trace;
|
||||
mod util;
|
||||
|
||||
use crate::baseliner::Baseliner;
|
||||
@@ -30,7 +31,7 @@ use crate::ratecontroller::Ratecontroller;
|
||||
use crate::reflector_selector::ReflectorSelector;
|
||||
use crate::settings::MeasurementType;
|
||||
use crate::settings::Settings;
|
||||
use ::log::{info, warn};
|
||||
use ::log::{debug, info, warn};
|
||||
use flume::RecvTimeoutError;
|
||||
use std::collections::HashMap;
|
||||
use std::net::IpAddr;
|
||||
@@ -243,6 +244,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
|
||||
}
|
||||
|
||||
let start_time = Instant::now();
|
||||
let trace = trace::Recorder::from_env(start_time, settings);
|
||||
let probe_id = probe_identifier();
|
||||
|
||||
let ReflectorSetup {
|
||||
@@ -334,6 +336,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
|
||||
selection_tx: selection_snapshot_tx,
|
||||
baseline_metrics,
|
||||
event_metrics: event_metrics.clone(),
|
||||
trace: trace.clone(),
|
||||
};
|
||||
|
||||
let mut main_traffic_control = traffic_control_backend();
|
||||
@@ -417,6 +420,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
|
||||
reflector_peers,
|
||||
reselect_tx,
|
||||
rate_metrics,
|
||||
trace.clone(),
|
||||
)?;
|
||||
|
||||
let err_tx = error_tx.clone();
|
||||
@@ -460,6 +464,8 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
|
||||
&up_shaper,
|
||||
);
|
||||
|
||||
trace.finish();
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
|
||||
+127
-11
@@ -14,6 +14,7 @@ use crate::platform::{
|
||||
};
|
||||
use crate::settings::Settings;
|
||||
use crate::time::Time;
|
||||
use crate::trace::{Recorder, v1};
|
||||
use crate::util::{ArcRwLock, RwLockExt};
|
||||
use flume::{Receiver, Sender};
|
||||
use log::{debug, info, warn};
|
||||
@@ -145,9 +146,11 @@ pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
|
||||
state_dl: State<T::Handle>,
|
||||
state_ul: State<T::Handle>,
|
||||
rng: fastrand::Rng,
|
||||
next_evaluation_id: u64,
|
||||
stats_provider: S,
|
||||
traffic_control: T,
|
||||
metrics: MetricsSender,
|
||||
trace: Recorder,
|
||||
}
|
||||
|
||||
impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
@@ -160,6 +163,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
mut stats_provider: S,
|
||||
mut traffic_control: T,
|
||||
mut rng: fastrand::Rng,
|
||||
trace: Recorder,
|
||||
) -> anyhow::Result<Self> {
|
||||
let dl_shaper =
|
||||
traffic_control.find_shaper(settings.network.download_interface.as_str())?;
|
||||
@@ -178,7 +182,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
let (cur_rx, cur_tx) =
|
||||
get_interface_stats(&mut stats_provider, &settings.network.upload_interface)?;
|
||||
|
||||
Ok(Self {
|
||||
let controller = Self {
|
||||
settings,
|
||||
control_snapshot: None,
|
||||
control_snapshot_rx,
|
||||
@@ -187,10 +191,32 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
state_dl: State::new(dl_shaper, cur_rx, dl_safe_rates),
|
||||
state_ul: State::new(ul_shaper, cur_tx, ul_safe_rates),
|
||||
rng,
|
||||
next_evaluation_id: 0,
|
||||
stats_provider,
|
||||
traffic_control,
|
||||
metrics,
|
||||
})
|
||||
trace,
|
||||
};
|
||||
|
||||
if controller.trace.is_enabled() {
|
||||
controller.trace.record_at(
|
||||
Instant::now(),
|
||||
v1::Event::ControllerInitialized {
|
||||
previous_rx_bytes: controller.state_dl.previous_bytes,
|
||||
previous_tx_bytes: controller.state_ul.previous_bytes,
|
||||
download_prev_mono_ns: controller
|
||||
.trace
|
||||
.monotonic_ns(controller.state_dl.prev_t),
|
||||
upload_prev_mono_ns: controller.trace.monotonic_ns(controller.state_ul.prev_t),
|
||||
download_history_index: controller.state_dl.nrate as u64,
|
||||
upload_history_index: controller.state_ul.nrate as u64,
|
||||
download_safe_rates_kbit: controller.state_dl.safe_rates.clone(),
|
||||
upload_safe_rates_kbit: controller.state_ul.safe_rates.clone(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
Ok(controller)
|
||||
}
|
||||
|
||||
fn request_initial_rates(&mut self) -> anyhow::Result<()> {
|
||||
@@ -198,6 +224,20 @@ 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;
|
||||
|
||||
if self.trace.is_enabled() {
|
||||
self.trace.record_at(
|
||||
Instant::now(),
|
||||
v1::Event::RequestedRates {
|
||||
evaluation_id: None,
|
||||
reason: v1::RateRequestReason::Startup,
|
||||
download_kbit: self.state_dl.current_rate.round() as u64,
|
||||
upload_kbit: self.state_ul.current_rate.round() as u64,
|
||||
download_requested: true,
|
||||
upload_requested: true,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
self.traffic_control.set_rate(
|
||||
&self.state_dl.shaper,
|
||||
self.state_dl.current_rate.round() as u64,
|
||||
@@ -212,7 +252,8 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn calculate_rate(&mut self, direction: Direction) -> anyhow::Result<()> {
|
||||
fn calculate_rate(&mut self, direction: Direction, evaluation_id: u64) -> anyhow::Result<()> {
|
||||
let trace = self.trace.clone();
|
||||
let (base_rate, delay_ms, min_rate, state) = if direction == Direction::Down {
|
||||
(
|
||||
self.settings.network.download_base_kbits,
|
||||
@@ -231,6 +272,20 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
|
||||
let now_t = Instant::now();
|
||||
let dur = now_t.duration_since(state.prev_t);
|
||||
let trace_enabled = trace.is_enabled();
|
||||
let trace_direction = match direction {
|
||||
Direction::Down => v1::Direction::Download,
|
||||
Direction::Up => v1::Direction::Upload,
|
||||
};
|
||||
if trace_enabled {
|
||||
trace.record_at(
|
||||
now_t,
|
||||
v1::Event::RateCalculation {
|
||||
evaluation_id,
|
||||
direction: trace_direction,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
if !state.deltas.is_empty() {
|
||||
state.next_rate = state.current_rate;
|
||||
@@ -269,11 +324,20 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
}
|
||||
|
||||
if state.delta_stat > delay_ms {
|
||||
match state
|
||||
.safe_rates
|
||||
.get(self.rng.usize(..state.safe_rates.len()))
|
||||
{
|
||||
let selected_index = self.rng.usize(..state.safe_rates.len());
|
||||
match state.safe_rates.get(selected_index) {
|
||||
Some(rnd_rate) => {
|
||||
if trace_enabled {
|
||||
trace.record_at(
|
||||
Instant::now(),
|
||||
v1::Event::RandomSafeRateChoice {
|
||||
evaluation_id,
|
||||
direction: trace_direction,
|
||||
index: selected_index as u64,
|
||||
rate_kbit: *rnd_rate,
|
||||
},
|
||||
);
|
||||
}
|
||||
state.next_rate = rnd_rate.min(0.9 * state.current_rate * state.load);
|
||||
}
|
||||
None => {
|
||||
@@ -349,7 +413,9 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
&mut self.stats_provider,
|
||||
&self.settings.network.upload_interface,
|
||||
)?;
|
||||
self.update_deltas()?;
|
||||
let evaluation_id = self.next_evaluation_id;
|
||||
self.next_evaluation_id = self.next_evaluation_id.wrapping_add(1);
|
||||
self.update_deltas(evaluation_id, now_t)?;
|
||||
|
||||
if self.state_dl.deltas.is_empty() || self.state_ul.deltas.is_empty() {
|
||||
warn!("No reflector data available, dropping to minimum rates");
|
||||
@@ -362,6 +428,20 @@ 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();
|
||||
|
||||
if self.trace.is_enabled() {
|
||||
self.trace.record_at(
|
||||
Instant::now(),
|
||||
v1::Event::RequestedRates {
|
||||
evaluation_id: Some(evaluation_id),
|
||||
reason: v1::RateRequestReason::NoReflectorData,
|
||||
download_kbit: self.state_dl.next_rate as u64,
|
||||
upload_kbit: self.state_ul.next_rate as u64,
|
||||
download_requested: true,
|
||||
upload_requested: true,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
self.traffic_control.set_rate(
|
||||
&self.state_dl.shaper,
|
||||
self.state_dl.next_rate as u64,
|
||||
@@ -378,8 +458,24 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
continue;
|
||||
}
|
||||
|
||||
self.calculate_rate(Direction::Down)?;
|
||||
self.calculate_rate(Direction::Up)?;
|
||||
self.calculate_rate(Direction::Down, evaluation_id)?;
|
||||
self.calculate_rate(Direction::Up, evaluation_id)?;
|
||||
|
||||
if self.trace.is_enabled() {
|
||||
let download_requested = self.state_dl.next_rate != self.state_dl.current_rate;
|
||||
let upload_requested = self.state_ul.next_rate != self.state_ul.current_rate;
|
||||
self.trace.record_at(
|
||||
Instant::now(),
|
||||
v1::Event::RequestedRates {
|
||||
evaluation_id: Some(evaluation_id),
|
||||
reason: v1::RateRequestReason::Control,
|
||||
download_kbit: self.state_dl.next_rate as u64,
|
||||
upload_kbit: self.state_ul.next_rate as u64,
|
||||
download_requested,
|
||||
upload_requested,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
if self.state_dl.next_rate != self.state_dl.current_rate
|
||||
|| self.state_ul.next_rate != self.state_ul.current_rate
|
||||
@@ -479,7 +575,9 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
}
|
||||
}
|
||||
|
||||
fn update_deltas(&mut self) -> anyhow::Result<()> {
|
||||
fn update_deltas(&mut self, evaluation_id: u64, loop_time: Instant) -> anyhow::Result<()> {
|
||||
let recorder = self.trace.clone();
|
||||
recorder.linearize(|trace| -> anyhow::Result<()> {
|
||||
drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot);
|
||||
let now_t = Instant::now();
|
||||
let reflectors = self.reflectors_lock.read_anyhow()?;
|
||||
@@ -492,6 +590,22 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
|
||||
self.state_dl.deltas = down_deltas;
|
||||
self.state_ul.deltas = up_deltas;
|
||||
|
||||
if trace.is_enabled() {
|
||||
trace.record_at(
|
||||
now_t,
|
||||
v1::Event::ControlEvaluation {
|
||||
evaluation_id,
|
||||
loop_mono_ns: self.trace.monotonic_ns(loop_time),
|
||||
rx_bytes: self.state_dl.current_bytes,
|
||||
tx_bytes: self.state_ul.current_bytes,
|
||||
active_reflectors: reflectors.clone(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
})?;
|
||||
|
||||
if self.state_dl.deltas.len() < 5 || self.state_ul.deltas.len() < 5 {
|
||||
// trigger reselection
|
||||
warn!(
|
||||
@@ -513,6 +627,7 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
|
||||
reflectors_lock: ArcRwLock<Vec<IpAddr>>,
|
||||
reselect_trigger: Sender<bool>,
|
||||
metrics: MetricsSender,
|
||||
trace: Recorder,
|
||||
) -> anyhow::Result<Self> {
|
||||
Self::new_with_backends(
|
||||
settings,
|
||||
@@ -523,6 +638,7 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
|
||||
interface_stats_provider(),
|
||||
traffic_control_backend(),
|
||||
fastrand::Rng::new(),
|
||||
trace,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,417 @@
|
||||
// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
|
||||
//
|
||||
// SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
pub(crate) mod v1;
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
mod writer;
|
||||
|
||||
use crate::settings::Settings;
|
||||
use std::time::Instant;
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
use std::time::SystemTime;
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
use flume::Sender;
|
||||
#[cfg(feature = "trace")]
|
||||
use log::warn;
|
||||
#[cfg(feature = "trace")]
|
||||
use std::fs::File;
|
||||
#[cfg(feature = "trace")]
|
||||
use std::io::{BufWriter, Write};
|
||||
#[cfg(feature = "trace")]
|
||||
use std::path::Path;
|
||||
#[cfg(feature = "trace")]
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
#[cfg(feature = "trace")]
|
||||
use std::sync::{Arc, Mutex};
|
||||
#[cfg(feature = "trace")]
|
||||
use std::thread::JoinHandle;
|
||||
|
||||
const TRACE_PATH_ENV: &str = "SQMA_TRACE_V1_PATH";
|
||||
#[cfg(feature = "trace")]
|
||||
const QUEUE_CAPACITY: usize = 4_096;
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
pub(crate) struct Recorder {
|
||||
#[cfg(feature = "trace")]
|
||||
inner: Option<Arc<Inner>>,
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
struct Inner {
|
||||
origin: Instant,
|
||||
records: Sender<v1::Record>,
|
||||
next_seq: Mutex<u64>,
|
||||
healthy: Arc<AtomicBool>,
|
||||
writer: Mutex<Option<JoinHandle<()>>>,
|
||||
}
|
||||
|
||||
pub(crate) struct OrderedRecorder<'a> {
|
||||
#[cfg(feature = "trace")]
|
||||
inner: Option<&'a Inner>,
|
||||
#[cfg(feature = "trace")]
|
||||
next_seq: Option<&'a mut u64>,
|
||||
#[cfg(not(feature = "trace"))]
|
||||
marker: std::marker::PhantomData<&'a ()>,
|
||||
}
|
||||
|
||||
impl Recorder {
|
||||
pub(crate) fn from_env(origin: Instant, settings: &Settings) -> Self {
|
||||
let Some(path) = std::env::var_os(TRACE_PATH_ENV) else {
|
||||
return Self::disabled();
|
||||
};
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
{
|
||||
let header = v1::Event::Header {
|
||||
format: v1::FORMAT_NAME.to_owned(),
|
||||
version: v1::FORMAT_VERSION,
|
||||
program_version: env!("CARGO_PKG_VERSION").to_owned(),
|
||||
control: v1::ControlConfig::from_settings(settings),
|
||||
};
|
||||
Self::start_path(Path::new(&path), origin, header)
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "trace"))]
|
||||
{
|
||||
let _ = (origin, settings, path);
|
||||
log::warn!(
|
||||
"{TRACE_PATH_ENV} is set, but this binary was built without the `trace` feature"
|
||||
);
|
||||
Self::disabled()
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn disabled() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
pub(crate) fn is_enabled(&self) -> bool {
|
||||
#[cfg(feature = "trace")]
|
||||
{
|
||||
self.inner
|
||||
.as_ref()
|
||||
.is_some_and(|inner| inner.healthy.load(Ordering::Acquire))
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "trace"))]
|
||||
{
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn monotonic_ns(&self, instant: Instant) -> u64 {
|
||||
#[cfg(feature = "trace")]
|
||||
{
|
||||
self.inner
|
||||
.as_ref()
|
||||
.map_or(0, |inner| v1::monotonic_ns(inner.origin, instant))
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "trace"))]
|
||||
{
|
||||
let _ = instant;
|
||||
0
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn record_at(&self, instant: Instant, event: v1::Event) -> Option<u64> {
|
||||
self.linearize(|recorder| recorder.record_at(instant, event))
|
||||
}
|
||||
|
||||
pub(crate) fn linearize<R>(&self, callback: impl FnOnce(&mut OrderedRecorder<'_>) -> R) -> R {
|
||||
#[cfg(feature = "trace")]
|
||||
{
|
||||
if self
|
||||
.inner
|
||||
.as_ref()
|
||||
.is_some_and(|inner| inner.healthy.load(Ordering::Acquire))
|
||||
{
|
||||
let inner = self
|
||||
.inner
|
||||
.as_ref()
|
||||
.expect("healthy trace recorder must have inner state");
|
||||
match inner.next_seq.lock() {
|
||||
Ok(mut next_seq) => {
|
||||
if inner.healthy.load(Ordering::Acquire) {
|
||||
let mut recorder = OrderedRecorder {
|
||||
inner: Some(inner),
|
||||
next_seq: Some(&mut next_seq),
|
||||
};
|
||||
return callback(&mut recorder);
|
||||
}
|
||||
}
|
||||
Err(_) => {
|
||||
if inner.healthy.swap(false, Ordering::AcqRel) {
|
||||
warn!("Legacy control trace sequencing lock failed; tracing disabled");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut recorder = OrderedRecorder {
|
||||
inner: None,
|
||||
next_seq: None,
|
||||
};
|
||||
callback(&mut recorder)
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "trace"))]
|
||||
{
|
||||
let mut recorder = OrderedRecorder {
|
||||
marker: std::marker::PhantomData,
|
||||
};
|
||||
callback(&mut recorder)
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn finish(&self) {
|
||||
#[cfg(feature = "trace")]
|
||||
if let Some(inner) = self.inner.as_ref() {
|
||||
self.linearize(|recorder| {
|
||||
recorder.record_at(Instant::now(), v1::Event::End { clean: true });
|
||||
inner.healthy.store(false, Ordering::Release);
|
||||
});
|
||||
|
||||
let writer = match inner.writer.lock() {
|
||||
Ok(mut writer) => writer.take(),
|
||||
Err(_) => {
|
||||
warn!("Legacy control trace writer lock failed during shutdown");
|
||||
None
|
||||
}
|
||||
};
|
||||
let writer_panicked = writer.map(|writer| writer.join().is_err()).unwrap_or(false);
|
||||
if writer_panicked {
|
||||
warn!("Legacy control trace writer panicked during shutdown");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
fn start_path(path: &Path, origin: Instant, header: v1::Event) -> Self {
|
||||
let file = match File::options()
|
||||
.create(true)
|
||||
.write(true)
|
||||
.truncate(true)
|
||||
.open(path)
|
||||
{
|
||||
Ok(file) => file,
|
||||
Err(error) => {
|
||||
warn!(
|
||||
"Failed to open legacy control trace at {}: {error}; tracing disabled",
|
||||
path.display()
|
||||
);
|
||||
return Self::disabled();
|
||||
}
|
||||
};
|
||||
|
||||
Self::start_writer(BufWriter::new(file), origin, header)
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
fn start_writer<W>(writer: W, origin: Instant, header: v1::Event) -> Self
|
||||
where
|
||||
W: Write + Send + 'static,
|
||||
{
|
||||
let (records, record_rx) = flume::bounded(QUEUE_CAPACITY);
|
||||
let healthy = Arc::new(AtomicBool::new(true));
|
||||
let header = v1::Record {
|
||||
seq: 0,
|
||||
timestamp: v1::timestamp(origin, origin, SystemTime::now()),
|
||||
event: header,
|
||||
};
|
||||
let writer = match writer::spawn(writer, record_rx, header, Arc::clone(&healthy)) {
|
||||
Ok(writer) => writer,
|
||||
Err(error) => {
|
||||
warn!("Failed to start legacy control trace writer: {error}; tracing disabled");
|
||||
return Self::disabled();
|
||||
}
|
||||
};
|
||||
|
||||
Self {
|
||||
inner: Some(Arc::new(Inner {
|
||||
origin,
|
||||
records,
|
||||
next_seq: Mutex::new(1),
|
||||
healthy,
|
||||
writer: Mutex::new(Some(writer)),
|
||||
})),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl OrderedRecorder<'_> {
|
||||
pub(crate) fn is_enabled(&self) -> bool {
|
||||
#[cfg(feature = "trace")]
|
||||
{
|
||||
self.inner.is_some()
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "trace"))]
|
||||
{
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn record_at(&mut self, instant: Instant, event: v1::Event) -> Option<u64> {
|
||||
#[cfg(feature = "trace")]
|
||||
{
|
||||
let inner = self.inner?;
|
||||
let next_seq = self.next_seq.as_deref_mut()?;
|
||||
let seq = *next_seq;
|
||||
let record = v1::Record {
|
||||
seq,
|
||||
timestamp: v1::timestamp(inner.origin, instant, SystemTime::now()),
|
||||
event,
|
||||
};
|
||||
|
||||
if inner.records.send(record).is_err() {
|
||||
if inner.healthy.swap(false, Ordering::AcqRel) {
|
||||
warn!("Legacy control trace channel closed; tracing disabled");
|
||||
}
|
||||
return None;
|
||||
}
|
||||
|
||||
*next_seq = (*next_seq).saturating_add(1);
|
||||
Some(seq)
|
||||
}
|
||||
|
||||
#[cfg(not(feature = "trace"))]
|
||||
{
|
||||
let _ = (instant, event);
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn disabled_recorder_is_a_no_op() {
|
||||
let recorder = Recorder::disabled();
|
||||
|
||||
assert!(!recorder.is_enabled());
|
||||
assert_eq!(
|
||||
recorder.record_at(Instant::now(), v1::Event::End { clean: true }),
|
||||
None
|
||||
);
|
||||
recorder.finish();
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
mod enabled {
|
||||
use super::*;
|
||||
use std::io;
|
||||
use std::net::{IpAddr, Ipv4Addr};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Barrier;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::thread;
|
||||
|
||||
static NEXT_PATH_ID: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
struct FailingWriter;
|
||||
|
||||
impl Write for FailingWriter {
|
||||
fn write(&mut self, _buffer: &[u8]) -> io::Result<usize> {
|
||||
Err(io::Error::other("injected writer failure"))
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
Err(io::Error::other("injected writer failure"))
|
||||
}
|
||||
}
|
||||
|
||||
fn temp_path() -> PathBuf {
|
||||
let id = NEXT_PATH_ID.fetch_add(1, Ordering::Relaxed);
|
||||
std::env::temp_dir().join(format!(
|
||||
"sqm-autorate-trace-v1-{}-{id}.jsonl",
|
||||
std::process::id()
|
||||
))
|
||||
}
|
||||
|
||||
fn ping_event(reflector: Ipv4Addr, value: f64) -> v1::Event {
|
||||
v1::Event::PingReply {
|
||||
reflector: IpAddr::V4(reflector),
|
||||
down_time_ms: value,
|
||||
up_time_ms: value,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn writer_preserves_global_sequence_order_from_multiple_producers() {
|
||||
let path = temp_path();
|
||||
let origin = Instant::now();
|
||||
let recorder = Recorder::start_path(&path, origin, v1::tests::header_event());
|
||||
let barrier = Arc::new(Barrier::new(3));
|
||||
let mut producers = Vec::new();
|
||||
|
||||
for producer in 1..=2 {
|
||||
let recorder = recorder.clone();
|
||||
let barrier = Arc::clone(&barrier);
|
||||
producers.push(thread::spawn(move || {
|
||||
barrier.wait();
|
||||
for value in 0..50 {
|
||||
recorder.record_at(
|
||||
origin,
|
||||
ping_event(Ipv4Addr::new(192, 0, 2, producer), value as f64),
|
||||
);
|
||||
}
|
||||
}));
|
||||
}
|
||||
|
||||
barrier.wait();
|
||||
for producer in producers {
|
||||
producer.join().unwrap();
|
||||
}
|
||||
recorder.finish();
|
||||
|
||||
let contents = std::fs::read_to_string(&path).unwrap();
|
||||
let records = contents
|
||||
.lines()
|
||||
.map(|line| serde_json::from_str::<v1::Record>(line).unwrap())
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
assert_eq!(records.len(), 102);
|
||||
assert!(matches!(
|
||||
&records.first().unwrap().event,
|
||||
v1::Event::Header { .. }
|
||||
));
|
||||
assert!(matches!(
|
||||
&records.last().unwrap().event,
|
||||
v1::Event::End { clean: true }
|
||||
));
|
||||
for (expected, record) in records.iter().enumerate() {
|
||||
assert_eq!(record.seq, expected as u64);
|
||||
}
|
||||
|
||||
std::fs::remove_file(path).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn writer_failure_disables_recorder_without_propagating() {
|
||||
let recorder =
|
||||
Recorder::start_writer(FailingWriter, Instant::now(), v1::tests::header_event());
|
||||
|
||||
recorder.record_at(Instant::now(), ping_event(Ipv4Addr::new(192, 0, 2, 1), 1.0));
|
||||
recorder.finish();
|
||||
|
||||
assert!(!recorder.is_enabled());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn open_failure_returns_disabled_recorder() {
|
||||
let path = temp_path().join("missing-parent").join("trace.jsonl");
|
||||
let recorder = Recorder::start_path(&path, Instant::now(), v1::tests::header_event());
|
||||
|
||||
assert!(!recorder.is_enabled());
|
||||
recorder.record_at(Instant::now(), v1::Event::End { clean: true });
|
||||
recorder.finish();
|
||||
}
|
||||
}
|
||||
}
|
||||
+234
@@ -0,0 +1,234 @@
|
||||
// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
|
||||
//
|
||||
// SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::net::IpAddr;
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
use crate::settings::Settings;
|
||||
#[cfg(any(feature = "trace", test))]
|
||||
use std::time::Duration;
|
||||
#[cfg(any(feature = "trace", test))]
|
||||
use std::time::Instant;
|
||||
#[cfg(feature = "trace")]
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
pub(crate) const FORMAT_VERSION: u16 = 1;
|
||||
#[cfg(feature = "trace")]
|
||||
pub(crate) const FORMAT_NAME: &str = "legacy-control-trace-v1";
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
|
||||
pub(crate) struct Timestamp {
|
||||
pub(crate) mono_ns: u64,
|
||||
pub(crate) unix_us: i64,
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
|
||||
pub(crate) struct Record {
|
||||
pub(crate) seq: u64,
|
||||
pub(crate) timestamp: Timestamp,
|
||||
#[serde(flatten)]
|
||||
pub(crate) event: Event,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
|
||||
#[serde(tag = "type", rename_all = "snake_case")]
|
||||
pub(crate) enum Event {
|
||||
#[cfg(feature = "trace")]
|
||||
Header {
|
||||
format: String,
|
||||
version: u16,
|
||||
program_version: String,
|
||||
control: ControlConfig,
|
||||
},
|
||||
PingReply {
|
||||
reflector: IpAddr,
|
||||
down_time_ms: f64,
|
||||
up_time_ms: f64,
|
||||
},
|
||||
ControllerInitialized {
|
||||
previous_rx_bytes: i128,
|
||||
previous_tx_bytes: i128,
|
||||
download_prev_mono_ns: u64,
|
||||
upload_prev_mono_ns: u64,
|
||||
download_history_index: u64,
|
||||
upload_history_index: u64,
|
||||
download_safe_rates_kbit: Vec<f64>,
|
||||
upload_safe_rates_kbit: Vec<f64>,
|
||||
},
|
||||
ControlEvaluation {
|
||||
evaluation_id: u64,
|
||||
loop_mono_ns: u64,
|
||||
rx_bytes: i128,
|
||||
tx_bytes: i128,
|
||||
active_reflectors: Vec<IpAddr>,
|
||||
},
|
||||
RateCalculation {
|
||||
evaluation_id: u64,
|
||||
direction: Direction,
|
||||
},
|
||||
RandomSafeRateChoice {
|
||||
evaluation_id: u64,
|
||||
direction: Direction,
|
||||
index: u64,
|
||||
rate_kbit: f64,
|
||||
},
|
||||
RequestedRates {
|
||||
evaluation_id: Option<u64>,
|
||||
reason: RateRequestReason,
|
||||
download_kbit: u64,
|
||||
upload_kbit: u64,
|
||||
download_requested: bool,
|
||||
upload_requested: bool,
|
||||
},
|
||||
End {
|
||||
clean: bool,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub(crate) enum Direction {
|
||||
Download,
|
||||
Upload,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub(crate) enum RateRequestReason {
|
||||
Startup,
|
||||
Control,
|
||||
NoReflectorData,
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
|
||||
pub(crate) struct ControlConfig {
|
||||
pub(crate) tick_interval_secs: f64,
|
||||
pub(crate) min_change_interval_secs: f64,
|
||||
pub(crate) download_base_kbit: f64,
|
||||
pub(crate) upload_base_kbit: f64,
|
||||
pub(crate) download_min_kbit: f64,
|
||||
pub(crate) upload_min_kbit: f64,
|
||||
pub(crate) download_delay_ms: f64,
|
||||
pub(crate) upload_delay_ms: f64,
|
||||
pub(crate) high_load_level: f64,
|
||||
pub(crate) speed_history_size: u32,
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
impl ControlConfig {
|
||||
pub(crate) fn from_settings(settings: &Settings) -> Self {
|
||||
Self {
|
||||
tick_interval_secs: settings.advanced_settings.tick_interval,
|
||||
min_change_interval_secs: settings.advanced_settings.min_change_interval,
|
||||
download_base_kbit: settings.network.download_base_kbits,
|
||||
upload_base_kbit: settings.network.upload_base_kbits,
|
||||
download_min_kbit: settings.network.download_min_kbits(),
|
||||
upload_min_kbit: settings.network.upload_min_kbits(),
|
||||
download_delay_ms: settings.advanced_settings.download_delay_ms,
|
||||
upload_delay_ms: settings.advanced_settings.upload_delay_ms,
|
||||
high_load_level: settings.advanced_settings.high_load_level,
|
||||
speed_history_size: settings.advanced_settings.speed_hist_size,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
pub(crate) fn timestamp(origin: Instant, monotonic: Instant, realtime: SystemTime) -> Timestamp {
|
||||
Timestamp {
|
||||
mono_ns: monotonic_ns(origin, monotonic),
|
||||
unix_us: unix_us(realtime),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(any(feature = "trace", test))]
|
||||
pub(crate) fn monotonic_ns(origin: Instant, monotonic: Instant) -> u64 {
|
||||
duration_ns(monotonic.saturating_duration_since(origin))
|
||||
}
|
||||
|
||||
#[cfg(any(feature = "trace", test))]
|
||||
fn duration_ns(duration: Duration) -> u64 {
|
||||
duration.as_nanos().min(u64::MAX as u128) as u64
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
fn unix_us(realtime: SystemTime) -> i64 {
|
||||
match realtime.duration_since(UNIX_EPOCH) {
|
||||
Ok(duration) => duration
|
||||
.as_micros()
|
||||
.min(i64::MAX as u128)
|
||||
.try_into()
|
||||
.unwrap_or(i64::MAX),
|
||||
Err(error) => {
|
||||
let micros = error.duration().as_micros().min(i64::MAX as u128) as i64;
|
||||
-micros
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(super) mod tests {
|
||||
use super::*;
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
fn control_config() -> ControlConfig {
|
||||
ControlConfig {
|
||||
tick_interval_secs: 0.5,
|
||||
min_change_interval_secs: 0.5,
|
||||
download_base_kbit: 100_000.0,
|
||||
upload_base_kbit: 20_000.0,
|
||||
download_min_kbit: 20_000.0,
|
||||
upload_min_kbit: 4_000.0,
|
||||
download_delay_ms: 15.0,
|
||||
upload_delay_ms: 15.0,
|
||||
high_load_level: 0.8,
|
||||
speed_history_size: 100,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
pub(crate) fn header_event() -> Event {
|
||||
Event::Header {
|
||||
format: FORMAT_NAME.to_owned(),
|
||||
version: FORMAT_VERSION,
|
||||
program_version: "test".to_owned(),
|
||||
control: control_config(),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
#[test]
|
||||
fn serializes_versioned_header() {
|
||||
let origin = Instant::now();
|
||||
let record = Record {
|
||||
seq: 0,
|
||||
timestamp: timestamp(origin, origin, UNIX_EPOCH + Duration::from_secs(1)),
|
||||
event: header_event(),
|
||||
};
|
||||
|
||||
let json = serde_json::to_value(&record).unwrap();
|
||||
|
||||
assert_eq!(json["seq"], 0);
|
||||
assert_eq!(json["type"], "header");
|
||||
assert_eq!(json["format"], FORMAT_NAME);
|
||||
assert_eq!(json["version"], FORMAT_VERSION);
|
||||
assert_eq!(json["timestamp"]["mono_ns"], 0);
|
||||
assert_eq!(serde_json::from_value::<Record>(json).unwrap(), record);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn converts_existing_instants_relative_to_origin() {
|
||||
let origin = Instant::now();
|
||||
|
||||
assert_eq!(
|
||||
monotonic_ns(origin, origin + Duration::from_nanos(123_456)),
|
||||
123_456
|
||||
);
|
||||
assert_eq!(monotonic_ns(origin, origin - Duration::from_nanos(1)), 0);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
|
||||
//
|
||||
// SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
use super::v1::{Event, Record};
|
||||
use flume::Receiver;
|
||||
use log::warn;
|
||||
use std::io::{self, Write};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::thread::{self, JoinHandle};
|
||||
|
||||
pub(super) fn spawn<W>(
|
||||
writer: W,
|
||||
records: Receiver<Record>,
|
||||
header: Record,
|
||||
healthy: Arc<AtomicBool>,
|
||||
) -> io::Result<JoinHandle<()>>
|
||||
where
|
||||
W: Write + Send + 'static,
|
||||
{
|
||||
thread::Builder::new()
|
||||
.name("trace-v1-writer".to_owned())
|
||||
.spawn(move || {
|
||||
if let Err(error) = write_records(writer, records, header) {
|
||||
warn!("Legacy control trace writer failed; tracing disabled: {error}");
|
||||
healthy.store(false, Ordering::Release);
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn write_records<W>(mut writer: W, records: Receiver<Record>, header: Record) -> io::Result<()>
|
||||
where
|
||||
W: Write,
|
||||
{
|
||||
write_record(&mut writer, &header)?;
|
||||
|
||||
while let Ok(record) = records.recv() {
|
||||
let clean_end = matches!(record.event, Event::End { clean: true });
|
||||
write_record(&mut writer, &record)?;
|
||||
|
||||
if clean_end {
|
||||
writer.flush()?;
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
writer.flush()
|
||||
}
|
||||
|
||||
fn write_record(writer: &mut impl Write, record: &Record) -> io::Result<()> {
|
||||
serde_json::to_writer(&mut *writer, record).map_err(io::Error::other)?;
|
||||
writer.write_all(b"\n")
|
||||
}
|
||||
Reference in new issue
Block a user