#!/usr/bin/env python3 """可复现的应用级批处理断点恢复演示；仅使用 Python 标准库。 样本文字是此脚本内置的合成文字，不读取用户文件，不运行 AI 模型。 驱动程序启动自己的 worker 子进程，worker 完成指定批次后 os._exit(75)， 随后由驱动程序重启。不会终止系统应用，不会执行电脑关机或真实断电。 默认状态文件写入本项目 work/checkpoint_demo/run_<随机ID>/，结果写入 outputs/xuhang/results/checkpoint_result.json。显式路径同样限制在本项目 work/（状态）或 outputs/（结果）以内，且不会递归删除现有文件。 验证内容：应用批处理已完成结果能够恢复；不代表 LLM KV cache、任意模型 执行中的状态恢复、UPS 切换或电池硬件验证。不从运行时间推断功耗或 Wh。 原子替换使用 flush/fsync + os.replace；操作系统/硬盘真实断电持久性另需测试。 """ from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import subprocess import sys import time import uuid OUTPUT_ROOT = Path(__file__).resolve().parent.parent PROJECT_ROOT = OUTPUT_ROOT.parent if OUTPUT_ROOT.name == "outputs" else OUTPUT_ROOT WORK_ROOT = PROJECT_ROOT / "work" EXIT_INTERRUPTED = 75 SCHEMA = "xuhang.checkpoint-demo.v1" SAMPLE_TEXTS = ( "A local document index can be rebuilt after a pause.", "Store completed batch results before accepting another batch.", "A router and a small computer are different electrical loads.", "Measured time is not a measurement of battery energy.", "Synthetic examples contain no personal or private records.", "A checkpoint preserves application progress at a known boundary.", ) def checked_path(candidate: Path, allowed_root: Path, label: str) -> Path: path = candidate.resolve() root = allowed_root.resolve() if path == root or root not in path.parents: raise ValueError(f"{label} must be strictly inside {root}: {path}") return path def atomic_json(path: Path, value: object) -> None: """The old file remains available until replace succeeds; no delete is used.""" path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_name(path.name + "." + uuid.uuid4().hex + ".tmp") with temporary.open("w", encoding="utf-8", newline="\n") as stream: json.dump(value, stream, ensure_ascii=False, sort_keys=True, indent=2) stream.write("\n") stream.flush() os.fsync(stream.fileno()) os.replace(temporary, path) def make_samples(count: int) -> list[str]: return [f"Synthetic sample {i:03d}. {SAMPLE_TEXTS[i % len(SAMPLE_TEXTS)]}" for i in range(count)] def fingerprint(samples: list[str], rounds: int, batch_size: int) -> str: encoded = json.dumps({"samples": samples, "rounds": rounds, "batch_size": batch_size}, sort_keys=True).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def process_sample(index: int, text: str, rounds: int) -> dict: """Deterministic text statistics plus CPU hashing; not inference or encryption.""" normalized = " ".join(text.lower().split()) words = normalized.split() value = normalized.encode("utf-8") for iteration in range(rounds): value = hashlib.sha256(value + iteration.to_bytes(4, "little")).digest() return {"index": index, "words": len(words), "unique_words": len(set(words)), "characters": len(normalized), "work_digest": value.hex()} def output_digest(outputs: list[dict]) -> str: payload = json.dumps(outputs, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(payload).hexdigest() def record_work(path: Path, phase: str, index: int) -> None: # Instrumentation counts actual completed work across process lifetimes. It stores # only item IDs, not results; worker never uses this log to restore its progress. with path.open("a", encoding="utf-8", newline="\n") as stream: stream.write(json.dumps({"phase": phase, "index": index}) + "\n") stream.flush() os.fsync(stream.fileno()) def run_worker(args: argparse.Namespace) -> int: if args.state_dir is None: raise ValueError("--worker requires --state-dir") directory = checked_path(Path(args.state_dir), WORK_ROOT, "state directory") directory.mkdir(parents=True, exist_ok=True) samples = make_samples(args.items) identity = fingerprint(samples, args.rounds, args.batch_size) checkpoint_path = directory / "checkpoint.json" outputs: list[dict] = [] restored_items = 0 if checkpoint_path.exists(): state = json.loads(checkpoint_path.read_text(encoding="utf-8")) if state.get("schema") != SCHEMA or state.get("dataset_fingerprint") != identity: raise ValueError("checkpoint does not match this workload configuration") outputs = state["outputs"] restored_items = state["next_index"] if (restored_items != len(outputs) or restored_items > len(samples) or restored_items % args.batch_size or [row["index"] for row in outputs] != list(range(restored_items))): raise ValueError("checkpoint progress is invalid") if state["outputs_digest"] != output_digest(outputs): raise ValueError("checkpoint digest does not match its stored results") started = time.perf_counter() checkpoint_writes = 0 print(f"worker {args.phase}: restored_items={restored_items}", flush=True) for start in range(restored_items, len(samples), args.batch_size): stop = min(start + args.batch_size, len(samples)) for index in range(start, stop): outputs.append(process_sample(index, samples[index], args.rounds)) record_work(directory / "work_audit.jsonl", args.phase, index) completed_batches = stop // args.batch_size if args.checkpoint_interval and completed_batches % args.checkpoint_interval == 0: atomic_json(checkpoint_path, { "schema": SCHEMA, "dataset_fingerprint": identity, "next_index": stop, "outputs": outputs, "outputs_digest": output_digest(outputs), }) checkpoint_writes += 1 # At a checkpoint boundary, the completed checkpoint is deliberately written # before exit; off-boundary work must be repeated after restart. if completed_batches == args.interrupt_after_batch: print(f"worker {args.phase}: deliberate os._exit({EXIT_INTERRUPTED}) " f"after batch {completed_batches}", flush=True) os._exit(EXIT_INTERRUPTED) atomic_json(directory / f"{args.phase}_result.json", { "schema": SCHEMA, "restored_items": restored_items, "completed_items": len(outputs), "final_digest": output_digest(outputs), "outputs": outputs, "checkpoint_writes": checkpoint_writes, "elapsed_seconds_including_audit_and_checkpoint_io": time.perf_counter() - started, }) print(f"worker {args.phase}: completed_items={len(outputs)}", flush=True) return 0 def child_call(args: argparse.Namespace, directory: Path, phase: str, interval: int, interruption: int) -> dict: command = [sys.executable, str(Path(__file__).resolve()), "--worker", "--state-dir", str(directory), "--phase", phase, "--items", str(args.items), "--batch-size", str(args.batch_size), "--rounds", str(args.rounds), "--checkpoint-interval", str(interval), "--interrupt-after-batch", str(interruption)] started = time.perf_counter() child = subprocess.run(command, capture_output=True, text=True, encoding="utf-8") if child.stdout: print(child.stdout, end="", flush=True) expected = EXIT_INTERRUPTED if interruption > 0 else 0 if child.returncode != expected: raise RuntimeError(f"worker exit {child.returncode}; expected {expected}: {child.stderr}") return {"phase": phase, "exit_code": child.returncode, "elapsed_seconds_including_process_startup": time.perf_counter() - started, "stdout": child.stdout.strip()} def scenario(args: argparse.Namespace, directory: Path, interval: int, interruption: int, reference: dict) -> dict: if directory.exists(): raise ValueError(f"scenario directory already exists; refusing to overwrite: {directory}") directory.mkdir(parents=True) first = child_call(args, directory, "initial", interval, interruption) restart = child_call(args, directory, "restart", interval, -1) final = json.loads((directory / "restart_result.json").read_text(encoding="utf-8")) rows = [json.loads(line) for line in (directory / "work_audit.jsonl").read_text( encoding="utf-8").splitlines()] counts: dict[int, int] = {} for row in rows: counts[row["index"]] = counts.get(row["index"], 0) + 1 redo_items = sum(count - 1 for count in counts.values()) checkpointed_items = ((interruption // interval) * interval * args.batch_size if interval else 0) expected_redo = interruption * args.batch_size - checkpointed_items exact_match = final["outputs"] == reference["outputs"] digest_match = final["final_digest"] == reference["final_digest"] passed = (exact_match and digest_match and final["completed_items"] == args.items and final["restored_items"] == checkpointed_items and redo_items == expected_redo and len(rows) == args.items + expected_redo) if not passed: raise AssertionError(f"verification failed for interval={interval}, interruption={interruption}") return {"interrupted_after_batch": interruption, "checkpoint_interval_batches": interval, "restored_items": final["restored_items"], "redo_items": redo_items, "total_counted_item_computations": len(rows), "unique_completed_items": len(counts), "expected_redo_items": expected_redo, "exact_item_results_match_reference": exact_match, "final_digest_matches_reference": digest_match, "final_digest": final["final_digest"], "verification_passed": passed, "process_runs": [first, restart]} def experiment(args: argparse.Namespace) -> int: base = checked_path(Path(args.state_dir) if args.state_dir else WORK_ROOT / "checkpoint_demo", WORK_ROOT, "state directory") run_directory = base / ("run_" + uuid.uuid4().hex) run_directory.mkdir(parents=True) output = checked_path(Path(args.output) if args.output else OUTPUT_ROOT / "xuhang" / "results" / "checkpoint_result.json", OUTPUT_ROOT, "result file") reference_dir = run_directory / "reference" child_call(args, reference_dir, "reference", 0, -1) reference = json.loads((reference_dir / "reference_result.json").read_text(encoding="utf-8")) cases = [] for interruption in args.interruption_positions: enabled = scenario(args, run_directory / f"batch_{interruption}_checkpoint", args.checkpoint_interval, interruption, reference) baseline = scenario(args, run_directory / f"batch_{interruption}_no_checkpoint", 0, interruption, reference) saved = baseline["total_counted_item_computations"] - enabled["total_counted_item_computations"] cases.append({"interrupted_after_batch": interruption, "with_checkpoint": enabled, "without_checkpoint": baseline, "saved_item_computations": saved, "computation_reduction_vs_no_checkpoint_percent": round( 100 * saved / baseline["total_counted_item_computations"], 6)}) print(f"verified batch {interruption}: checkpoint redo={enabled['redo_items']}; " f"no-checkpoint redo={baseline['redo_items']}; saved={saved}", flush=True) result = { "schema": SCHEMA, "workload": "synthetic text statistics and repeated SHA-256 hashing", "contains_user_files": False, "ai_model_used": False, "actual_worker_termination": "self-controlled subprocess os._exit(75)", "batch_size": args.batch_size, "items": args.items, "hash_rounds_per_item": args.rounds, "checkpoint_interval_batches": args.checkpoint_interval, "dataset_fingerprint": fingerprint(make_samples(args.items), args.rounds, args.batch_size), "reference_final_digest": reference["final_digest"], "all_verifications_passed": True, "state_directory": str(run_directory), "cases": cases, "measurement_scope": "Actual software computations counted from durable audit logs; time " "includes instrumentation and I/O. No power or battery energy measurement.", "limitations": [ "Demonstrates application batch boundary recovery, not LLM KV cache recovery.", "No physical power failure, UPS transition, battery hardware or sodium chemistry was tested.", "Synthetic deterministic workload and chosen interruption positions; not user research.", "Atomic file replacement does not establish hardware crash durability under real power loss.", "Audit instrumentation is part of the demonstration and changes elapsed-time overhead.", "Saved item computations are not claimed to equal energy, time or money saved.", ], } atomic_json(output, result) print(f"All verifications passed. Result: {output}", flush=True) return 0 def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--worker", action="store_true") parser.add_argument("--state-dir", help="strictly inside this project's work/") parser.add_argument("--output", help="strictly inside this project's outputs/") parser.add_argument("--phase", choices=("initial", "restart", "reference"), default="initial") parser.add_argument("--items", type=int, default=36) parser.add_argument("--batch-size", type=int, default=3) parser.add_argument("--rounds", type=int, default=3000) parser.add_argument("--checkpoint-interval", type=int, default=2) parser.add_argument("--interrupt-after-batch", type=int, default=-1) parser.add_argument("--interruption-positions", type=int, nargs="+", default=[1, 4, 5, 11]) args = parser.parse_args() if args.items < 1 or args.batch_size < 1 or args.items % args.batch_size: parser.error("items and batch-size must be positive; items must be divisible by batch-size") if args.rounds < 1 or args.rounds > 1_000_000: parser.error("rounds must be between 1 and 1000000") if args.checkpoint_interval < 0 or (not args.worker and args.checkpoint_interval == 0): parser.error("experiment checkpoint-interval must be positive") batches = args.items // args.batch_size if not args.worker and (not args.interruption_positions or any( position < 1 or position >= batches for position in args.interruption_positions)): parser.error("all interruption positions must fall before the final batch") if args.worker and args.interrupt_after_batch != -1 and not 1 <= args.interrupt_after_batch < batches: parser.error("worker interruption must be -1 or a batch before the final batch") return args if __name__ == "__main__": options = parse_args() try: sys.exit(run_worker(options) if options.worker else experiment(options)) except (ValueError, RuntimeError, AssertionError, OSError, KeyError) as error: print(f"ERROR: {error}", file=sys.stderr) sys.exit(1)