#!/usr/bin/env python3 # SPDX-License-Identifier: CC0-1.0 """Observe Linux file-to-pipe source lifetime; never open existing user files.""" import argparse import contextlib import datetime import fcntl import hashlib import json import os from pathlib import Path import platform import subprocess import tempfile CASES = ( "buffered-overwrite", "sendfile-overwrite", "splice-overwrite", "sendfile-overwrite-fsync", "sendfile-replace-path", "sendfile-replace-then-write-old", "sendfile-close-then-overwrite", "sendfile-consume-then-overwrite", ) def require(condition, message): if not condition: raise RuntimeError(message) def digest(data): return hashlib.sha256(data).hexdigest() def write_all(fd, data): offset = 0 while offset < len(data): n = os.pwrite(fd, data[offset:], offset) require(n > 0, "pwrite made no progress") offset += n def read_exact(fd, size): chunks = [] remaining = size while remaining: chunk = os.read(fd, remaining) require(bool(chunk), "early pipe EOF") chunks.append(chunk) remaining -= len(chunk) return b"".join(chunks) def payload_summary(data): return { "bytes": len(data), "sha256": digest(data), "uniform_byte": chr(data[0]) if data and data == data[:1] * len(data) else None, } def probe(case, parent=None): if case not in CASES: raise ValueError("unknown case") size = os.sysconf("SC_PAGESIZE") original, changed, old_changed = (letter * size for letter in (b"A", b"B", b"C")) with tempfile.TemporaryDirectory(prefix="hesela-zero-copy-", dir=parent) as directory: source = Path(directory) / "source" with contextlib.ExitStack() as cleanup: def own(fd): cleanup.callback(os.close, fd) return fd writer = own(os.open(source, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600)) write_all(writer, original) os.fsync(writer) reader = os.open(source, os.O_RDONLY) try: initial_identity = os.fstat(reader) before = os.pread(reader, size, 0) require(before == original, "invalid initial payload") before_digest = digest(before) read_pipe, write_pipe = os.pipe2(os.O_NONBLOCK | os.O_CLOEXEC) own(read_pipe) own(write_pipe) capacity = fcntl.fcntl(write_pipe, fcntl.F_GETPIPE_SZ) require(size <= capacity, "one page does not fit in the pipe") if case.startswith("buffered"): # A userspace bytes object holds a copy, not a file-page reference. transferred = os.write(write_pipe, before) elif case.startswith("splice"): transferred = os.splice(reader, write_pipe, size, offset_src=0) else: transferred = os.sendfile(write_pipe, reader, 0, size) # Fail explicitly on short transfer instead of interpreting it as a full page. require(transferred == size, "short transfer; no observation published") events = ["initial-fsync", "hash-source", "enqueue-return"] received = None if case == "sendfile-consume-then-overwrite": received = read_exact(read_pipe, size) events.append("consume") if case == "sendfile-close-then-overwrite": os.close(reader) reader = None events.append("close-input-descriptor") if case in ("sendfile-replace-path", "sendfile-replace-then-write-old"): replacement = Path(directory) / "replacement" replacement_fd = own(os.open(replacement, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600)) write_all(replacement_fd, changed) os.fsync(replacement_fd) os.replace(replacement, source) events.append("replace-path") if case == "sendfile-replace-then-write-old": write_all(writer, old_changed) events.append("overwrite-old-inode") else: write_all(writer, changed) events.append("overwrite-source") if case == "sendfile-overwrite-fsync": os.fsync(writer) events.append("second-fsync") if received is None: received = read_exact(read_pipe, size) events.append("consume") path_fd = own(os.open(source, os.O_RDONLY)) path_identity = os.fstat(path_fd) path_payload = os.pread(path_fd, size, 0) require(len(path_payload) == size, "source path size changed unexpectedly") old_payload = os.pread(writer, size, 0) return { "case": case, "events": events, "page_size_bytes": size, "pipe_capacity_bytes": capacity, "enqueue_return_bytes": transferred, "source_before": payload_summary(before), "received": payload_summary(received), "path_after": payload_summary(path_payload), "old_inode_after": payload_summary(old_payload), "path_same_inode": (initial_identity.st_dev, initial_identity.st_ino) == (path_identity.st_dev, path_identity.st_ino), "received_matches_pretransfer_hash": digest(received) == before_digest, } finally: if reader is not None: os.close(reader) def run(repetitions=5, parent=None): if type(repetitions) is not int or not 1 <= repetitions <= 100: raise ValueError("repetitions must be an integer from 1 to 100") require(platform.system() == "Linux", "Linux is required") observations = [] with tempfile.TemporaryDirectory(prefix="hesela-zero-copy-env-", dir=parent) as directory: filesystem = subprocess.run( ["stat", "-f", "-c", "%T", directory], check=True, capture_output=True, text=True, timeout=5, ).stdout.strip() for repetition in range(1, repetitions + 1): for case in CASES: observations.append({"repetition": repetition, **probe(case, directory)}) return { "schemaVersion": 1, "generatedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(), "experiment": "linux-file-to-pipe-source-lifetime", "license": "CC0-1.0", "method": "Sequential one-page file-to-pipe tests; source mutation finishes before pipe consumption except the consume-first control.", "limits": [ "One host and filesystem-magic label; not a filesystem comparison.", "Synthetic A/B/C payloads; measured API observations, not storage-fleet statistics.", "No TCP, MSG_ZEROCOPY, crash, power-loss, throughput or cache-eviction test.", "Mutation cases intentionally violate the documented sendfile source-lifetime requirement.", "A different supported path may copy data; post-mutation B/C results are observations, not a universal API guarantee.", ], "environment": { "kernel": platform.release(), "machine": platform.machine(), "python": platform.python_version(), "filesystem_magic_label": filesystem, }, "sourceSha256": digest(Path(__file__).read_bytes()), "repetitions": repetitions, "casesPerRepetition": len(CASES), "observations": observations, } def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repetitions", type=int, default=5) parser.add_argument("--temp-dir") parser.add_argument("--output", type=Path) args = parser.parse_args() result = run(args.repetitions, args.temp_dir) payload = json.dumps(result, indent=2) + "\n" if args.output: args.output.write_text(payload, encoding="utf-8") else: print(payload, end="") if __name__ == "__main__": main()