#!/usr/bin/env python3 # SPDX-License-Identifier: CC0-1.0 """Generate the compact dataset and optional COM mirror from frozen observations.""" import argparse import csv import hashlib import io import json from pathlib import Path HERE = Path(__file__).resolve().parent SLUG = "zero-copy-source-lifetime" COLUMNS = [ ("repetition", "Sequential repetition number; not an independent fleet sample."), ("case", "Named sequence defined by the Python harness."), ("page_bytes", "Requested and received payload length in bytes."), ("enqueue_bytes", "Successful file-to-pipe or buffered pipe-write return value in bytes."), ("received_byte", "Uniform ASCII byte observed at pipe consumption."), ("path_byte", "Uniform ASCII byte read by opening the source pathname after the operation."), ("old_inode_byte", "Uniform ASCII byte read through the original writable descriptor."), ("path_same_inode", "Whether device and inode identity still match the initial source."), ("matches_before", "Whether received SHA-256 equals the hash computed before enqueue."), ] def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest() def require(condition, message): if not condition: raise ValueError(message) def validate_observations(raw, supplement=False): """Reject contradictory receipts before any derived output is written.""" cases = ("sendfile", "buffered") if supplement else ( "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", ) repetitions = raw.get("repetitions") require(type(repetitions) is int and 1 <= repetitions <= 100, "Invalid repetitions") require(raw.get("schemaVersion") == 1, "Unsupported observation schema") require(raw.get("casesPerRepetition") == len(cases), "Case count mismatch") rows = raw.get("observations") require(isinstance(rows, list) and len(rows) == repetitions * len(cases), "Incomplete observation population") seen, outcomes = set(), {} for row in rows: repetition = row.get("repetition") case = row.get("mode" if supplement else "case") require(type(repetition) is int and 1 <= repetition <= repetitions, "Invalid repetition identity") require(case in cases and (repetition, case) not in seen, "Unknown or duplicate case") seen.add((repetition, case)) size = row.get("page_size_bytes") require(type(size) is int and 0 < size <= 1024 * 1024, "Unsupported page size") require(type(row.get("enqueue_return_bytes")) is int and row["enqueue_return_bytes"] == size, "Short or invalid enqueue") names = ("first_consumer", "second_consumer") if supplement else ( "source_before", "received", "path_after", "old_inode_after") for name in names: payload = row[name] require(type(payload.get("bytes")) is int and payload["bytes"] == size, "Payload length mismatch") letter = payload.get("uniform_byte") require(letter in ("A", "B", "C"), "Unsupported payload; review raw evidence") expected = hashlib.sha256(letter.encode("ascii") * size).hexdigest() require(payload.get("sha256") == expected, "Payload hash mismatch") if supplement: require(type(row.get("tee_return_bytes")) is int and row["tee_return_bytes"] == size, "Short or invalid tee") comparison = row.get("consumer_hashes_equal") expected = row["first_consumer"]["sha256"] == row["second_consumer"]["sha256"] else: require(type(row.get("path_same_inode")) is bool, "Invalid inode identity flag") require(row["source_before"]["uniform_byte"] == "A", "Invalid initial payload") require(type(row.get("pipe_capacity_bytes")) is int and row["pipe_capacity_bytes"] >= size, "Insufficient pipe capacity") comparison = row.get("received_matches_pretransfer_hash") expected = row["source_before"]["sha256"] == row["received"]["sha256"] require(type(comparison) is bool and comparison == expected, "Hash comparison mismatch") outcome = tuple(row[name]["sha256"] for name in names) require(case not in outcomes or outcomes[case] == outcome, "Repetitions differ; revise the uniform-outcome publication before exporting") outcomes[case] = outcome def generate(com_root=None): raw_path = HERE / "zero-copy-lifetime.json" raw = json.loads(raw_path.read_text()) if raw["sourceSha256"] != sha(HERE / "zero-copy-lifetime.py"): raise ValueError("Source hash mismatch: regenerate observations first") supplement = json.loads((HERE / "tee-source-lifetime.json").read_text()) if (supplement["sourceSha256"] != sha(HERE / "tee-source-lifetime.py") or supplement["helperSha256"] != raw["sourceSha256"]): raise ValueError("Supplement source hashes mismatch") validate_observations(raw) validate_observations(supplement, supplement=True) rows = [] for row in raw["observations"]: rows.append(dict(zip([c[0] for c in COLUMNS], [ row["repetition"], row["case"], row["page_size_bytes"], row["enqueue_return_bytes"], row["received"]["uniform_byte"], row["path_after"]["uniform_byte"], row["old_inode_after"]["uniform_byte"], row["path_same_inode"], row["received_matches_pretransfer_hash"], ]))) buffer = io.StringIO(newline="") writer = csv.DictWriter(buffer, fieldnames=[c[0] for c in COLUMNS], lineterminator="\n") writer.writeheader() writer.writerows(rows) csv_path = HERE.parent / "data" / (SLUG + ".csv") (HERE.parent / "data/datasets").mkdir(parents=True, exist_ok=True) csv_path.write_text(buffer.getvalue(), encoding="utf-8") meta = { "slug": SLUG, "title": "Linux zero-copy source lifetime: eight file-to-pipe checks", "summary": f"{len(rows)} file-to-pipe observations, plus {len(supplement['observations'])} separate tee(2) checks, distinguish source stability from enqueue and consumer completion.", "published": "2026-10-18", "updated": "2026-10-18", "method": raw["method"], "limits": raw["limits"], "license": "Hesela's original harness, generated observations and derived CSV are dedicated under CC0-1.0. Linked documentation and Linux source retain their own licenses and are not redistributed.", "licenseUrl": "https://creativecommons.org/publicdomain/zero/1.0/", "downloadPath": "/data/" + SLUG + ".csv", "rawPath": "/research/zero-copy-lifetime.json", "codePath": "/research/zero-copy-lifetime.py", "generatorPath": "/research/build-zero-copy-dataset.py", "testsPath": "/research/test_zero_copy_lifetime.py", "validationTestsPath": "/research/test_zero_copy_dataset.py", "humanPage": "https://hesela.com/analyses/zero-copy-source-lifetime/", "rowCount": len(rows), "columns": [{"name": name, "description": description} for name, description in COLUMNS], "rows": rows, "environment": raw["environment"], "generatedAt": raw["generatedAt"], "repetitions": raw["repetitions"], "citations": json.loads((HERE / "zero-copy-sources.json").read_text()), "supplement": { "title": "tee(2) split-consumer observations", "rawPath": "/research/tee-source-lifetime.json", "codePath": "/research/tee-source-lifetime.py", "testsPath": "/research/test_tee_lifetime.py", "rowCount": len(supplement["observations"]), "sha256": {name: sha(HERE / name) for name in ["tee-source-lifetime.json", "tee-source-lifetime.py", "test_tee_lifetime.py"]}, }, "sha256": { "raw": sha(raw_path), "csv": sha(csv_path), "harness": sha(HERE / "zero-copy-lifetime.py"), "tests": sha(HERE / "test_zero_copy_lifetime.py"), "generator": sha(Path(__file__)), "validationTests": sha(HERE / "test_zero_copy_dataset.py"), }, } target = HERE.parent / "data/datasets" / (SLUG + ".json") target.write_text(json.dumps(meta, indent=2) + "\n", encoding="utf-8") if com_root: mirror = Path(com_root) / "sites/hesela-com/src/data/research/zero-copy-lifetime.json" mirror.parent.mkdir(parents=True, exist_ok=True) mirror.write_bytes(raw_path.read_bytes()) (mirror.parent / "tee-source-lifetime.json").write_bytes( (HERE / "tee-source-lifetime.json").read_bytes()) print(json.dumps({"rows": len(rows), "hashes": meta["sha256"]})) if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--com-root", type=Path) args = parser.parse_args() generate(args.com_root)