gc drops a retired generation once nothing older than its retirement can read it
Signed-off-by: dawn <90008@klbr.net>
63 unmodified lines646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146 unmodified linesExpand all29229329429522229629729829930030130230330430530622630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833937 unmodified lines377378379380381382383384385233 unmodified linesExpand all63 unmodified lines let _ = std::fs::remove_file(&path); false}
/// when the oldest process that may read a retired generation started, since one started after the/// retirement can't read it. readers by role: workers of the nginx master in the pid file, cgi/// requests fcgiwrap runs, `git-ssh` sessions, and what those start. long-lived services such as/// the firehose read no generation and don't count. unknown off linux or without nginx's pid file.fn oldest_reader(output_root: &Path) -> Option<u64> { // SAFETY: sysconf only reads a constant of the running system let ticks = unsafe { libc::sysconf(libc::_SC_CLK_TCK) }; oldest_reader_in(Path::new("/proc"), output_root, u64::try_from(ticks).ok()?)}
struct Process { parent: u32, name: String, git_ssh: bool, started: u64,}
fn oldest_reader_in(proc: &Path, output_root: &Path, ticks: u64) -> Option<u64> { let master: u32 = std::fs::read_to_string(output_root.join("runtime/nginx.pid")) .ok()? .trim() .parse() .ok()?; let boot: u64 = std::fs::read_to_string(proc.join("stat")) .ok()? .lines() .find_map(|line| line.strip_prefix("btime ")?.trim().parse().ok())?; let mut processes = BTreeMap::new(); for entry in std::fs::read_dir(proc).ok()? { let entry = entry.ok()?; let Some(pid) = (entry.file_name().to_str()).and_then(|name| name.parse::<u32>().ok()) else { continue; }; // a process that exited since the listing reads nothing let Ok(stat) = std::fs::read_to_string(entry.path().join("stat")) else { continue; }; // the name is in parentheses and may hold spaces, the fields after it don't let (open, close) = (stat.find('(')?, stat.rfind(')')?); let fields: Vec<&str> = stat.get(close + 2..)?.split(' ').collect(); let command = std::fs::read(entry.path().join("cmdline")).unwrap_or_default(); let process = Process { parent: fields.get(1)?.parse().ok()?, name: stat[open + 1..close].to_string(), git_ssh: command .split(|&byte| byte == 0) .any(|arg| arg == b"git-ssh"), started: boot + fields.get(19)?.parse::<u64>().ok()? / ticks, }; processes.insert(pid, process); } let me = std::process::id(); let reads = |mut pid: u32| { while let Some(process) = processes.get(&pid) { let parent = processes.get(&process.parent); if process.parent == master || process.git_ssh || (process.name != "fcgiwrap" && parent.is_some_and(|parent| parent.name == "fcgiwrap")) { return true; } pid = process.parent; } false }; let oldest = (processes.iter()) .filter(|&(&pid, _)| pid != me && reads(pid)) .map(|(_, process)| process.started) .min(); Some(oldest.unwrap_or(u64::MAX))}
#[derive(Deserialize, Serialize)]struct Live { #[serde(default)] repos: Vec<LiveRepo>,146 unmodified linesExpand all
fn collect_locking(output_root: &Path, keep: usize, drain_secs: u64) -> anyhow::Result<Removed> { let _output_lock = crate::output::lock(output_root)?; crate::publish::reconcile(output_root)?; collect_at(output_root, keep, crate::publish::unix_now(), drain_secs) let reader = oldest_reader(output_root); collect_with( output_root, keep, crate::publish::unix_now(), drain_secs, reader, )}
pub(crate) fn collect(output_root: &Path, keep: usize) -> anyhow::Result<Removed> { collect_at(output_root, keep, crate::publish::unix_now(), DRAIN_SECS) let reader = oldest_reader(output_root); collect_with( output_root, keep, crate::publish::unix_now(), DRAIN_SECS, reader, )}
#[cfg(test)]fn collect_at( output_root: &Path, keep: usize, now: u64, drain_secs: u64,) -> anyhow::Result<Removed> { collect_with(output_root, keep, now, drain_secs, None)}
/// a retired generation stays for the drain window, or until nothing that started before its/// retirement still runs, when `oldest_reader` says when the oldest of those started.fn collect_with( output_root: &Path, keep: usize, now: u64, drain_secs: u64, oldest_reader: Option<u64>,) -> anyhow::Result<Removed> { anyhow::ensure!( !crate::tombstone::present(output_root), "activation tombstone present, the served generation is unknown" );37 unmodified lines kept.extend( superseded .iter() .filter(|entry| now.saturating_sub(entry.at_unix) < drain_secs) .filter(|entry| oldest_reader.is_none_or(|start| start <= entry.at_unix)) .map(|entry| entry.generation.clone()), ); // build time only fills in generations no activation recorded. let mut targets: Vec<&str> = superseded233 unmodified linesExpand all436 unmodified linesExpand all437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548436 unmodified linesExpand all .collect(); cached.sort(); assert_eq!(cached, ["g3", "g4"]);}
#[test]fn a_retired_generation_goes_once_nothing_older_than_its_retirement_reads() { let root = knot(&["g1", "g2", "g3"]); let root = root.path(); activate(root, "g3", NOW, &[("g1", NOW - 10), ("g2", NOW)]); let kept = |reader| { collect_with(root, 1, NOW, DRAIN_SECS, reader).unwrap(); generations(root) }; assert_eq!( kept(None), ["g1", "g2", "g3"], "unknown readers wait out the window" ); assert_eq!( kept(Some(NOW - 20)), ["g1", "g2", "g3"], "an old worker still runs" ); assert_eq!( kept(Some(NOW - 10)), ["g1", "g2", "g3"], "one started the second g1 retired" ); assert_eq!(kept(Some(NOW - 9)), ["g2", "g3"]);}
#[test]fn the_oldest_reader_is_a_worker_or_a_request_started_first() { let root = tempfile::tempdir().unwrap(); let (proc, output) = (root.path().join("proc"), root.path().join("out")); std::fs::create_dir_all(output.join("runtime")).unwrap(); std::fs::create_dir_all(&proc).unwrap(); std::fs::write(proc.join("stat"), "cpu 1 2 3\nbtime 1000\n").unwrap(); let process = |pid: u32, name: &str, parent: u32, started_ticks: u64, args: &[&str]| { let dir = proc.join(pid.to_string()); std::fs::create_dir_all(&dir).unwrap(); let zeros = "0 ".repeat(17); let stat = format!("{pid} ({name}) S {parent} {zeros}{started_ticks} 0 0\n"); std::fs::write(dir.join("stat"), stat).unwrap(); let command: Vec<u8> = args .iter() .flat_map(|arg| [arg.as_bytes(), b"\0"].concat()) .collect(); std::fs::write(dir.join("cmdline"), command).unwrap(); }; assert_eq!(oldest_reader_in(&proc, &output, 100), None, "no nginx pid"); std::fs::write(output.join("runtime/nginx.pid"), "10\n").unwrap(); process(1, "systemd", 0, 0, &["init"]); process(10, "nginx", 1, 100, &["nginx: master process"]); process(20, "fcgiwrap", 1, 200, &["fcgiwrap"]); process(21, "fcgiwrap", 20, 200, &["fcgiwrap"]); // services up since boot read no generation process( 30, "knot-static-fir", 1, 300, &["knot-static-firehose", "/out/firehose"], ); process( 31, "knot-static", 1, 300, &["knot-static", "maintenance", "--all"], ); process(32, "git", 31, 400, &["git", "repack"]); assert_eq!( oldest_reader_in(&proc, &output, 100), Some(u64::MAX), "nothing reads" ); process(11, "nginx", 10, 5000, &["nginx: worker process"]); process(40, "knot-static", 21, 3000, &["knot-static"]); process(41, "git", 40, 9000, &["git", "upload-pack"]); assert_eq!( oldest_reader_in(&proc, &output, 100), Some(1030), "a cgi request" ); process(50, "sshd", 1, 100, &["sshd"]); process( 51, "knot-static", 50, 2500, &["knot-static", "--config", "c", "git-ssh"], ); assert_eq!( oldest_reader_in(&proc, &output, 100), Some(1025), "a git-ssh session" ); process( 12, "nginx", 10, 2000, &["nginx: worker process is shutting down"], ); assert_eq!( oldest_reader_in(&proc, &output, 100), Some(1020), "an old worker" );}