A job's finish carries its peak disk and memory 8f211e20bea5

The runner reads the blocks the job's scratch directory took at the end
(its disks are sparse files there, so that is their peak) and samples
the sandbox's peak resident memory while it runs (the VMM's, or the
syscall tier's helper's), and sends both with the finish, for the hub to
learn what the job's next run needs.

refs #41

Author: Andy McCall @andymccall <[email protected]>
When: 24 days ago · 2026-10-11 22:04 UTC
Signature: ✔ signed by Andy McCall
Hash: 8f211e20bea5e995d04f82b9aab8961aea9b729aafe87a2894cadba90cd0c722
Depends on: 280ef951d193 3007d8f011fa 62829d15ef09 a67b97e9a717 e0918b887850 ea56c56eaf90
Depended on by: 1215f05ff8d0 3554afefb6aa 337afb8f6043 077a65242359 ea81796bb757 d60aac6c3f8a
In bookmarks: main newer-pipeline push-uses-login
In tags: v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
Edits: 15 atoms, 0 tree operations, 5 files

Browse the tree at this change

hide whitespace changes

modified crates/flow-core/src/api.rs +7 −0

modified crates/flow/src/facts.rs +33 −0

modified crates/flow/src/microvm.rs +9 −0

modified crates/flow/src/runner.rs +24 −2

modified crates/flow/src/syscall.rs +7 −0

		…	274 unchanged lines
275	275		pub signed_receipt: Option<String>,
276	276		/// Why the runner could not run the job (rootfs missing, VM failed).
277	277		pub error: Option<String>,
	278	+	/// The most disk the job's disks and workspace took, in MiB, and the
	279	+	/// most memory its sandbox held (oxygenflow#41): the hub learns from
	280	+	/// them what the job's next run needs.
	281	+	#[serde(default, skip_serializing_if = "Option::is_none")]
	282	+	pub peak_disk_mib: Option<u64>,
	283	+	#[serde(default, skip_serializing_if = "Option::is_none")]
	284	+	pub peak_memory_mib: Option<u64>,
278	285		}
279	286
280	287		/// Longest text of one record the hub keeps; longer lines are cut.
		…	3 unchanged lines

		…	32 unchanged lines
33	33		Some((s.f_bavail as u64).saturating_mul(s.f_frsize as u64) >> 20)
34	34		}
35	35
	36	+	/// The disk the files under `dir` take (their blocks, not their sizes: a
	37	+	/// job's disks are sparse), in MiB; links are not followed.
	38	+	pub fn allocated_mib(dir: &std::path::Path) -> Option<u64> {
	39	+	use std::os::unix::fs::MetadataExt;
	40	+	let mut total: u64 = 0;
	41	+	let mut stack = vec![dir.to_path_buf()];
	42	+	while let Some(d) = stack.pop() {
	43	+	for e in std::fs::read_dir(&d).ok()?.flatten() {
	44	+	let m = std::fs::symlink_metadata(e.path()).ok()?;
	45	+	if m.is_dir() {
	46	+	stack.push(e.path());
	47	+	} else {
	48	+	total = total.saturating_add(m.blocks() * 512);
	49	+	}
	50	+	}
	51	+	}
	52	+	Some(total >> 20)
	53	+	}
	54	+
	55	+	/// A process's peak resident memory (`VmHWM`), in MiB.
	56	+	pub fn peak_rss_mib(pid: u32) -> Option<u64> {
	57	+	let text = std::fs::read_to_string(format!("/proc/{pid}/status")).ok()?;
	58	+	let kib: u64 = text.lines().find_map(\|l\| l.strip_prefix("VmHWM:"))?.trim().trim_end_matches("kB").trim().parse().ok()?;
	59	+	Some(kib / 1024)
	60	+	}
	61	+
36	62		/// The 1-minute load average from /proc/loadavg, times 100.
37	63		pub fn load_x100() -> Option<u32> {
38	64		let text = std::fs::read_to_string("/proc/loadavg").ok()?;
		…	36 unchanged lines
75	101		fn the_cheap_facts_are_read_and_classes_double() {
76	102		assert!(free_disk_mib(std::path::Path::new("/")).is_some());
77	103		assert!(load_x100().is_some());
	104	+	assert!(peak_rss_mib(std::process::id()).is_some_and(\|m\| m > 0));
	105	+	let dir = tempfile::tempdir().unwrap();
	106	+	let f = std::fs::File::create(dir.path().join("sparse")).unwrap();
	107	+	f.set_len(1 << 30).unwrap();
	108	+	std::fs::write(dir.path().join("full"), vec![7u8; 3 << 20]).unwrap();
	109	+	let mib = allocated_mib(dir.path()).unwrap();
	110	+	assert!((3..=4).contains(&mib), "the sparse gigabyte takes nothing, the written 3 MiB do: {mib}");
78	111		assert_eq!(Facts::class_of(100), 3, "the reference machine");
79	112		assert_eq!((Facts::class_of(50), Facts::class_of(200), Facts::class_of(500), Facts::class_of(10)), (2, 4, 5, 1));
80	113		}
		…	1 unchanged lines

		…	19 unchanged lines
20	20		pub timed_out: bool,
21	21		/// Stopped because the run was cancelled (oxygenflow#32).
22	22		pub cancelled: bool,
	23	+	/// The most memory the sandbox held, in MiB (oxygenflow#41).
	24	+	pub peak_memory_mib: Option<u64>,
23	25		}
24	26
25	27		/// Where a job's records go, and whether it is still wanted: asked a few
		…	122 unchanged lines
148	150		let mut cancelled = false;
149	151		let deadline = started + timeout;
150	152		let t = \|s: &Instant\| s.elapsed().as_millis() as u64;
	153	+	// The VMM's peak resident memory is the guest memory it touched: read
	154	+	// while it runs, since it is gone once it exits.
	155	+	let mut peak_memory_mib = None;
151	156		loop {
	157	+	if let Some(m) = crate::facts::peak_rss_mib(child.id()) {
	158	+	peak_memory_mib = Some(m);
	159	+	}
152	160		match rx.recv_timeout(Duration::from_millis(200)) {
153	161		Ok((from_console, line)) => {
154	162		let text = String::from_utf8_lossy(&line).trim_end_matches('\r').to_string();
		…	49 unchanged lines
204	212		environment,
205	213		timed_out,
206	214		cancelled,
	215	+	peak_memory_mib,
207	216		})
208	217		}
209	218
		…	62 unchanged lines

		…	38 unchanged lines
39	39		const CANCELLED: &str = "the run was cancelled";
40	40
41	41		fn cancelled() -> Finish {
42		-	Finish { exit_code: None, duration_ms: 0, receipt: None, signed_receipt: None, error: Some(CANCELLED.into()) }
	42	+	Finish {
	43	+	exit_code: None,
	44	+	duration_ms: 0,
	45	+	receipt: None,
	46	+	signed_receipt: None,
	47	+	error: Some(CANCELLED.into()),
	48	+	peak_disk_mib: None,
	49	+	peak_memory_mib: None,
	50	+	}
43	51		}
44	52
45	53		pub struct Hub {
		…	467 unchanged lines
513	521		receipt: None,
514	522		signed_receipt: None,
515	523		error: Some(format!("{e:#}")),
	524	+	peak_disk_mib: None,
	525	+	peak_memory_mib: None,
516	526		});
517	527		eprintln!(
518	528		"job {id}: exit {:?}{}",
		…	219 unchanged lines
738	748		receipt,
739	749		signed_receipt,
740	750		error: output_error.or_else(\|\| outcome.timed_out.then(\|\| "stopped at the time limit".to_string())),
	751	+	// What the job took (oxygenflow#41): its disks are sparse files
	752	+	// in the scratch directory, so their blocks now are their peak.
	753	+	peak_disk_mib: crate::facts::allocated_mib(&scratch),
	754	+	peak_memory_mib: outcome.peak_memory_mib,
741	755		})
742	756		})();
743	757		let _ = workspace::remove_tree(&scratch);
		…	267 unchanged lines
1011	1025		/// through server errors, and a refusal ends the trying at once.
1012	1026		#[test]
1013	1027		fn a_result_is_reported_until_the_hub_takes_it() {
1014		-	let f = Finish { exit_code: Some(0), duration_ms: 1, receipt: None, signed_receipt: None, error: None };
	1028	+	let f = Finish {
	1029	+	exit_code: Some(0),
	1030	+	duration_ms: 1,
	1031	+	receipt: None,
	1032	+	signed_receipt: None,
	1033	+	error: None,
	1034	+	peak_disk_mib: None,
	1035	+	peak_memory_mib: None,
	1036	+	};
1015	1037		let quick = Duration::from_millis(10);
1016	1038		let (base, seen) = fake_hub(vec![502, 503, 500, 500, 204]);
1017	1039		hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).unwrap();
		…	63 unchanged lines

		…	220 unchanged lines
221	221		let _ = child.kill();
222	222		group.kill();
223	223		};
	224	+	// The sandbox helper's peak resident memory (oxygenflow#41): the job's
	225	+	// own processes are its children, so this is a floor, not the whole.
	226	+	let mut peak_memory_mib = None;
224	227		loop {
	228	+	if let Some(m) = crate::facts::peak_rss_mib(child.id()) {
	229	+	peak_memory_mib = Some(m);
	230	+	}
225	231		match rx.recv_timeout(Duration::from_millis(200)) {
226	232		Ok((stdout, line)) => {
227	233		let text = String::from_utf8_lossy(&line).into_owned();
		…	51 unchanged lines
279	285		environment,
280	286		timed_out,
281	287		cancelled,
	288	+	peak_memory_mib,
282	289		})
283	290		}
284	291
		…	573 unchanged lines

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

A job's finish carries its peak disk and memory 8f211e20bea5

Keyboard