#!/usr/bin/env python3 """Measure Linux child-I/O inheritance with an external observer. SPDX-License-Identifier: MIT """ import argparse import datetime import hashlib import json import mmap import os from pathlib import Path import platform import signal import struct import subprocess import sys import tempfile import traceback FIELDS = ("rchar", "wchar", "syscr", "syscw", "read_bytes", "write_bytes", "cancelled_write_bytes") STAGES = ("before_write", "after_write", "exited_wnowait", "reaped") def require(condition, message): if not condition: raise RuntimeError(message) def snapshot(pid, task_only=False): suffix = f"task/{pid}/io" if task_only else "io" values = dict(line.split(":", 1) for line in Path(f"/proc/{pid}/{suffix}").read_text().splitlines()) return {key: int(values[key]) for key in FIELDS} def difference(before, after): return {key: after[key] - before[key] for key in FIELDS} def pause(): os.kill(os.getpid(), signal.SIGSTOP) def wait_stopped(pid): actual, status = os.waitpid(pid, os.WUNTRACED) require(actual == pid and os.WIFSTOPPED(status) and os.WSTOPSIG(status) == signal.SIGSTOP, "Worker failed to stop") def wait_exited(pid): actual, status = os.waitpid(pid, 0) require(actual == pid and os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, "Worker failed") def write_all(fd, data): view = memoryview(data) while view: count = os.write(fd, view) require(count > 0, "Write made no progress") view = view[count:] def measured_parent(path, size, shared): child = os.fork() if child == 0: try: pause() if size: fd = os.open(path, os.O_WRONLY) try: write_all(fd, b"C" * size) os.fsync(fd) finally: os.close(fd) pause() os._exit(0) except BaseException: traceback.print_exc() os._exit(1) try: struct.pack_into("q", shared, 0, child) wait_stopped(child) pause() os.kill(child, signal.SIGCONT) wait_stopped(child) pause() os.kill(child, signal.SIGCONT) info = os.waitid(os.P_PID, child, os.WEXITED | os.WNOWAIT) require(info.si_code == os.CLD_EXITED and info.si_status == 0, "Child did not exit successfully") pause() wait_exited(child) child = None pause() finally: if child is not None: try: os.kill(child, signal.SIGKILL) os.waitpid(child, 0) except (ProcessLookupError, ChildProcessError): pass def summarize(stages): base = stages["before_write"] child = stages["after_write"]["child"] before = stages["exited_wnowait"]["parent"] after = stages["reaped"]["parent"] parent_delta = difference(before, after) own_delta = difference(stages["exited_wnowait"]["parent_task"], stages["reaped"]["parent_task"]) inherited = {key: parent_delta[key] - own_delta[key] for key in FIELDS} return { "child_last_stopped": child, "parent_reap_delta": parent_delta, "parent_task_reap_delta": own_delta, "inherited_after_own_task_adjustment": inherited, "parent_delta_by_stage": { name: difference(base["parent"], stage["parent"]) for name, stage in stages.items() }, "retained_child_plus_reaped_parent_wchar": child["wchar"] + after["wchar"] - base["parent"]["wchar"], } def trial(root, size, repetition): path = root / f"child-{size}-{repetition}" path.touch() with mmap.mmap(-1, 8) as shared: parent = os.fork() if parent == 0: code = 0 try: os.setsid() measured_parent(path, size, shared) except BaseException: traceback.print_exc() code = 1 os._exit(code) reaped = False try: stages = {} for name in STAGES: wait_stopped(parent) child = struct.unpack_from("q", shared)[0] stages[name] = { "parent": snapshot(parent), "parent_task": snapshot(parent, task_only=True), "child": snapshot(child) if name in ("before_write", "after_write") else None, } if name == "exited_wnowait": status = Path(f"/proc/{child}/status").read_text() state = next(line for line in status.splitlines() if line.startswith("State:")) require(state.split()[1] == "Z", "Child is not a zombie") if name == "reaped": require(not Path(f"/proc/{child}").exists(), "Child still exists after reap") os.kill(parent, signal.SIGCONT) wait_exited(parent) reaped = True finally: if not reaped: # A separate session contains only this trial's two workers. try: os.killpg(parent, signal.SIGKILL) except ProcessLookupError: try: os.kill(parent, signal.SIGKILL) except ProcessLookupError: pass try: os.waitpid(parent, 0) except ChildProcessError: pass require(path.read_bytes() == b"C" * size, "File payload mismatch") result = summarize(stages) require(result["child_last_stopped"]["wchar"] == size, "Child payload count mismatch") require(result["inherited_after_own_task_adjustment"] == result["child_last_stopped"], "Inherited counters differ from child pre-exit counters") return {"repetition": repetition, "payloadBytes": size, "stages": stages, "summary": result} def timeout(_signum, _frame): raise TimeoutError("Experiment exceeded 60 seconds") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--repetitions", type=int, default=5, choices=range(1, 21)) parser.add_argument("--mib", type=int, nargs="+", default=[0, 1, 4], choices=range(17)) parser.add_argument("--temp-dir") parser.add_argument("--output", type=Path) args = parser.parse_args() require(sys.platform == "linux", "Linux is required") if args.output and args.output.exists(): raise FileExistsError(args.output) signal.signal(signal.SIGCHLD, signal.SIG_DFL) signal.signal(signal.SIGALRM, timeout) signal.alarm(60) try: with tempfile.TemporaryDirectory(prefix="hesela-child-io-", dir=args.temp_dir) as directory: filesystem = subprocess.check_output( ["stat", "-f", "-c", "%T", directory], text=True).strip() observations = [trial(Path(directory), mib * 1048576, i + 1) for mib in args.mib for i in range(args.repetitions)] result = { "schemaVersion": 1, "kind": "measured-linux-child-io-inheritance", "generatedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(), "sourceSha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "license": "MIT", "environment": { "kernel": platform.release(), "machine": platform.machine(), "python": platform.python_version(), "filesystemMagicLabel": filesystem, "pageSizeBytes": os.sysconf("SC_PAGE_SIZE"), }, "parameters": {"repetitionsPerSize": args.repetitions, "payloadBytes": [mib * 1048576 for mib in args.mib]}, "method": "External observer reads stopped parent and stopped child; child zombie state is checked without reading zombie I/O counters; anonymous shared memory carries only the child PID. Parent waits for stopped child, then WEXITED|WNOWAIT, then reaps. Child fsyncs positive-size payloads before exit.", "limits": [ "One host and single-threaded parent/child, default SIGCHLD disposition.", "Upstream v6.8 source explains the mechanism; running Ubuntu kernel may contain downstream patches.", "Counters are accounting, not measured physical device traffic or latency.", "No grandchild chains, orphan adoption, subreapers, ptrace or concurrent thread reaping tested.", "Child I/O sampled at its last stop before _exit; zombie I/O read is not required (permission denied on the measured host).", ], "observations": observations, } serialized = json.dumps(result, indent=2) + "\n" if args.output: with args.output.open("x") as output: output.write(serialized) else: print(serialized, end="") finally: signal.alarm(0) if __name__ == "__main__": main()