use std::time::Duration;
pub const SETTLE_PAST_TIMERS: Duration = Duration::from_secs(40);
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct Reading {
pub tasks: usize,
pub descriptors: usize,
pub outstanding: usize,
pub resident_kb: usize,
}
#[derive(Debug, Clone, Copy)]
pub struct Tolerance {
pub tasks: usize,
pub descriptors: usize,
pub outstanding: usize,
pub resident_kb: usize,
}
impl Default for Tolerance {
fn default() -> Self {
Self {
tasks: 4,
descriptors: 8,
outstanding: 0,
resident_kb: 16 * 1024,
}
}
}
#[derive(Debug, Clone)]
pub struct Soak {
pub before: Reading,
pub after: Reading,
pub settled_for: Duration,
}
impl Soak {
#[must_use]
pub fn leaks(&self, tolerance: Tolerance) -> Vec<String> {
let mut found = Vec::new();
for (what, before, after, allowed) in [
(
"tasks",
self.before.tasks,
self.after.tasks,
tolerance.tasks,
),
(
"descriptors",
self.before.descriptors,
self.after.descriptors,
tolerance.descriptors,
),
(
"outstanding",
self.before.outstanding,
self.after.outstanding,
tolerance.outstanding,
),
(
"resident_kb",
self.before.resident_kb,
self.after.resident_kb,
tolerance.resident_kb,
),
] {
let grew = after.saturating_sub(before);
if grew > allowed {
found.push(format!(
"{what} grew from {before} to {after} (+{grew}, tolerance {allowed})"
));
}
}
found
}
#[must_use]
pub fn is_flat(&self, tolerance: Tolerance) -> bool {
self.leaks(tolerance).is_empty()
}
#[must_use]
pub fn report(&self, tolerance: Tolerance) -> String {
let leaks = self.leaks(tolerance);
if leaks.is_empty() {
return format!(
"flat after {:.0}s: tasks {}→{}, descriptors {}→{}, outstanding {}→{}, \
resident {} kB→{} kB",
self.settled_for.as_secs_f64(),
self.before.tasks,
self.after.tasks,
self.before.descriptors,
self.after.descriptors,
self.before.outstanding,
self.after.outstanding,
self.before.resident_kb,
self.after.resident_kb
);
}
format!("leaked:\n {}", leaks.join("\n "))
}
}
#[must_use]
pub fn open_descriptors() -> usize {
std::fs::read_dir("/proc/self/fd").map_or(0, std::iter::Iterator::count)
}
#[must_use]
pub fn resident_kb() -> usize {
let Ok(statm) = std::fs::read_to_string("/proc/self/statm") else {
return 0;
};
let Some(pages) = statm.split_whitespace().nth(1) else {
return 0;
};
let Ok(pages) = pages.parse::<usize>() else {
return 0;
};
pages.saturating_mul(4)
}
#[must_use]
pub fn alive_tasks() -> usize {
tokio::runtime::Handle::try_current().map_or(0, |handle| handle.metrics().num_alive_tasks())
}
#[must_use]
pub fn sample(outstanding: usize) -> Reading {
Reading {
tasks: alive_tasks(),
descriptors: open_descriptors(),
outstanding,
resident_kb: resident_kb(),
}
}
pub async fn soak<L, Load, O, Count>(settle: Duration, outstanding: O, load: L) -> Soak
where
L: FnOnce() -> Load,
Load: std::future::Future<Output = ()>,
O: Fn() -> Count,
Count: std::future::Future<Output = usize>,
{
let before = sample(outstanding().await);
load().await;
tokio::time::sleep(settle).await;
let after = sample(outstanding().await);
Soak {
before,
after,
settled_for: settle,
}
}
#[cfg(test)]
#[allow(
clippy::unwrap_used,
clippy::expect_used,
clippy::panic,
clippy::indexing_slicing
)]
mod tests {
use super::*;
fn reading(tasks: usize, descriptors: usize, outstanding: usize) -> Reading {
Reading {
tasks,
descriptors,
outstanding,
resident_kb: 0,
}
}
fn soak_of(before: Reading, after: Reading) -> Soak {
Soak {
before,
after,
settled_for: Duration::from_secs(1),
}
}
#[tokio::test]
async fn an_injected_leak_fails_the_soak() {
let held: std::sync::Arc<std::sync::Mutex<Vec<tokio::task::JoinHandle<()>>>> =
std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let leaking = std::sync::Arc::clone(&held);
let result = soak(
Duration::from_millis(200),
|| async { 0 },
|| async move {
for _ in 0..40 {
let handle = tokio::spawn(async {
std::future::pending::<()>().await;
});
leaking.lock().expect("not poisoned").push(handle);
}
tokio::time::sleep(Duration::from_millis(50)).await;
},
)
.await;
assert!(
!result.is_flat(Tolerance::default()),
"forty leaked tasks must fail the run: {}",
result.report(Tolerance::default())
);
assert!(
result.report(Tolerance::default()).contains("tasks grew"),
"and must say what leaked: {}",
result.report(Tolerance::default())
);
for handle in held.lock().expect("not poisoned").drain(..) {
handle.abort();
}
}
#[test]
fn the_settling_floor_outlasts_a_sip_transaction() {
assert!(
SETTLE_PAST_TIMERS > Duration::from_secs(32),
"Timer J is 32 s; anything shorter counts a completed transaction as a leak"
);
}
#[tokio::test]
async fn a_clean_run_is_flat() {
let result = soak(
Duration::from_millis(200),
|| async { 0 },
|| async {
for _ in 0..40 {
tokio::spawn(async {
tokio::time::sleep(Duration::from_millis(10)).await;
});
}
tokio::time::sleep(Duration::from_millis(100)).await;
},
)
.await;
assert!(
result.is_flat(Tolerance::default()),
"{}",
result.report(Tolerance::default())
);
}
#[test]
fn growth_within_a_ceiling_is_still_a_leak() {
let run = soak_of(reading(10, 20, 0), reading(10, 20, 40));
assert!(!run.is_flat(Tolerance::default()));
assert!(
run.report(Tolerance::default())
.contains("outstanding grew")
);
}
#[test]
fn every_dimension_that_grew_is_named() {
let run = soak_of(reading(10, 20, 0), reading(100, 200, 40));
let leaks = run.leaks(Tolerance::default());
assert_eq!(leaks.len(), 3, "{leaks:?}");
assert!(leaks.iter().any(|l| l.starts_with("tasks")));
assert!(leaks.iter().any(|l| l.starts_with("descriptors")));
assert!(leaks.iter().any(|l| l.starts_with("outstanding")));
}
#[test]
fn ordinary_runtime_drift_is_not_a_leak() {
let run = soak_of(reading(10, 20, 0), reading(12, 24, 0));
assert!(
run.is_flat(Tolerance::default()),
"{}",
run.report(Tolerance::default())
);
}
#[test]
fn a_reading_that_fell_is_not_growth() {
let run = soak_of(reading(100, 200, 40), reading(10, 20, 0));
assert!(run.is_flat(Tolerance::default()));
}
#[test]
fn one_leftover_transaction_is_one_too_many() {
let run = soak_of(reading(10, 20, 0), reading(10, 20, 1));
assert!(
!run.is_flat(Tolerance::default()),
"a single leftover transaction is a leak: {}",
run.report(Tolerance::default())
);
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod memory_tests {
use super::*;
#[test]
fn memory_growth_is_a_leak_the_other_dimensions_would_miss() {
let run = Soak {
before: Reading {
tasks: 10,
descriptors: 20,
outstanding: 0,
resident_kb: 50_000,
},
after: Reading {
tasks: 10,
descriptors: 20,
outstanding: 0,
resident_kb: 150_000,
},
settled_for: Duration::from_secs(1),
};
assert!(!run.is_flat(Tolerance::default()));
assert!(
run.leaks(Tolerance::default())
.iter()
.any(|leak| leak.starts_with("resident_kb")),
"{:?}",
run.leaks(Tolerance::default())
);
}
#[test]
fn a_few_megabytes_of_allocator_drift_is_not_a_leak() {
let run = Soak {
before: Reading {
tasks: 10,
descriptors: 20,
outstanding: 0,
resident_kb: 50_000,
},
after: Reading {
tasks: 10,
descriptors: 20,
outstanding: 0,
resident_kb: 54_000,
},
settled_for: Duration::from_secs(1),
};
assert!(
run.is_flat(Tolerance::default()),
"{}",
run.report(Tolerance::default())
);
}
#[test]
fn resident_memory_is_readable_here() {
if std::path::Path::new("/proc/self/statm").exists() {
assert!(resident_kb() > 0, "a running process has a resident set");
}
}
}