"""Deterministic teaching traces, not a reproduction of NSDI 2024 benchmarks. SPDX-License-Identifier: MIT Copyright (c) 2026 Hesela """ from datetime import datetime, timezone import argparse import hashlib import importlib.metadata import itertools import json import platform from pathlib import Path import libcachesim as lcs VERSION = "0.3.5" POLICIES = {"LRU": lcs.LRU, "FIFO": lcs.FIFO, "CLOCK": lcs.Clock, "SIEVE": lcs.Sieve} CASES = [ ("sieve-wins-lru", "AABCDA", 3), ("sieve-loses-lru", "AABCDB", 3), ("sieve-differs-clock", "AABCCDBA", 3), ("hot-pair", "ABABABAB", 3), ("cold-scan", "ABCDEFGH", 3), ("four-object-loop", "ABCDABCDABCD", 3), ("warm-scan-return", "ABCABCDEFABCABC", 3), ("all-visited", "ABCABCD", 3), ] def require(condition, message): """Keep publication checks active even under python -O.""" if not condition: raise AssertionError(message) def validate(refs, capacity): if type(capacity) is not int or capacity < 1: raise ValueError("capacity must be a positive integer") if not isinstance(refs, str) or any(c not in "ABCDEFGHIJKLMNOPQRSTUVWXYZ" for c in refs): raise ValueError("references must be uppercase ASCII object IDs") def replay(refs, capacity, policy): validate(refs, capacity) if policy not in POLICIES: raise ValueError("unknown policy") options = {"cache_size": capacity, "hashpower": 4, "consider_obj_metadata": False} if policy == "CLOCK": options.update(init_freq=0, n_bit_counter=1) cache = POLICIES[policy](**options) outcomes = [ bool(cache.get(lcs.Request(obj_id=ord(c), obj_size=1, clock_time=0))) for c in refs ] require(cache.get_n_obj() <= capacity, "cache exceeded capacity") return {"hits": sum(outcomes), "misses": len(refs) - sum(outcomes), "outcomes": "".join("H" if hit else "M" for hit in outcomes)} def sieve_oracle(refs, capacity): """List-based Algorithm 1 oracle; used for semantics, never for timings.""" validate(refs, capacity) order, visited, hand, outcomes, probes = [], {}, None, [], [] for key in refs: hit = key in visited outcomes.append(hit) inspected = 0 if hit: visited[key] = True else: if len(order) == capacity: candidate = order[0] if hand is None else hand while True: inspected += 1 idx = order.index(candidate) following = order[idx + 1] if idx + 1 < len(order) else None if not visited[candidate]: hand = following order.remove(candidate) del visited[candidate] break visited[candidate] = False candidate = order[0] if following is None else following order.append(key) visited[key] = False probes.append(inspected) return {"outcomes": "".join("H" if hit else "M" for hit in outcomes), "evictionProbes": probes} def check(): require(importlib.metadata.version("libcachesim") == VERSION, "libcachesim version mismatch; install " + VERSION) require(replay("ABAB", 2, "LRU")["outcomes"] == "MMHH", "LRU baseline") require(replay("ABCDABCD", 3, "LRU")["misses"] == 8, "LRU loop") require(replay("ABACAB", 2, "FIFO")["outcomes"] == "MMHMMM", "FIFO baseline") count = 0 suites = [("ABC", 6, (1, 2, 3)), ("ABCD", 8, (3,))] for alphabet, length, capacities in suites: for capacity in capacities: for sequence in itertools.product(alphabet, repeat=length): refs = "".join(sequence) actual = replay(refs, capacity, "SIEVE") expected = sieve_oracle(refs, capacity) require(actual["outcomes"] == expected["outcomes"], ("SIEVE differential mismatch", refs, capacity)) count += 1 for _, refs, capacity in CASES: require(replay(refs, capacity, "SIEVE")["outcomes"] == sieve_oracle(refs, capacity)["outcomes"], ("published row", refs)) require(sieve_oracle("ABCABCD", 3)["evictionProbes"][-1] == 4, "full scan") for bad in (0, -1, 1.5, True, "3"): try: replay("A", bad, "SIEVE") except ValueError: pass else: raise AssertionError("invalid capacity accepted") for bad in (None, [], 5, "a", "A B", "A0"): try: replay(bad, 3, "SIEVE") except ValueError: pass else: raise AssertionError("invalid references accepted") try: replay("A", 3, "unknown") except ValueError: pass else: raise AssertionError("invalid policy accepted") return count def result(): checks = check() rows = [] for name, refs, capacity in CASES: rows.append({"name": name, "references": refs, "capacityObjects": capacity, "requests": len(refs), "policies": {p: replay(refs, capacity, p) for p in POLICIES}, "oracle": sieve_oracle(refs, capacity)}) return { "schemaVersion": 1, "evidenceType": "constructed-trace replay", "generatedAt": datetime.now(timezone.utc).isoformat(), "license": "MIT (Hesela script and constructed results only)", "differentialSuites": [ {"alphabet": "ABC", "length": 6, "capacities": [1, 2, 3], "cases": 2187}, {"alphabet": "ABCD", "length": 8, "capacities": [3], "cases": 65536} ], "library": {"name": "libcachesim", "version": VERSION, "url": "https://pypi.org/project/libcachesim/0.3.5/", "wheelFilename": "libcachesim-0.3.5-cp312-cp312-manylinux_2_34_x86_64.whl", "wheelSha256": "c51bec8f990bc52a2f21bd6667591d94170b1c87cc22e662ae1538ce2327d5fc"}, "pythonVersion": platform.python_version(), "scriptSha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "differentialCases": checks, "seed": None, "assumptions": [ "Empty cache at the beginning of each row; count all cold misses.", "Equal-size objects represented as one byte, so integer byte capacity equals object slots.", "Every miss is admitted, no metadata charged to capacity, no expiration or writes.", "CLOCK uses init_freq=0 and n_bit_counter=1; a different variant can change results.", "One sequential requester, instant fill; no device I/O, backend queueing or throughput timing.", "Traces are constructed teaching examples, not randomly sampled production observations.", "Oracle eviction probes are algorithmic inspections, not library CPU-operation measurements." ], "rows": rows } if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--output", type=Path) parser.add_argument("--discover", action="store_true") args = parser.parse_args() if args.discover: found = {} for n in range(4, 10): for seq in itertools.product("ABCD", repeat=n): refs = "".join(seq) if len(set(refs)) < 4: continue s = replay(refs, 3, "SIEVE")["hits"] l = replay(refs, 3, "LRU")["hits"] c = replay(refs, 3, "CLOCK")["hits"] for label, matches in (("sieve-wins-lru", s > l), ("sieve-loses-lru", s < l), ("sieve-differs-clock", s != c)): if matches and label not in found: found[label] = {"references": refs, "SIEVE": s, "LRU": l, "CLOCK": c} if len(found) == 3: break print(json.dumps(found, indent=2)) else: data = json.dumps(result(), indent=2) + "\n" if args.output: args.output.write_text(data) else: print(data, end="")