From 60a0dad94fb060b129d43b149961fd4504b1c2d3 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Wed, 28 Oct 2026 18:28:06 +0800 Subject: [PATCH] perf(delivery): balance verified restore byte groups --- .../tests/test_delivery_validation_work.py | 4 +- judge/crates/sandbox/src/restore_copy/main.rs | 123 ++++++++++++++++-- scripts/release/README.md | 6 +- scripts/release/policy.py | 4 +- 4 files changed, 121 insertions(+), 16 deletions(-) diff --git a/backend/tests/test_delivery_validation_work.py b/backend/tests/test_delivery_validation_work.py index 708a34a..ef13c54 100644 --- a/backend/tests/test_delivery_validation_work.py +++ b/backend/tests/test_delivery_validation_work.py @@ -87,9 +87,9 @@ def test_lexical_cache_is_bounded_and_custom_hash_cannot_bypass_policy(): with pytest.raises(policy.DeliveryError, match="private or unsafe"): policy.relative_name(Collision("backend/../escape")) assert policy._cached_relative_name.cache_info() == before - for index in range(9000): + for index in range(20000): policy.relative_name("backend/member-" + str(index)) - assert policy._cached_relative_name.cache_info().currsize == 8192 + assert policy._cached_relative_name.cache_info().currsize == 16384 finally: policy._cached_relative_name.cache_clear() diff --git a/judge/crates/sandbox/src/restore_copy/main.rs b/judge/crates/sandbox/src/restore_copy/main.rs index b15578f..5c53814 100644 --- a/judge/crates/sandbox/src/restore_copy/main.rs +++ b/judge/crates/sandbox/src/restore_copy/main.rs @@ -29,6 +29,29 @@ fn remaining(deadline: u64) -> Result<()> { Ok(()) } +// Preserve contiguous hash/path order while minimizing the heavier byte group. +// Equal-byte choices prefer the original balanced file count (including zeroes). +fn balanced_split(rows: &[restore_io::RestoreFile]) -> usize { + if rows.len() <= 1 { + return rows.len(); + } + let middle = rows.len().div_ceil(2); + let total: u64 = rows.iter().map(|row| row.bytes).sum(); + let mut prefix = 0; + let mut best = (u64::MAX, usize::MAX); + let mut chosen = middle; + for (index, row) in rows.iter().take(rows.len() - 1).enumerate() { + prefix += row.bytes; + let split = index + 1; + let cost = (prefix.max(total - prefix), split.abs_diff(middle)); + if cost < best { + best = cost; + chosen = split; + } + } + chosen +} + enum Job { Copy(Vec, u64), Read(Vec, u64), @@ -89,18 +112,18 @@ impl Pool { roots: &Roots, deadline: u64, ) -> Result> { - let size = rows.len().div_ceil(2); - let mut groups = vec![Vec::new(), Vec::new()]; - for (i, row) in rows.into_iter().enumerate() { - groups[i / size].push(row); - } + let count = rows.len(); + let split = balanced_split(&rows); + let mut rows = rows; + let second = rows.split_off(split); + let groups = [rows, second]; for (worker, group) in self.workers.iter().zip(groups) { worker .sender .send(Job::Copy(group, deadline)) .map_err(|_| Error::Filesystem)?; } - let mut pending = Vec::with_capacity(size * 2); + let mut pending = Vec::with_capacity(count); let mut error = None; // Observe both completions even when the first failed. No sync/readback // and no success response can outrun an outstanding copy worker. @@ -119,17 +142,15 @@ impl Pool { return Err(e); } roots.sync(deadline)?; - let mut groups = vec![Vec::new(), Vec::new()]; - for (i, row) in pending.into_iter().enumerate() { - groups[i / size].push(row); - } + let second = pending.split_off(split); + let groups = [pending, second]; for (worker, group) in self.workers.iter().zip(groups) { worker .sender .send(Job::Read(group, deadline)) .map_err(|_| Error::Filesystem)?; } - let mut verified = Vec::with_capacity(size * 2); + let mut verified = Vec::with_capacity(count); for worker in &self.workers { match worker.receiver.recv() { Ok(Completed::Read(Ok(rows))) => verified.extend(rows), @@ -386,6 +407,86 @@ mod tests { roots.unchanged().unwrap_or_else(|_| panic!("root anchors")); } + fn weighted_rows(weights: &[u64]) -> Vec { + weights + .iter() + .enumerate() + .map(|(index, &bytes)| restore_io::RestoreFile { + path: format!("artifacts/file-{index}"), + uid: 1000, + gid: 1000, + mode: 0o644, + bytes, + sha256: "0".repeat(64), + }) + .collect() + } + + #[test] + fn every_small_weight_inventory_has_minimal_contiguous_heavier_group() { + assert_eq!(balanced_split(&[]), 0); + assert_eq!(balanced_split(&weighted_rows(&[9])), 1); + assert_eq!(balanced_split(&weighted_rows(&[0; 5])), 3); + assert_eq!(balanced_split(&weighted_rows(&[1, 1, 1, 20])), 3); + assert_eq!(balanced_split(&weighted_rows(&[20, 1, 1, 1])), 1); + for count in 2u32..=6 { + for mut encoded in 0..3usize.pow(count) { + let weights: Vec<_> = (0..count) + .map(|_| { + let weight = [0, 1, 13][encoded % 3]; + encoded /= 3; + weight + }) + .collect(); + let rows = weighted_rows(&weights); + let split = balanced_split(&rows); + assert!(split > 0 && split < rows.len()); + let cost = |at| { + let (a, b) = weights.split_at(at); + a.iter().sum::().max(b.iter().sum()) + }; + assert!((1..rows.len()).all(|other| cost(split) <= cost(other))); + } + } + } + + #[test] + fn skewed_groups_keep_every_copy_sync_readback_and_final_witness() { + for weights in [[64, 64, 64, 4096], [4096, 64, 64, 64]] { + let fixture = files::Fixture::new(); + let roots = Arc::new(fixture.roots()); + let mut rows = weighted_rows(&weights); + use std::os::unix::fs::MetadataExt; + let owner = fs::metadata(fixture.source()).unwrap(); + for row in &mut rows { + let data = vec![b'x'; row.bytes as usize]; + row.sha256 = format!("{:x}", Sha256::digest(&data)); + row.uid = owner.uid(); + row.gid = owner.gid(); + let source = fixture.source().join(&row.sha256); + fs::write(&source, &data).unwrap(); + fs::set_permissions(&source, fs::Permissions::from_mode(0o600)).unwrap(); + } + let pool = Pool::new(Arc::clone(&roots)).unwrap_or_else(|_| panic!("pool")); + let verified = pool + .materialize(rows, &roots, now().unwrap() + 10_000_000_000) + .unwrap_or_else(|_| panic!("skewed verified copies")); + assert_eq!(verified.len(), weights.len()); + for row in &verified { + roots + .final_file(row) + .unwrap_or_else(|_| panic!("final witness")); + let name = row.row.path.strip_prefix("artifacts/").unwrap(); + assert_eq!( + fs::read(fixture.target().join(name)).unwrap(), + vec![b'x'; row.row.bytes as usize] + ); + } + drop(pool); + roots.unchanged().unwrap_or_else(|_| panic!("root anchors")); + } + } + #[test] fn request_line_requires_finite_complete_frame() { assert!(matches!( diff --git a/scripts/release/README.md b/scripts/release/README.md index de0edbf..2b473b1 100644 --- a/scripts/release/README.md +++ b/scripts/release/README.md @@ -478,7 +478,7 @@ time from a locked iterator; there is no Future for each file or eagerly queued inventory. Failure or deadline expiry stops dispatch, and all running calls join before an error or result returns. Results retain the input order. Payload path spelling checks use the same immutable allowlist and may retain at -most 8192 successful, length-bounded strings per process. Custom string subclasses +most 16384 successful, length-bounded strings per process. Custom string subclasses bypass this cache. Byte reads, filesystem topology, ownership and all metadata checks remain live; no filesystem or execution result enters the lexical cache. @@ -511,6 +511,10 @@ directory finalization and success receipts wait for all dispatched work. On failure, queued work is cancelled and running calls are joined before the error returns. Private partial data is retained for inspection. The original operator CPU, memory, PID and deadline limits still apply. +The native restore helper keeps two joined threads and partitions each fixed batch +into contiguous groups that minimize the heavier byte count. Ties prefer balanced +file counts. Copy order, synchronization, complete byte readback and final +nine-field witnesses are preserved; no extra worker or batch capacity is added. Old configuration and maintenance receipts go into inactive `recovery/ID/history/`. The new boot's Sandbox preparation owns cgroup recreation. Unknown/nonempty targets and differing existing files are refused; partial restores diff --git a/scripts/release/policy.py b/scripts/release/policy.py index f445843..b9eba33 100644 --- a/scripts/release/policy.py +++ b/scripts/release/policy.py @@ -98,8 +98,8 @@ def _relative_name(value: str) -> str: # Only immutable lexical policy results are reused. No byte, path existence, # metadata, ownership, topology or execution decision is cached here. At most -# 8192 already length-bounded strings survive in one process. -_cached_relative_name = lru_cache(maxsize=8192)(_relative_name) +# 16384 already length-bounded strings survive in one process. +_cached_relative_name = lru_cache(maxsize=16384)(_relative_name) def relative_name(value: Any) -> str: