From 39a7b2902afc7a801c88ca96bea78b5f2dad6ba3 Mon Sep 17 00:00:00 2001 From: kronecker <2026692687@qq.com> Date: Wed, 28 Oct 2026 02:00:58 +0800 Subject: [PATCH] perf(delivery): batch installs and parallelize cold recovery --- backend/tests/test_delivery.py | 55 +++++++++++++- backend/tests/test_delivery_restore_io.py | 82 +++++++++++++++++++-- scripts/release/install.py | 32 +++++--- scripts/release/restore_io.py | 89 +++++++++++++++-------- 4 files changed, 206 insertions(+), 52 deletions(-) diff --git a/backend/tests/test_delivery.py b/backend/tests/test_delivery.py index 0f7f750..b852b00 100644 --- a/backend/tests/test_delivery.py +++ b/backend/tests/test_delivery.py @@ -365,7 +365,28 @@ def test_changed_copy_retains_incomplete_file_and_never_overwrites_it(tmp_path, assert target.read_bytes() == b"changed" -def test_frozen_copy_reuses_verified_records_and_checks_full_destination(tmp_path, monkeypatch): +@pytest.fixture +def frozen_filesystem_model(tmp_path, monkeypatch): + """Only fsync tiny owned fixtures; real filesystem flush remains VM-only.""" + + class Sync: + def __call__(self, descriptor): + assert stat.S_ISDIR(os.fstat(descriptor).st_mode) + for path in tmp_path.rglob("*"): + if path.is_file(): + assert path.stat().st_uid == os.getuid() and path.stat().st_size < 4096 + with path.open("rb") as stream: + os.fsync(stream.fileno()) + return 0 + + monkeypatch.setattr( + installer.restore_io.ctypes, "CDLL", lambda *_a, **_kw: SimpleNamespace(syncfs=Sync()) + ) + + +def test_frozen_copy_reuses_verified_records_and_checks_full_destination( + tmp_path, monkeypatch, frozen_filesystem_model +): monkeypatch.setattr(installer.preflight, "protected", lambda *args, **kwargs: None) source, target = tmp_path / "source", tmp_path / "target" source.mkdir() @@ -390,7 +411,9 @@ def test_frozen_copy_reuses_verified_records_and_checks_full_destination(tmp_pat assert scanned == [target, target] -def test_frozen_copy_rejects_destination_io_mismatch_before_acceptance(tmp_path, monkeypatch): +def test_frozen_copy_rejects_destination_io_mismatch_before_acceptance( + tmp_path, monkeypatch, frozen_filesystem_model +): # Simulated I/O mismatch in an ordinary owned fixture; Root protection and # actual disk/error isolation are still separate VM acceptance. monkeypatch.setattr(installer.preflight, "protected", lambda *args, **kwargs: None) @@ -416,7 +439,9 @@ def test_frozen_copy_rejects_destination_io_mismatch_before_acceptance(tmp_path, installer.frozen_copy(source, target, verified_records=records) -def test_frozen_copy_rejects_changed_source_against_trusted_records(tmp_path, monkeypatch): +def test_frozen_copy_rejects_changed_source_against_trusted_records( + tmp_path, monkeypatch, frozen_filesystem_model +): # Ordinary fixture bytes exercise the deferred batch validation. Actual # Root protection and installation are separately verified in the VM. monkeypatch.setattr(installer.preflight, "protected", lambda *args, **kwargs: None) @@ -439,6 +464,30 @@ def test_frozen_copy_rejects_changed_source_against_trusted_records(tmp_path, mo installer.frozen_copy(source, target, verified_records=records) +def test_frozen_copy_rejects_sync_failure_even_when_destination_bytes_match(tmp_path, monkeypatch): + class FailedSync: + def __call__(self, _descriptor): + return -1 + + monkeypatch.setattr(installer.preflight, "protected", lambda *args, **kwargs: None) + monkeypatch.setattr( + installer.restore_io.ctypes, + "CDLL", + lambda *_a, **_kw: SimpleNamespace(syncfs=FailedSync()), + ) + source, target = tmp_path / "source", tmp_path / "target" + source.mkdir() + (source / "backend").mkdir() + payload = source / "backend/ordinary.txt" + payload.write_bytes(b"known ordinary bytes") + payload.chmod(0o444) + (source / "backend").chmod(0o555) + records = manifest.inventory(source) + with pytest.raises(policy.DeliveryError, match="filesystem synchronization failed"): + installer.frozen_copy(source, target, verified_records=records) + assert (target / "backend/ordinary.txt").read_bytes() == payload.read_bytes() + + def test_kernel_observation_keeps_cumulative_counters_and_rejects_duplicate_rows(): raw = ( "usage_usec 17\nuser_usec 11\nsystem_usec 6\nnr_periods 9\nnr_throttled 2\nthrottled_usec 3" diff --git a/backend/tests/test_delivery_restore_io.py b/backend/tests/test_delivery_restore_io.py index 586ca32..53809c1 100644 --- a/backend/tests/test_delivery_restore_io.py +++ b/backend/tests/test_delivery_restore_io.py @@ -2,6 +2,7 @@ import errno import hashlib +import multiprocessing import os import stat import time @@ -65,6 +66,19 @@ def assert_closed(descriptors): assert failure.value.errno == errno.EBADF +class ReadAfterSync: + """Picklable observation of actual tiny fixture reads in the child processes.""" + + def __init__(self, synced, reads, descriptors): + self.synced, self.reads, self.descriptors = synced, reads, descriptors + + def __call__(self, path, item, uid, gid, mode, deadline): + assert self.synced.read_bytes() == b"sync completed" + assert all(os.fstat(fd).st_ino > 0 for fd in self.descriptors) + (self.reads / item["path"]).touch() + return restore.readback(path, item, uid, gid, mode, deadline) + + def test_bounded_batches_group_identical_objects_without_losing_paths(): rows = [ {"path": str(index), "sha256": str(index % 2), "bytes": 1} @@ -89,22 +103,31 @@ def test_repeated_objects_expand_to_independent_durable_verified_destinations( rows, source, destination = fixture_rows(tmp_path, [b"first", b"other", b"first", b"first"]) events = [] target = destination(rows[0])[0].parent + synced, reads = tmp_path / "synced", tmp_path / "reads" + reads.mkdir() with closing(filesystem_model(monkeypatch, target, events)) as filesystems: descriptors = [descriptor for _, descriptor, _ in filesystems.handles.values()] assert list(target.iterdir()) == [] - def readback(path, item, uid, gid, mode, deadline): - assert events and events[0][0] == "sync" - assert all(os.fstat(fd).st_ino > 0 for fd in descriptors) - events.append(("read", item["path"])) - return restore.readback(path, item, uid, gid, mode, deadline) + original_sync = filesystems.sync + + def sync(deadline): + original_sync(deadline) + synced.write_bytes(b"sync completed") + + monkeypatch.setattr(filesystems, "sync", sync) verified = restore_io.materialize( - rows, source, destination, readback, filesystems, time.monotonic() + 120 + rows, + source, + destination, + ReadAfterSync(synced, reads, descriptors), + filesystems, + time.monotonic() + 120, ) assert_closed(descriptors) assert set(verified) == {row["path"] for row in rows} - assert len([event for event in events if event[0] == "read"]) == len(rows) + assert len(list(reads.iterdir())) == len(rows) inodes = set() for row in rows: path, uid, gid, mode = destination(row) @@ -117,6 +140,7 @@ def test_repeated_objects_expand_to_independent_durable_verified_destinations( def test_existing_wrong_bytes_are_refused_without_overwrite(tmp_path, monkeypatch): + prior_children = {child.pid for child in multiprocessing.active_children()} rows, source, destination = fixture_rows(tmp_path, [b"new"]) path, _, _, mode = destination(rows[0]) path.write_bytes(b"bad") @@ -128,6 +152,7 @@ def test_existing_wrong_bytes_are_refused_without_overwrite(tmp_path, monkeypatc rows, source, destination, restore.readback, filesystems, time.monotonic() + 120 ) assert path.read_bytes() == b"bad" and restore.file_witness(path.stat()) == before + assert {child.pid for child in multiprocessing.active_children()} == prior_children def test_failed_sync_keeps_private_files_and_cannot_return_verified_witnesses( @@ -259,3 +284,46 @@ def test_constructor_closes_earlier_cursor_when_later_anchor_is_invalid(tmp_path restore_io.Filesystems([valid, invalid], time.monotonic() + 120) assert len(opened) == 1 assert_closed(opened) + + +def test_actual_two_copy_processes_join_before_return_and_preserve_all_bytes(tmp_path, monkeypatch): + rows, source, destination = fixture_rows( + tmp_path, [b"known first", b"known second", b"known third", b"known fourth"] + ) + prior_children = {child.pid for child in multiprocessing.active_children()} + markers = tmp_path / "worker-identities" + markers.mkdir() + original = restore_io.pending_file + + def record_worker(*args): + (markers / str(os.getpid())).touch() + deadline = time.monotonic() + 5 + while len(list(markers.iterdir())) < 2: + assert time.monotonic() < deadline + time.sleep(0.01) + return original(*args) + + monkeypatch.setattr(restore_io, "pending_file", record_worker) + with closing(filesystem_model(monkeypatch, destination(rows[0])[0].parent, [])) as fs: + verified = restore_io.materialize( + rows, source, destination, restore.readback, fs, time.monotonic() + 120 + ) + assert len(list(markers.iterdir())) == 2 + assert {child.pid for child in multiprocessing.active_children()} == prior_children + assert all(not (restore_io.Path("/proc") / p.name).exists() for p in markers.iterdir()) + assert set(verified) == {row["path"] for row in rows} + for row in rows: + path, uid, gid, mode = destination(row) + assert path.read_bytes() == source(row).read_bytes() + restore.unchanged_file(path, verified[row["path"]], uid, gid, mode) + + +def test_process_copy_refuses_fork_from_a_multithreaded_owner(tmp_path, monkeypatch): + rows, source, destination = fixture_rows(tmp_path, [b"known fixture"]) + monkeypatch.setattr(restore_io.threading, "active_count", lambda: 2) + with closing(filesystem_model(monkeypatch, destination(rows[0])[0].parent, [])) as fs: + with pytest.raises(DeliveryError, match="one POSIX owner thread"): + restore_io.materialize( + rows, source, destination, restore.readback, fs, time.monotonic() + 120 + ) + assert not destination(rows[0])[0].exists() diff --git a/scripts/release/install.py b/scripts/release/install.py index 53344f9..f1e8162 100644 --- a/scripts/release/install.py +++ b/scripts/release/install.py @@ -8,7 +8,7 @@ import stat import subprocess from pathlib import Path -from . import file_copy, manifest, preflight +from . import file_copy, manifest, preflight, restore_io from .layout import RELEASES as RELEASES from .policy import DeliveryError, canonical, digest, fields, read_json, sha256 @@ -101,6 +101,7 @@ def _copy_verified( *, readback: bool, expected_bytes: int | None = None, + synchronize: bool = True, ) -> None: preflight.protected(source) sha256(expected) @@ -137,7 +138,8 @@ def _copy_verified( "source changed after bundle verification; incomplete copy retained" ) os.fchmod(outgoing, mode) - os.fsync(outgoing) + if synchronize: + os.fsync(outgoing) if readback and digest(target) != expected: raise DeliveryError("installed bytes differ; incomplete release retained") finally: @@ -242,9 +244,6 @@ def frozen_copy( def _copy_new_tree(source: Path, target: Path, records: list[dict], directories: list[str]) -> None: """Create a new sealed tree; the caller must verify the complete destination.""" - target.mkdir(mode=0o700) - for name in sorted(directories, key=lambda name: (name.count("/"), name)): - (target / name).mkdir(mode=0o700) def copy(record: dict) -> None: # frozen_copy verifies this image immediately; a release staging tree @@ -256,14 +255,25 @@ def _copy_new_tree(source: Path, target: Path, records: list[dict], directories: int(record["mode"], 8), readback=False, expected_bytes=record["bytes"], + synchronize=False, ) - file_copy.bounded_copies(records, copy) - for name in sorted(directories, key=lambda name: name.count("/"), reverse=True): - (target / name).chmod(0o555) - sync_directory(target / name) - target.chmod(0o555) - sync_directory(target) + # Retain the filesystem error cursor before creating any destination. + # The enclosing installation service/caller owns its existing deadline. + filesystems = restore_io.Filesystems([target], None) + try: + target.mkdir(mode=0o700) + for name in sorted(directories, key=lambda name: (name.count("/"), name)): + (target / name).mkdir(mode=0o700) + for batch in restore_io.batches(records): + file_copy.bounded_copies(batch, copy) + filesystems.sync(None) + for name in sorted(directories, key=lambda name: name.count("/"), reverse=True): + (target / name).chmod(0o555) + target.chmod(0o555) + filesystems.sync(None) + finally: + filesystems.close() def files(bundle: Path, expected: str) -> dict: diff --git a/scripts/release/restore_io.py b/scripts/release/restore_io.py index 1b43f39..33d7daa 100644 --- a/scripts/release/restore_io.py +++ b/scripts/release/restore_io.py @@ -1,11 +1,13 @@ """Bounded cold-copy batches, durable before full destination byte readback.""" import ctypes +import multiprocessing import os import stat +import threading import time from collections.abc import Callable, Iterator, Sequence -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import ProcessPoolExecutor from pathlib import Path from . import file_copy @@ -13,10 +15,11 @@ from .policy import DeliveryError BATCH_FILES = 256 BATCH_BYTES = 64 * 1024**2 +COPY_PROCESSES = 2 -def remaining(deadline: float) -> None: - if time.monotonic() >= deadline: +def remaining(deadline: float | None) -> None: + if deadline is not None and time.monotonic() >= deadline: raise DeliveryError("recovery copy deadline exceeded; private partial target retained") @@ -41,7 +44,7 @@ def batches(rows: Sequence[dict]) -> Iterator[list[dict]]: class Filesystems: """Hold filesystem error cursors from before the first destination write.""" - def __init__(self, paths: Sequence[Path], deadline: float): + def __init__(self, paths: Sequence[Path], deadline: float | None): self.handles: dict[int, tuple[Path, int, int]] = {} self.closed = False remaining(deadline) @@ -82,7 +85,7 @@ class Filesystems: if self.closed or not stat.S_ISDIR(parent.st_mode) or parent.st_dev not in self.handles: raise DeliveryError("recovery destination has no pre-write filesystem cursor") - def sync(self, deadline: float) -> None: + def sync(self, deadline: float | None) -> None: if self.closed: raise DeliveryError("recovery filesystem cursor is closed") for device, (path, descriptor, inode) in self.handles.items(): @@ -148,6 +151,19 @@ def pending_file( return witness +def _copy_group(group: list[tuple], deadline: float) -> list[tuple[int, ...]]: + return [ + pending_file(source, path, item, uid, gid, mode, deadline) + for item, source, path, uid, gid, mode in group + ] + + +def _read_group(readback: Callable, group: list[tuple], deadline: float) -> list[tuple[int, ...]]: + return [ + readback(path, item, uid, gid, mode, deadline) for item, path, uid, gid, mode, _ in group + ] + + def materialize( rows: Sequence[dict], source: Callable[[dict], Path], @@ -158,42 +174,53 @@ def materialize( ) -> dict[str, tuple[int, ...]]: """Copy, join, sync and independently read each complete destination. - Only four calls are dispatched at a time. A batch retains at most 256 file - records and 64 MiB of logical contents; a larger file is handled alone. + Two joined child processes use the existing exclusive node CPU ceiling; + no private service/cgroup or extra resource budget is introduced. Each + worker processes a bounded half-batch without one Future per file. The + parent retains the pre-write error cursors and flushes only after both + copy groups join, then dispatches full byte readback. A batch retains at + most 256 file records and 64 MiB; a larger file is handled alone. Every returned witness follows an actual filesystem flush and full SHA256 destination read. Exceptions join outstanding work and retain partial files. """ remaining(deadline) + if os.name != "posix" or threading.active_count() != 1: + raise DeliveryError("recovery process workers require one POSIX owner thread") verified = {} - with ThreadPoolExecutor(max_workers=file_copy.IN_FLIGHT) as executor: + executor = ProcessPoolExecutor( + max_workers=COPY_PROCESSES, mp_context=multiprocessing.get_context("fork") + ) + # Shutdown joins every child on either success or failure. The fixed unit + # deadline additionally bounds a blocked syscall or unresponsive child. + with executor: for batch in batches(rows): pending = [] - for offset in range(0, len(batch), file_copy.IN_FLIGHT): - remaining(deadline) - group = [ - (item, *destination(item)) - for item in batch[offset : offset + file_copy.IN_FLIGHT] + remaining(deadline) + group_size = max(1, (len(batch) + COPY_PROCESSES - 1) // COPY_PROCESSES) + groups = [ + [ + (item, source(item), *destination(item)) + for item in batch[offset : offset + group_size] ] - for _, path, _, _, _ in group: + for offset in range(0, len(batch), group_size) + ] + for group in groups: + for _, _, path, _, _, _ in group: filesystems.contains(path) - calls = [ - executor.submit( - pending_file, source(item), path, item, uid, gid, mode, deadline - ) - for item, path, uid, gid, mode in group - ] - for record, call in zip(group, calls, strict=True): - pending.append((*record, call.result())) + calls = [executor.submit(_copy_group, group, deadline) for group in groups] + for group, call in zip(groups, calls, strict=True): + for (item, _, path, uid, gid, mode), witness in zip( + group, call.result(), strict=True + ): + pending.append((item, path, uid, gid, mode, witness)) filesystems.sync(deadline) - for offset in range(0, len(pending), file_copy.IN_FLIGHT): - remaining(deadline) - group = pending[offset : offset + file_copy.IN_FLIGHT] - calls = [ - executor.submit(readback, path, item, uid, gid, mode, deadline) - for item, path, uid, gid, mode, _ in group - ] - for (item, _, _, _, _, before), call in zip(group, calls, strict=True): - actual = call.result() + groups = [ + pending[offset : offset + group_size] + for offset in range(0, len(pending), group_size) + ] + calls = [executor.submit(_read_group, readback, group, deadline) for group in groups] + for group, call in zip(groups, calls, strict=True): + for (item, _, _, _, _, before), actual in zip(group, call.result(), strict=True): if actual != before: raise DeliveryError("recovery destination changed across synchronization") verified[item["path"]] = actual