~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

knotless-run keeps a floor of free disk, shedding caches before a run

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:run/disk-floor
crates/knotless-run/src/caches.rs+102−0
View file
@@ -40,6 +40,11 @@
//! newer, so the base never goes back to an older tip. Each base made so
//! is marked with that tip's generation, in [`MARKS`].
//!
//! The disk the caches are on is shared, with the site on the same host
//! among others, so the runner keeps a floor of it free: below it, before
//! a run starts, [`Caches::shed`] removes every kept run's layers and
//! every base no run reads, and the runs after start afresh.
//!
//! Under the root: `bases/<cache>/<generation>`, `runs/<run>/<n>/` with
//! each layer's `upper` and the `work` directory overlayfs needs beside
//! it, `kept/<run>/`, a kept run's [`KEPT`] and its `<n>/upper`s, and
@@ -77,8 +82,13 @@ pub const KEPT_AGE: u64 = 3 * 24 * 60 * 60;
/// The most runs whose layers are kept at once.
pub const KEPT_MAX: usize = 16;
/// Says the bytes free on the disk holding a path.
pub type Space = Box<dyn Fn(&Path) -> io::Result<u64> + Send + Sync>;
pub struct Caches {
root: PathBuf,
/// The bytes free on the disk the caches are on: [`free`], but in tests.
space: Space,
/// This runner's start, so its runs' names are not an earlier one's.
boot: u128,
runs: AtomicU64,
@@ -187,6 +197,16 @@ impl Caches {
/// layers kept of the runs `kept` names, whose checks a push may yet
/// take, and of no other.
pub fn open(root: &Path, alive: &[String], kept: &[String]) -> Result<Arc<Caches>> {
Caches::open_with(root, alive, kept, Box::new(free))
}
/// [`Caches::open`], told the bytes free on its disk by `space`.
pub fn open_with(
root: &Path,
alive: &[String],
kept: &[String],
space: Space,
) -> Result<Arc<Caches>> {
let (bases_dir, runs_dir) = (root.join("bases"), root.join("runs"));
std::fs::create_dir_all(&bases_dir).context("making the caches' directory")?;
std::fs::create_dir_all(&runs_dir)?;
@@ -250,6 +270,7 @@ impl Caches {
.as_millis();
Ok(Arc::new(Caches {
root: root.to_path_buf(),
space,
boot,
runs: AtomicU64::new(0),
bases: Mutex::new(bases),
@@ -278,6 +299,42 @@ impl Caches {
Ok(())
}
/// The bytes free on the disk the caches are on.
pub fn free(&self) -> io::Result<u64> {
(self.space)(&self.root)
}
/// Makes room on the disk, as it runs short: removes the layers kept
/// of every run, which a push would only have saved a run's work
/// with, and every cache's base no run reads, as [`Caches::drop_repo`]
/// drops them, for the next run of each to start afresh. Says how
/// many bases and kept runs went.
pub fn shed(&self) -> Result<(usize, usize)> {
let mut runs = 0;
{
let _kept = self.kept.lock().unwrap();
for run in std::fs::read_dir(self.root.join("kept"))? {
remove(&run?.path())?;
runs += 1;
}
}
let mut bases = self.bases.lock().unwrap();
let mut gone = 0;
let current: Vec<(String, u64)> = bases.current.clone().into_iter().collect();
for (name, generation) in current {
if bases.readers.contains_key(&(name.clone(), generation)) {
continue;
}
bases.current.remove(&name);
bases.marks.remove(&name);
for generation in bases.unused(&name) {
remove(&self.base(&name, generation))?;
gone += 1;
}
}
Ok((gone, runs))
}
/// Where the layers kept of the run `run` are.
fn kept_dir(&self, run: &str) -> PathBuf {
self.root.join("kept").join(keys::ref_hash(run))
@@ -600,6 +657,12 @@ impl Drop for Lease {
}
}
/// The bytes free to the runner on the disk holding `path`.
pub fn free(path: &Path) -> io::Result<u64> {
let stat = rustix::fs::statvfs(path)?;
Ok(stat.f_bavail.saturating_mul(stat.f_frsize))
}
/// The seconds since the epoch.
fn now() -> u64 {
let since = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
@@ -1122,4 +1185,43 @@ mod tests {
caches.forgetting(&[], 1 << 30, late).unwrap();
assert_eq!(kept_of(root.path()), [keys::ref_hash("wanted")]);
}
#[tokio::test]
async fn short_of_disk_it_sheds_what_no_run_reads() {
let root = tempfile::tempdir().unwrap();
// The disk as statvfs tells it, of somewhere real.
assert!(free(root.path()).unwrap() > 0);
assert!(free(&root.path().join("gone")).is_err());
let space: Space = Box::new(|_| Ok(7));
let caches = Caches::open_with(root.path(), &[], &[], space).unwrap();
assert_eq!(caches.free().unwrap(), 7);
let wanted = names(&[("read", "/a"), ("unread", "/b")]);
// Bases for both caches, and a pull's run kept.
let main = caches.lease(&wanted).unwrap();
write(&main.layers[0].upper.join("x"), "x");
write(&main.layers[1].upper.join("y"), "y");
main.promote(1 << 30).await.unwrap();
let pull = caches.lease(&wanted).unwrap();
write(&pull.layers[0].upper.join("z"), "z");
pull.keep("pull-run", 1 << 30).await.unwrap();
assert!(caches.kept_dir("pull-run").exists());
// A run reads one cache's base as the disk runs short.
let reading = caches.lease(&names(&[("read", "/a")])).unwrap();
let (read, unread) = {
let all = caches.lease(&wanted).unwrap();
(all.layers[0].lower.clone(), all.layers[1].lower.clone())
};
assert_eq!(caches.shed().unwrap(), (1, 1));
assert!(!caches.kept_dir("pull-run").exists());
assert!(!unread.exists());
assert!(read.exists());
// The next run of the one shed starts afresh; the one read goes
// on from its base.
let next = caches.lease(&wanted).unwrap();
assert_eq!(next.layers[0].lower, read);
assert!(listing(&next.layers[1].lower).is_empty());
drop((next, reading));
// Nothing left to shed but the bases no run reads now.
assert_eq!(caches.shed().unwrap(), (2, 0));
}
}
crates/knotless-run/src/lib.rs+123−2
View file
@@ -116,7 +116,7 @@ use tokio::sync::{Notify, Semaphore};
use turns::{Class, Rank, Turns, Whose};
use workflow::{Superseded, Workflow};
const USAGE: &str = "usage: knotless-run [--jetstream <wss://host>] [--executor docker|shell] [--image <image>] [--secrets <file>] [--secrets-key <file>] [--state <file>] [--work <dir>] [--step-timeout <secs>] [--run-timeout <secs>] [--cpus <n>] [--memory <GiB>] [--pids <n>] [--network <name>] [--runtime <name>] [--trusted-runtime <name>] [--trust <did>] [--resolv <file>] [--jobs <n>] [--pull-jobs <n>] [--owner-jobs <n>] [--allowance <minutes>] [--private-allowance <minutes>] [--cache-max <GiB>] [--caches <dir>] [--paused <file>] [--keep-checks <days>] [--sweep <secs>] [--stop-wait <secs>] [--poll <secs>] [--pulls] [--pulls-only <handle>/<name>] [--unmetered <did>] [--queue [--batch <n>] [--depth 1|2] [--notice <secs>]] [--discover] [<handle>/<name>...]";
const USAGE: &str = "usage: knotless-run [--jetstream <wss://host>] [--executor docker|shell] [--image <image>] [--secrets <file>] [--secrets-key <file>] [--state <file>] [--work <dir>] [--step-timeout <secs>] [--run-timeout <secs>] [--cpus <n>] [--memory <GiB>] [--pids <n>] [--network <name>] [--runtime <name>] [--trusted-runtime <name>] [--trust <did>] [--resolv <file>] [--jobs <n>] [--pull-jobs <n>] [--owner-jobs <n>] [--allowance <minutes>] [--private-allowance <minutes>] [--cache-max <GiB>] [--disk-floor <GiB>] [--caches <dir>] [--paused <file>] [--keep-checks <days>] [--sweep <secs>] [--stop-wait <secs>] [--poll <secs>] [--pulls] [--pulls-only <handle>/<name>] [--unmetered <did>] [--queue [--batch <n>] [--depth 1|2] [--notice <secs>]] [--discover] [<handle>/<name>...]";
#[derive(Debug, Clone, PartialEq)]
pub struct Options {
@@ -179,6 +179,11 @@ pub struct Options {
/// The bytes a cache's base may grow to before it is dropped, to
/// start afresh at the next run.
pub cache_max: u64,
/// The bytes kept free on the disk the caches are on: below them, a
/// run sheds the caches no run reads before it starts, and waits for
/// room if that is not enough, so that what else the disk holds, as
/// the site on the same host, never runs out. 0 keeps none.
pub disk_floor: u64,
/// Where the caches' bases, and each run's layers over them, are kept:
/// a path the docker daemon knows by the same name. Defaults to one in
/// the cache directory.
@@ -246,6 +251,7 @@ impl Options {
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 16 << 30,
disk_floor: 20 << 30,
caches: None,
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(15 * 60),
@@ -401,6 +407,12 @@ pub fn parse_args(args: &[String]) -> Result<Options, String> {
.map_err(|_| "--cache-max takes GiB".to_string())?;
opts.cache_max = gib << 30;
}
"--disk-floor" => {
let gib: u64 = value()?
.parse()
.map_err(|_| "--disk-floor takes GiB".to_string())?;
opts.disk_floor = gib << 30;
}
"--jobs" => opts.jobs = count(flag, "runs", value()?)?,
"--pull-jobs" => opts.pull_jobs = count(flag, "runs", value()?)?,
"--owner-jobs" => opts.owner_jobs = count(flag, "runs", value()?)?,
@@ -617,6 +629,58 @@ fn waited(past: Option<u64>, took: Duration) -> String {
}
}
/// How often a run waiting for room on the disk looks again.
const ROOM_EVERY: Duration = Duration::from_secs(30);
/// Waits until the disk the caches are on has `floor` bytes free, looking
/// each `every`: below it, the caches no run reads are shed first, once,
/// and then the runs going are waited for, as they end and free theirs.
/// Says what the run's log says of it: nothing when there was room.
async fn room(caches: &Arc<caches::Caches>, floor: u64, every: Duration) -> String {
let (began, mut said, mut shed) = (tokio::time::Instant::now(), String::new(), false);
loop {
let free = match caches.free() {
Ok(free) => free,
Err(err) => {
telemetry::failed("reading the disk's free space", &err.into());
return said;
}
};
if free >= floor {
if began.elapsed() >= every {
let took = began.elapsed().as_secs();
said += &format!("==> this one waited {took}s for room on the disk\n");
}
return said;
}
let gib = |bytes: u64| bytes as f64 / (1u64 << 30) as f64;
let short = format!(
"{:.1} GiB free on the disk, under the runner's floor of {:.1} GiB",
gib(free),
gib(floor)
);
if shed {
telemetry::short_of_disk(&short, false);
tokio::time::sleep(every).await;
continue;
}
shed = true;
telemetry::short_of_disk(&short, true);
let caches = caches.clone();
let shedding = tokio::task::spawn_blocking(move || caches.shed()).await;
let (bases, kept) = match shedding.map_err(anyhow::Error::from).and_then(|r| r) {
Ok(gone) => gone,
Err(err) => {
telemetry::failed("shedding caches", &err);
(0, 0)
}
};
said += &format!(
"==> {short}: shed {bases} caches' bases no run read and {kept} kept runs' layers, so this one starts afresh\n"
);
}
}
/// Today, as days since the epoch.
fn today() -> u64 {
let since = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
@@ -2246,10 +2310,14 @@ impl Runner {
let began = std::time::Instant::now();
let turn = until_cancelled(cancelled, self.turns.clone().take(rank, whose, Vec::new()));
let turn = turn.await?;
let mut said = waited(over, began.elapsed());
// Holding its job, so that no other run starts meanwhile.
let room = room(&self.caches, self.opts.disk_floor, ROOM_EVERY);
said += &until_cancelled(cancelled, room).await?;
Some(Held {
on_ref: kept,
job: turn,
said: waited(over, began.elapsed()),
said,
})
}
@@ -3768,6 +3836,53 @@ mod tests {
// On paused time, so the 5ms tick always comes before the 30ms pass:
// on a busy host both could be due at once, and select! take either.
#[tokio::test(start_paused = true)]
async fn a_run_waits_for_room_on_the_disk() {
use std::sync::atomic::AtomicI64;
let root = tempfile::tempdir().unwrap();
// The bytes free, as the test sets them, under 0 unreadable; and
// how many looks find 5 first, as runs going hold the rest.
let (free, short) = (Arc::new(AtomicI64::new(10)), Arc::new(AtomicI64::new(0)));
let (told, shorted) = (free.clone(), short.clone());
let space: caches::Space = Box::new(move |_| {
if shorted.fetch_sub(1, Ordering::SeqCst) > 0 {
return Ok(5);
}
match told.load(Ordering::SeqCst) {
n if n < 0 => Err(std::io::Error::other("no disk")),
n => Ok(n as u64),
}
});
let caches = caches::Caches::open_with(root.path(), &[], &[], space).unwrap();
let every = Duration::from_secs(30);
// Room enough, or no floor: nothing to say.
assert_eq!(room(&caches, 10, every).await, "");
assert_eq!(room(&caches, 0, every).await, "");
// A disk it cannot read holds no run back.
free.store(-1, Ordering::SeqCst);
assert_eq!(room(&caches, 10, every).await, "");
free.store(20, Ordering::SeqCst);
// Short, it sheds the caches, and room made so, says so alone.
short.store(1, Ordering::SeqCst);
let said = room(&caches, 10, every).await;
assert!(said.contains("under the runner's floor"), "{said}");
assert!(said.contains("shed 0 caches' bases"), "{said}");
assert!(!said.contains("waited"), "{said}");
// Short after shedding, it waits for the runs going to free some:
// a look before the shed, one after, and one each 30s.
short.store(4, Ordering::SeqCst);
let said = room(&caches, 10, every).await;
assert!(
said.ends_with("waited 90s for room on the disk\n"),
"{said}"
);
// Caches it cannot shed are logged, and waited past.
std::fs::remove_dir_all(root.path().join("kept")).unwrap();
short.store(2, Ordering::SeqCst);
let said = room(&caches, 10, every).await;
assert!(said.contains("shed 0 caches' bases"), "{said}");
}
#[tokio::test(start_paused = true)]
async fn a_long_pass_is_watched() {
let stage = Stage::default();
@@ -4088,6 +4203,12 @@ mod tests {
args("--cache-max x a/b").unwrap_err(),
"--cache-max takes GiB"
);
assert_eq!(Options::new(Vec::new()).disk_floor, 20 << 30);
assert_eq!(args("--disk-floor 0 a/b").unwrap().disk_floor, 0);
assert_eq!(
args("--disk-floor x a/b").unwrap_err(),
"--disk-floor takes GiB"
);
assert_eq!(args("--jobs 2 a/b").unwrap().jobs, 2);
// Pulls take half the jobs, or as many as asked, up to all of them.
assert_eq!(opts.pull_share(), 2);
crates/knotless-run/src/telemetry.rs+17−0
View file
@@ -148,6 +148,21 @@ pub(crate) fn long_pass(label: &str, stage: &str, took: Duration) {
telemetry::note(Level::WARN, "pass running long", &fields);
}
/// Logs that the disk is `short` of the runner's floor, as a run sheds
/// caches, or as it waits for room once it has.
pub(crate) fn short_of_disk(short: &str, shedding: bool) {
let fields = Fields {
reason: Some(short),
..Fields::default()
};
let message = if shedding {
"disk short: shedding caches"
} else {
"disk short: waiting for room"
};
telemetry::note(Level::WARN, message, &fields);
}
/// Logs `message` about a batch of `pulls` pulls into `branch` of
/// `label`, at `tip`, and why, if said.
pub(crate) fn batch(
@@ -287,6 +302,8 @@ mod tests {
watching(&["a/b".into()], "did:plc:x");
found("did:plc:x/b");
left("did:plc:x/b");
short_of_disk("1.0 GiB free", true);
short_of_disk("1.0 GiB free", false);
}
#[test]
crates/knotless-run/tests/run.rs+7−0
View file
@@ -167,6 +167,7 @@ fn runner_opts(runner: &Sandbox, owner: &Sandbox, js: &Jetstream) -> knotless_ru
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
@@ -782,6 +783,7 @@ fn what_goes_wrong_is_reported() {
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: None,
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
@@ -988,6 +990,7 @@ fn a_deploy_is_never_cut_off() {
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
@@ -1129,6 +1132,7 @@ fn a_stop_drops_what_has_not_started() {
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
@@ -1398,6 +1402,7 @@ fn a_failing_pds_wedges_nothing() {
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(1),
stop_wait: Duration::from_secs(2),
@@ -2135,6 +2140,7 @@ fn a_merged_pull_passes_its_push() {
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
@@ -3675,6 +3681,7 @@ fn a_workflow_asked_for_by_name() {
unmetered: Vec::new(),
trust: Vec::new(),
cache_max: 1 << 30,
disk_floor: 0,
caches: Some(runner.path("caches")),
sweep: Duration::from_secs(600),
stop_wait: Duration::from_secs(60),
deploy/README.md+4−1
View file
@@ -454,7 +454,10 @@ three times in all. It reads every repository again every `--sweep`
left. A base over `--cache-max` GiB (16) once made is dropped, for the
next run to start afresh. Disk, then: each base once, each run going's
own writes on top (a test run's build, a few GB), and up to
`--cache-max` GiB of kept layers. The host needs overlayfs over ext4
`--cache-max` GiB of kept layers. Below `--disk-floor` GiB free (20),
a run sheds the bases no run reads and the kept layers before it starts,
and waits for room if that is not enough, so the site on the same disk
keeps room to write. The host needs overlayfs over ext4
with user xattrs, as Linux 5.11 and up has. The docker daemon mounts
the overlays by the host's paths, one more reason `KNOTLESS_RUN_DATA` is
the same path inside and out. The first start after the switch from
docs/design/runner.md+22−0
View file
@@ -686,6 +686,28 @@ warms the rest.
On start, the runner removes its leftover overlays and half-made bases,
and sweeps no old base while an old overlay is still mounted.
### Room on the disk
The caches share their disk with whatever else the host runs. On
knotless.dev's own host that includes the site's repositories and index.
On 2026-10-31 a full disk took the site down. So the runner keeps
`--disk-floor` free on the disk `--caches` is on: 20 GiB by default. That
is about two of the largest bases the host has held, so one run's build
can land on the floor without the site running out.
Before a run starts, holding its job so no other run starts meanwhile, the
runner checks the free space. When it is below the floor, the runner sheds
first: every kept run's layers, and every cache's base no run reads. Those
caches' next runs start afresh, as after a dropped base. If that is not
enough, the run waits, checking again every 30 seconds, until the runs
going end and free their layers. It stays queued, and can be cancelled as
usual. Its log says what was shed and how long it waited. The runner logs
`disk short` at warn each time it looks and finds the disk short. A disk
it cannot read holds nothing back.
The floor holds only at a run's start. A run already going can still fill
the disk.
### Where pushes run elsewhere
A repository named with `--pulls-only` has its pushes run by another
docs/reference/cli.md+1−0
View file
@@ -421,6 +421,7 @@ add` make it one, and invite it, which `--discover` reads.
| `--stop-wait <secs>` | 900 | how long a stop waits for runs going before it cancels them; 0 waits not at all |
| `--poll <secs>` | 10 | how often a private repository's space is asked whether anything was written there, which Jetstream never carries; while `KNOTLESS_APPVIEW`'s appview passes on its writes, no more often than once a minute (see [Private repositories](../design/runner.md#private-repositories)) |
| `--cache-max <GiB>` | 16 | a whole number of GiB: the size at which a cache's new base is not used, to start afresh, and the most kept layers may take in all |
| `--disk-floor <GiB>` | 20 | a whole number of GiB kept free on the disk `--caches` is on: below it, a run about to start first sheds every cache's base no run reads and every kept run's layers, then waits for room, looking every 30s; 0 keeps no floor (see [Room on the disk](../design/runner.md#room-on-the-disk)) |
| `--caches <dir>` | `<cache dir>/run/caches` | where caches' bases and runs' layers are kept; a path the docker daemon sees under the same name |
| `--secrets <file>` | | the [secrets file](workflows.md#secrets-file) |
| `--secrets-key <file>` | `secrets-key.json` beside the state file | where the runner keeps the key owners seal secrets to, made if missing, and published as its `dev.knotless.runnerKey` record (see [The runner's key](../design/runner.md#the-runners-key)) |
docs/reference/runner-internals.md+14−0
View file
@@ -59,6 +59,20 @@ kept directories not wanted or broken, are removed, and the newest
generation of each cache is the current one. A base named `knotless-cache-<hash>`, as runners named them before a cache's name held its repository's, is removed. Marks of a base no longer
current are let go.
## Free space
The free space is `statvfs(2)` of `--caches`: the blocks available to an
unprivileged writer (`f_bavail`) times the fragment size (`f_frsize`).
Below `--disk-floor`, a run about to start removes every `kept/*`
directory and, for each cache whose current base no run reads, that base
and its older generations no run reads, forgetting the cache's mark. It
does this once for each run that finds the disk short. The run's log
starts `==> <n> GiB free on the disk, under the runner's floor of <n>
GiB: shed <n> caches' bases no run read and <n> kept runs' layers`, then
says `==> this one waited <n>s for room on the disk` if it waited 30s or
more. The runner logs `disk short: shedding caches` and `disk short:
waiting for room` at warn, with the free space in `reason`.
## What a pass logs
Each pass over a repository, as its door rings, logs `pass ended` with