#!/usr/bin/env python3 """Linux task-I/O observations. Only new files in a private temp directory. 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 subprocess import sys import tempfile import traceback FIELDS = ("rchar", "wchar", "syscr", "syscw", "read_bytes", "write_bytes", "cancelled_write_bytes") CHILDREN = set() def require(condition, message): if not condition: raise RuntimeError(message) def snapshot(pid): values = {} for line in Path(f"/proc/{pid}/io").read_text().splitlines(): key, value = line.split(":", 1) values[key] = int(value) require(all(key in values for key in FIELDS), "Missing I/O accounting fields") return {key: values[key] for key in FIELDS} def delta(before, after): result = {key: after[key] - before[key] for key in FIELDS} require(all(value >= 0 for value in result.values()), "Raw counter decreased") return result def stopped(pid): actual, status = os.waitpid(pid, os.WUNTRACED) if os.WIFEXITED(status) or os.WIFSIGNALED(status): CHILDREN.discard(pid) require(actual == pid and os.WIFSTOPPED(status) and os.WSTOPSIG(status) == signal.SIGSTOP, f"Worker did not stop normally: wait status {status}") return snapshot(pid) def spawn(action): pid = os.fork() if pid == 0: try: os.kill(os.getpid(), signal.SIGSTOP) action() os.kill(os.getpid(), signal.SIGSTOP) os._exit(0) except BaseException: traceback.print_exc() os._exit(1) CHILDREN.add(pid) return pid, stopped(pid) def finish(pid): os.kill(pid, signal.SIGCONT) actual, status = os.waitpid(pid, 0) CHILDREN.discard(pid) require(actual == pid and os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, "Worker failed during exit") def measure(action): pid, before = spawn(action) os.kill(pid, signal.SIGCONT) after = stopped(pid) return pid, {"before": before, "after": after, "delta": delta(before, after)} def write_all(fd, data): remaining = memoryview(data) while remaining: count = os.write(fd, remaining) require(count > 0, "No write progress") remaining = remaining[count:] def trial(root, size, repetition): cached = root / f"cached-{repetition}" with cached.open("wb", buffering=0) as stream: write_all(stream.fileno(), b"A" * size) os.fsync(stream.fileno()) def read_cached(): with cached.open("rb", buffering=0) as stream: count = 0 while True: data = stream.read(65536) if not data: break require(data == b"A" * len(data), "Read content mismatch") count += len(data) require(count == size, "Read length mismatch") pid, read = measure(read_cached) finish(pid) require(read["delta"]["rchar"] == size, "Read accounting differs from returned bytes") mapped = root / f"mapped-{repetition}" with mapped.open("wb", buffering=0) as stream: stream.truncate(size) os.fsync(stream.fileno()) def modify_mapping(): with mapped.open("r+b", buffering=0) as stream: with mmap.mmap(stream.fileno(), size, access=mmap.ACCESS_WRITE) as view: view[:] = b"M" * size pid, mapped_write = measure(modify_mapping) finish(pid) require(mapped_write["delta"]["wchar"] == 0, "Unexpected write-family payload") require(mapped.read_bytes() == b"M" * size, "Mapped contents mismatch") return { "repetition": repetition, "cached_read": read, "mapped_write": mapped_write, "dirty_truncate": cross_truncate(root, size, repetition, False), "synced_truncate": cross_truncate(root, size, repetition, True), } def cross_truncate(root, size, repetition, sync_first): shared = root / f"shared-{repetition}-{sync_first}" shared.touch() def write_buffered(): with shared.open("r+b", buffering=0) as stream: write_all(stream.fileno(), b"D" * size) if sync_first: os.fsync(stream.fileno()) writer_pid, writer = measure(write_buffered) def truncate(): with shared.open("r+b", buffering=0) as stream: os.ftruncate(stream.fileno(), 0) truncator_pid, truncator = measure(truncate) writer_after_truncate = snapshot(writer_pid) require(shared.stat().st_size == 0, "Truncation did not complete") require(writer_after_truncate == writer["after"], "Stopped writer counters changed") require(writer["delta"]["wchar"] == size, "Writer byte count mismatch") finish(truncator_pid) finish(writer_pid) net = truncator["delta"]["write_bytes"] - truncator["delta"]["cancelled_write_bytes"] return { "writer_fsync_before_truncate": sync_first, "writer": writer, "truncator": truncator, "writer_after_truncate": writer_after_truncate, "truncator_write_minus_cancelled_bytes": net, "final_file_size_bytes": 0, } def timeout(_signum, _frame): raise TimeoutError("Experiment exceeded its 60-second limit") 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, default=4, choices=range(1, 17)) parser.add_argument("--temp-dir", help="Existing directory for disposable files") parser.add_argument("--output", type=Path, help="New output path; refuses overwrite") args = parser.parse_args() require(sys.platform == "linux", "Linux is required") size = args.mib * 1024 * 1024 signal.signal(signal.SIGALRM, timeout) signal.alarm(60) try: with tempfile.TemporaryDirectory(prefix="hesela-io-", dir=args.temp_dir) as directory: root = Path(directory) filesystem = subprocess.check_output( ["stat", "-f", "-c", "%T", directory], text=True).strip() observations = [trial(root, size, i + 1) for i in range(args.repetitions)] result = { "schemaVersion": 1, "kind": "measured-linux-task-accounting", "generatedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(), "sourceSha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "environment": { "kernel": platform.release(), "machine": platform.machine(), "python": platform.python_version(), "filesystemMagicLabel": filesystem, "pageSizeBytes": os.sysconf("SC_PAGE_SIZE"), }, "parameters": {"repetitions": args.repetitions, "payloadBytes": size}, "method": "External /proc/PID/io snapshots while workers are SIGSTOP-stopped; no worker reads its own counters; snapshots precede wait/reaping; no cache dropping or system tuning.", "limits": [ "One host; not device traffic, a performance benchmark or a durability test.", "Read cache residency and cancellation depend on writeback/reclaim timing; zero cancellation is a valid result.", "Filesystem magic label does not identify exact ext-family version or mount options.", "No NFS, SMB, direct I/O, cgroups, process-tree aggregation or flash-device counters tested.", ], "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) for pid in list(CHILDREN): try: os.kill(pid, signal.SIGKILL) except ProcessLookupError: pass try: os.waitpid(pid, 0) except ChildProcessError: pass if __name__ == "__main__": main()