diff options
Diffstat (limited to 'research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources')
3 files changed, 634 insertions, 0 deletions
diff --git a/research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources/benchmark.c b/research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources/benchmark.c new file mode 100644 index 0000000..496ad5e --- /dev/null +++ b/research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources/benchmark.c @@ -0,0 +1,320 @@ +#define _POSIX_C_SOURCE 200809L +#include <assert.h> +#include <errno.h> +#include <inttypes.h> +#include <limits.h> +#include <signal.h> +#include <stdbool.h> +#include <stdint.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/resource.h> +#include <time.h> +#include <unistd.h> + +/* A measurement apparatus, not the Pure runtime or a general document parser. */ +enum { CACHE_SLOTS = 16, SCRATCH = 65536, HIST_SIZE = 512, MAX_FRAME = 8*1024*1024 }; +typedef struct Block { size_t capacity; uint64_t since; unsigned char bytes[]; } Block; +typedef struct Item { + Block *block; + size_t offset, length; + uint64_t start, ready; +} Item; +typedef struct Engine { + Item *queue; + Block *cache[CACHE_SLOTS]; + unsigned char scratch[SCRATCH]; + uint64_t histogram[HIST_SIZE]; + size_t slots, head, count, fixed, budget, reserved, active, cached; + size_t peak_reserved, peak_active, peak_queue; + uint64_t allocations, reuses, copies, read_bytes, written_bytes; + uint64_t accepted, emitted, copied_records, borrowed_records, pressure_copies; + uint64_t budget_waits, slot_waits, sleep_ns, hold_ns, service_ns, last_emit; + uint64_t elapsed_ns, cpu_ns, latency_sum, latency_max, start, cancel_after; + long double active_byte_ns; + unsigned ratio; + bool pool; + const char *policy, *status; +} Engine; +static volatile sig_atomic_t interrupted; +static void stop(int sig) { (void)sig; interrupted = 1; } +static uint64_t ns(clockid_t clock) { + struct timespec t; + if (clock_gettime(clock, &t)) { perror("clock_gettime"); exit(2); } + return (uint64_t)t.tv_sec * UINT64_C(1000000000) + (uint64_t)t.tv_nsec; +} +static size_t charge(const Block *b) { return sizeof(*b) + b->capacity; } +static size_t rounded(size_t n) { + size_t r = 64; + while (r < n) r *= 2; + return r; +} +static void peaks(Engine *e) { + assert(e->active + e->cached == e->reserved); + assert(e->fixed + e->reserved <= e->budget); + if (e->reserved > e->peak_reserved) e->peak_reserved = e->reserved; + if (e->active > e->peak_active) e->peak_active = e->active; +} +static void evict(Engine *e, size_t i) { + Block *b = e->cache[i]; + if (!b) return; + e->cached -= charge(b); e->reserved -= charge(b); + free(b); e->cache[i] = NULL; +} +static Block *take(Engine *e, size_t n) { + size_t capacity = rounded(n), need = sizeof(Block) + capacity; + if (e->pool) { + for (size_t i = 0; i < CACHE_SLOTS; ++i) { + Block *b = e->cache[i]; + if (b && b->capacity == capacity) { + e->cache[i] = NULL; e->cached -= need; e->active += need; + ++e->reuses; b->since = ns(CLOCK_MONOTONIC); peaks(e); return b; + } + } + } + for (size_t i = 0; i < CACHE_SLOTS && need > e->budget-e->fixed-e->reserved; ++i) + evict(e, i); + if (need > e->budget-e->fixed-e->reserved) return NULL; + Block *b = malloc(need); + if (!b) { e->status = "allocation_failed"; return NULL; } + b->capacity = capacity; b->since = ns(CLOCK_MONOTONIC); + e->reserved += need; e->active += need; ++e->allocations; peaks(e); return b; +} +static void give(Engine *e, Block *b) { + if (!b) return; + size_t bytes = charge(b); + e->active_byte_ns += (long double)bytes * (ns(CLOCK_MONOTONIC)-b->since); + e->active -= bytes; + if (e->pool && bytes <= e->budget/4-e->cached) { + for (size_t i = 0; i < CACHE_SLOTS; ++i) if (!e->cache[i]) { + e->cache[i] = b; e->cached += bytes; peaks(e); return; + } + } + e->reserved -= bytes; free(b); peaks(e); +} +static bool sleep_until(Engine *e, uint64_t deadline) { + uint64_t t = ns(CLOCK_MONOTONIC), start = t; + while (t < deadline && !interrupted) { + uint64_t delta = deadline-t; + struct timespec request = {(time_t)(delta/1000000000), (long)(delta%1000000000)}; + if (nanosleep(&request, NULL) && errno != EINTR) { e->status="sleep_failed"; return false; } + t = ns(CLOCK_MONOTONIC); + } + e->sleep_ns += t-start; + return !interrupted; +} +/* floor(log2(ns)) with eight subdivisions: upper bound error <= 12.5%. */ +static unsigned bucket(uint64_t n) { + unsigned exponent = 0; + for (uint64_t x = n; x > 1; x >>= 1) ++exponent; + if (exponent < 3) return (unsigned)n; + return exponent*8 + (unsigned)((n-(UINT64_C(1)<<exponent))>>(exponent-3)); +} +static uint64_t upper(unsigned b) { + if (b < 24) return b+1; + unsigned exponent = b/8; + if (exponent >= 63) return UINT64_MAX; + return (UINT64_C(1)<<exponent) + (uint64_t)(b%8+1)*(UINT64_C(1)<<(exponent-3)); +} +static bool write_all(Engine *e, const void *src, size_t n) { + const unsigned char *p = src; + while (n && !interrupted) { + ssize_t written = write(STDOUT_FILENO, p, n); + if (written < 0 && errno == EINTR) continue; + if (written <= 0) { e->status = "output_failed"; return false; } + p += written; n -= (size_t)written; e->written_bytes += (uint64_t)written; + } + return !interrupted; +} +static bool emit(Engine *e) { + assert(e->count); + Item *item = &e->queue[e->head]; + uint64_t ready = item->ready; + if (e->last_emit + e->service_ns > ready) ready = e->last_emit + e->service_ns; + if (!sleep_until(e, ready)) return false; + uint32_t len = (uint32_t)item->length; + unsigned char header[4] = {(unsigned char)len, (unsigned char)(len>>8), + (unsigned char)(len>>16), (unsigned char)(len>>24)}; + if (!write_all(e, header, sizeof header) || + !write_all(e, item->block->bytes+item->offset, item->length)) return false; + uint64_t now = ns(CLOCK_MONOTONIC), latency = now-item->start; + ++e->histogram[bucket(latency)]; e->latency_sum += latency; + if (latency > e->latency_max) e->latency_max = latency; + e->last_emit = now; ++e->emitted; + give(e, item->block); item->block = NULL; + e->head = (e->head+1)%e->slots; --e->count; + if (e->cancel_after && e->emitted >= e->cancel_after) { + e->status = "cancelled"; return false; + } + return true; +} +static Block *acquire(Engine *e, size_t n) { + for (;;) { + Block *b = take(e, n); + if (b) return b; + if (strcmp(e->status, "ok")) return NULL; + if (!e->count) { e->status = "budget_too_small"; return NULL; } + ++e->budget_waits; + if (!emit(e)) return NULL; + } +} +/* Returns 1 on complete read, 0 on clean EOF before any byte, -1 on failure. */ +static int read_exact(Engine *e, void *dst, size_t n) { + unsigned char *p = dst; size_t done = 0; + while (done < n && !interrupted) { + ssize_t got = read(STDIN_FILENO, p+done, n-done); + if (got < 0 && errno == EINTR) continue; + if (got < 0) { e->status="input_failed"; return -1; } + if (!got) { if (!done) return 0; e->status="truncated_input"; return -1; } + done += (size_t)got; e->read_bytes += (uint64_t)got; + } + return interrupted ? -1 : 1; +} +static bool discard(Engine *e, size_t n) { + while (n) { + size_t chunk = n < SCRATCH ? n : SCRATCH; + if (read_exact(e, e->scratch, chunk) != 1) { + if (!strcmp(e->status,"ok")) e->status="truncated_input"; + return false; + } + n -= chunk; + } + return true; +} +static uint32_t le32(const unsigned char *p) { + return (uint32_t)p[0] | (uint32_t)p[1]<<8 | (uint32_t)p[2]<<16 | (uint32_t)p[3]<<24; +} +static void run(Engine *e) { + for (;;) { + if (interrupted) break; + if (e->count) { + Item *head = &e->queue[e->head]; + uint64_t now = ns(CLOCK_MONOTONIC); + if (now >= head->ready && now >= e->last_emit+e->service_ns) { + if (!emit(e)) break; + continue; + } + } + if (e->count == e->slots) { ++e->slot_waits; if (!emit(e)) break; } + unsigned char header[12]; + uint64_t start = ns(CLOCK_MONOTONIC); + int got = read_exact(e, header, sizeof header); + if (got != 1) { + if (!got) while (e->count && !interrupted) if (!emit(e)) break; + break; + } + size_t payload = le32(header), offset = le32(header+4), length = le32(header+8); + if (!payload || payload > MAX_FRAME || !length || offset > payload || length > payload-offset) { + e->status="invalid_frame"; break; + } + bool project = !strcmp(e->policy,"project"); + Block *source = acquire(e, project ? length : payload); + if (!source) break; + bool valid; + if (project) { + valid = discard(e, offset) && read_exact(e, source->bytes, length)==1 && + discard(e, payload-offset-length); + offset = 0; + } else valid = read_exact(e, source->bytes, payload)==1; + if (!valid) { + if (!strcmp(e->status,"ok")) e->status="truncated_input"; + give(e, source); break; + } + bool copy = !strcmp(e->policy,"copy") || + (!strcmp(e->policy,"threshold") && rounded(payload)/rounded(length) >= e->ratio); + if (!strcmp(e->policy,"pressure")) { + /* Online rule using current charged live buffers; no future lifetime. */ + copy = rounded(length) < source->capacity && + e->active > (e->budget-e->fixed)/2; + if (copy) ++e->pressure_copies; + } + if (!project && copy) { + Block *small = acquire(e, length); + if (!small) { give(e, source); break; } + memcpy(small->bytes, source->bytes+offset, length); + e->copies += length; ++e->copied_records; + give(e, source); source = small; offset = 0; + } else if (!project) ++e->borrowed_records; + size_t tail = (e->head+e->count)%e->slots; + e->queue[tail] = (Item){source, offset, length, start, ns(CLOCK_MONOTONIC)+e->hold_ns}; + ++e->count; ++e->accepted; + if (e->count > e->peak_queue) e->peak_queue=e->count; + } + if (interrupted) e->status="cancelled"; +} +static uint64_t number(const char *s, uint64_t maximum) { + char *end; errno=0; + unsigned long long n = strtoull(s, &end, 10); + if (!*s || *s=='-' || *end || errno || n>maximum) { fprintf(stderr,"invalid number\n"); exit(2); } + return n; +} +int main(int argc, char **argv) { + Engine *e = calloc(1, sizeof *e); + if (!e) return 2; + e->policy="borrow"; e->status="ok"; e->budget=8*1024*1024; e->slots=512; + e->ratio=4; e->pool=true; + for (int i=1; i<argc; i+=2) { + if (i+1==argc) { fprintf(stderr,"options require values\n"); free(e); return 2; } + const char *key=argv[i], *value=argv[i+1]; + if (!strcmp(key,"--policy")) e->policy=value; + else if (!strcmp(key,"--budget")) e->budget=(size_t)number(value,1073741824); + else if (!strcmp(key,"--slots")) e->slots=(size_t)number(value,4096); + else if (!strcmp(key,"--hold-us")) e->hold_ns=number(value,10000000)*1000; + else if (!strcmp(key,"--service-us")) e->service_ns=number(value,10000000)*1000; + else if (!strcmp(key,"--ratio")) e->ratio=(unsigned)number(value,UINT_MAX); + else if (!strcmp(key,"--pool")) e->pool=number(value,1)!=0; + else if (!strcmp(key,"--cancel-after")) e->cancel_after=number(value,UINT64_MAX); + else { fprintf(stderr,"unknown option\n"); free(e); return 2; } + } + bool known = !strcmp(e->policy,"borrow") || !strcmp(e->policy,"copy") || + !strcmp(e->policy,"threshold") || !strcmp(e->policy,"pressure") || !strcmp(e->policy,"project"); + if (!known || !e->slots || e->slots>4096 || e->ratio<2 || e->budget>UINT64_C(1073741824) || + e->hold_ns>UINT64_C(10000000000) || e->service_ns>UINT64_C(10000000000)) { + fprintf(stderr,"invalid configuration\n"); free(e); return 2; + } + e->fixed=sizeof(*e)+e->slots*sizeof(Item); + if (e->budget < e->fixed) { fprintf(stderr,"budget below fixed metadata\n"); free(e); return 2; } + e->queue=calloc(e->slots,sizeof(Item)); + if (!e->queue) { free(e); return 2; } + signal(SIGINT,stop); signal(SIGTERM,stop); signal(SIGPIPE,SIG_IGN); + e->start=ns(CLOCK_MONOTONIC); uint64_t cpu=ns(CLOCK_PROCESS_CPUTIME_ID); + run(e); + size_t live_at_stop=e->active, cached_at_stop=e->cached; + while (e->count) { + give(e,e->queue[e->head].block); e->head=(e->head+1)%e->slots; --e->count; + } + for (size_t i=0;i<CACHE_SLOTS;++i) evict(e,i); + assert(!e->active && !e->reserved && !e->cached); + e->elapsed_ns=ns(CLOCK_MONOTONIC)-e->start; e->cpu_ns=ns(CLOCK_PROCESS_CPUTIME_ID)-cpu; + uint64_t target=e->emitted-e->emitted/100, sum=0, p99=0; + for (unsigned b=0; e->emitted && b<HIST_SIZE; ++b) { + sum+=e->histogram[b]; if (sum>=target) { p99=upper(b); break; } + } + struct rusage usage; long rss=0; + if (!getrusage(RUSAGE_SELF,&usage)) rss=usage.ru_maxrss; +#if defined(__APPLE__) + uint64_t rss_bytes=(uint64_t)rss; +#else + uint64_t rss_bytes=(uint64_t)rss*1024; +#endif + fprintf(stderr,"{\"status\":\"%s\",\"policy\":\"%s\",\"pool\":%s," + "\"budget_bytes\":%zu,\"fixed_bytes\":%zu,\"peak_reserved_bytes\":%zu," + "\"peak_active_bytes\":%zu,\"peak_queue\":%zu,\"rss_peak_bytes\":%" PRIu64 "," + "\"elapsed_ns\":%" PRIu64 ",\"cpu_ns\":%" PRIu64 ",\"input_bytes\":%" PRIu64 "," + "\"output_bytes\":%" PRIu64 ",\"accepted\":%" PRIu64 ",\"emitted\":%" PRIu64 "," + "\"copy_bytes\":%" PRIu64 ",\"copied_records\":%" PRIu64 ",\"borrowed_records\":%" PRIu64 "," + "\"pressure_copies\":%" PRIu64 ",\"allocations\":%" PRIu64 ",\"reuses\":%" PRIu64 "," + "\"budget_waits\":%" PRIu64 ",\"slot_waits\":%" PRIu64 ",\"sleep_ns\":%" PRIu64 "," + "\"latency_mean_ns\":%.3f,\"latency_p99_upper_ns\":%" PRIu64 ",\"latency_max_ns\":%" PRIu64 "," + "\"active_byte_seconds\":%.6Lf,\"live_at_stop_bytes\":%zu,\"cached_at_stop_bytes\":%zu," + "\"unreleased_buffer_bytes\":%zu}\n", + e->status,e->policy,e->pool?"true":"false",e->budget,e->fixed,e->fixed+e->peak_reserved, + e->fixed+e->peak_active,e->peak_queue,rss_bytes,e->elapsed_ns,e->cpu_ns,e->read_bytes, + e->written_bytes,e->accepted,e->emitted,e->copies,e->copied_records,e->borrowed_records, + e->pressure_copies,e->allocations,e->reuses,e->budget_waits,e->slot_waits,e->sleep_ns, + e->emitted?(double)e->latency_sum/(double)e->emitted:0,p99,e->latency_max, + e->active_byte_ns/1e9L,live_at_stop,cached_at_stop,e->reserved); + int result=!strcmp(e->status,"ok")?0:!strcmp(e->status,"cancelled")?3:2; + free(e->queue); free(e); return result; +} diff --git a/research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources/plan.json b/research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources/plan.json new file mode 100644 index 0000000..5040006 --- /dev/null +++ b/research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources/plan.json @@ -0,0 +1,33 @@ +{ + "version": 1, + "written_before_measurement": true, + "seed": 20260922, + "repeats": 5, + "slots": 512, + "policies": [ + {"name": "borrow", "policy": "borrow", "pool": 1}, + {"name": "copy", "policy": "copy", "pool": 1}, + {"name": "threshold2", "policy": "threshold", "ratio": 2, "pool": 1}, + {"name": "threshold4", "policy": "threshold", "ratio": 4, "pool": 1}, + {"name": "threshold8", "policy": "threshold", "ratio": 8, "pool": 1}, + {"name": "pressure", "policy": "pressure", "pool": 1}, + {"name": "project", "policy": "project", "pool": 1}, + {"name": "copy_no_pool", "policy": "copy", "pool": 0} + ], + "cases": [ + {"name": "sparse_slow", "shape": "sparse", "records": 2048, "budget": 8388608, "hold_us": 5000, "service_us": 20}, + {"name": "dense_slow", "shape": "dense", "records": 2048, "budget": 8388608, "hold_us": 5000, "service_us": 20}, + {"name": "mixed_slow", "shape": "mixed", "records": 2048, "budget": 8388608, "hold_us": 5000, "service_us": 20}, + {"name": "mixed_tight", "shape": "mixed", "records": 2048, "budget": 2097152, "hold_us": 5000, "service_us": 20}, + {"name": "sparse_fast", "shape": "sparse", "records": 2048, "budget": 8388608, "hold_us": 0, "service_us": 0} + ], + "expectations": [ + "Borrow retains full source allocations for small selections.", + "Copy can add CPU and transient source plus destination memory for dense selections.", + "A current-pressure rule need not outperform a static size rule or early projection.", + "Pool reuse can reduce allocation calls while increasing reserved idle memory.", + "No Pure-specific advantage, application SLO, or statistical significance is presumed." + ], + "comparison_rule": "Report all presets and repetitions; no post-hoc winner is called an out-of-sample tuned policy.", + "limits": "Single FIFO local pipeline with synthetic delayed sink. No Sonic/Node/.NET benchmark, fan-out, persistent cache, generic JSON parser, external open-loop arrivals, or production SLO." +} diff --git a/research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources/run.py b/research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources/run.py new file mode 100644 index 0000000..94657cc --- /dev/null +++ b/research/experiments/002-buffer-retention/results/2026-09-22-linux-aarch64/sources/run.py @@ -0,0 +1,281 @@ +#!/usr/bin/env python3 +"""Build, independently verify, and measure the retention experiment.""" +import argparse +import csv +import hashlib +import json +import os +from pathlib import Path +import platform +import random +import shlex +import signal +import statistics +import struct +import subprocess +import tempfile +import time + +ROOT = Path(__file__).resolve().parent +FLAGS = ["-std=c17", "-Wall", "-Wextra", "-Wpedantic", "-Werror", "-O2", "-g"] + + +def digest(path): + with open(path, "rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def dump(path, value): + path.write_text(json.dumps(value, indent=2, ensure_ascii=False) + "\n") + + +def build(cc, output, sanitizer=False): + command = [*cc, *FLAGS] + if sanitizer: + command += ["-fsanitize=address,undefined", "-fno-omit-frame-pointer"] + command += [str(ROOT / "benchmark.c"), "-o", str(output)] + result = subprocess.run(command, capture_output=True, text=True, check=True) + return {"command": command, "stderr": result.stderr, "sha256": digest(output), + "size_bytes": output.stat().st_size} + + +def shape_for(shape, rng, i): + if shape == "sparse": + return 65536, 64 + if shape == "dense": + return 65536, 49152 + if shape == "mixed": + source = rng.choice([4096, 16384, 65536, 262144]) + keep = max(16, source // rng.choice([1, 2, 4, 8, 64, 1024])) + return source, keep + if shape == "test": + return [(1, 1), (63, 17), (4097, 12), (65536, 49152), (127, 127)][i % 5] + raise ValueError(shape) + + +def fixture(path, shape, count, seed): + rng = random.Random(seed) + expected = hashlib.sha256() + total = 0 + with path.open("wb") as stream: + for i in range(count): + size, keep = shape_for(shape, rng, i) + offset = rng.randrange(size - keep + 1) + # All input bytes exist on disk; no sparse files or seeking past data. + tile = rng.randbytes(256) + payload = (tile * ((size + 255) // 256))[:size] + stream.write(struct.pack("<III", size, offset, keep)) + stream.write(payload) + out = struct.pack("<I", keep) + payload[offset:offset + keep] + expected.update(out) + total += len(out) + return {"path": path.name, "shape": shape, "records": count, "seed": seed, + "input_bytes": path.stat().st_size, "input_sha256": digest(path), + "expected_output_bytes": total, "expected_output_sha256": expected.hexdigest()} + + +def reference(data): + result = bytearray() + at = 0 + while at < len(data): + if len(data) - at < 12: + raise ValueError("short header") + size, offset, keep = struct.unpack_from("<III", data, at) + at += 12 + if not 0 < size <= 8 * 1024 * 1024 or not 0 < keep <= size or offset > size - keep: + raise ValueError("invalid frame") + if len(data) - at < size: + raise ValueError("short payload") + result.extend(struct.pack("<I", keep)) + result.extend(data[at + offset:at + offset + keep]) + at += size + return bytes(result) + + +def command(binary, case, policy, *extra): + args = [str(binary), "--policy", policy["policy"], "--pool", str(policy.get("pool", 1)), + "--ratio", str(policy.get("ratio", 4)), "--budget", str(case.get("budget", 8388608)), + "--slots", str(case.get("slots", 512)), "--hold-us", str(case.get("hold_us", 0)), + "--service-us", str(case.get("service_us", 0))] + return args + list(extra) + + +def metrics(stderr): + # Sanitizer diagnostics must not be silently ignored. + text = stderr.decode().strip() + try: + result = json.loads(text) + except json.JSONDecodeError as exc: + raise AssertionError(f"Expected metrics only; tool/runtime diagnostics:\n{text}") from exc + assert result["unreleased_buffer_bytes"] == 0, result + assert result["peak_active_bytes"] <= result["peak_reserved_bytes"] <= result["budget_bytes"], result + return result + + +def check(binary, tmp, policies): + checks = [] + test = tmp / "test.frames" + fixture(test, "test", 45, 177) + data = test.read_bytes() + expected = reference(data) + for policy in policies: + for slow in [False, True]: + case = {"slots": 7, "hold_us": 100 if slow else 0, "budget": 300000} + result = subprocess.run(command(binary, case, policy), input=data, capture_output=True, timeout=15) + m = metrics(result.stderr) + assert result.returncode == 0 and m["status"] == "ok", (policy, result.stderr) + assert result.stdout == expected and m["emitted"] == 45 + assert m["input_bytes"] == len(data) and m["output_bytes"] == len(expected) + assert m["live_at_stop_bytes"] == 0 + checks.append(f'{policy["name"]}: byte equality, accounting, cleanup, slow={slow}') + # Truncation at both header and payload boundaries, including exact missing payload. + bad_inputs = [b"x", struct.pack("<III", 12, 0, 4), + struct.pack("<III", 12, 0, 4) + b"123", + struct.pack("<III", 12, 11, 4) + b"x" * 12, + struct.pack("<III", 0, 0, 0), + struct.pack("<III", 8 * 1024 * 1024 + 1, 0, 1)] + for bad in bad_inputs: + result = subprocess.run(command(binary, {"hold_us": 1000}, policy), + input=data[:13] + bad, capture_output=True, timeout=15) + m = metrics(result.stderr) + assert result.returncode == 2 and m["status"] in {"invalid_frame", "truncated_input"} + checks.append(f'{policy["name"]}: six malformed inputs with pending buffer cleanup') + result = subprocess.run(command(binary, {"hold_us": 1000}, policy, "--cancel-after", "3"), + input=data, capture_output=True, timeout=15) + m = metrics(result.stderr) + prefix_end = 0 + for _ in range(3): + prefix_end += 4 + struct.unpack_from("<I", expected, prefix_end)[0] + assert result.returncode == 3 and m["status"] == "cancelled" and m["emitted"] == 3 + assert result.stdout == expected[:prefix_end] + checks.append(f'{policy["name"]}: cancellation preserves completed output prefix and releases pending buffers') + result = subprocess.run(command(binary, {}, policy), input=b"", capture_output=True, timeout=15) + assert result.returncode == 0 and metrics(result.stderr)["emitted"] == 0 + checks.append(f'{policy["name"]}: empty stream') + # Tight budget must refuse a whole-frame requirement, while streaming projection can work. + large = struct.pack("<III", 1048576, 17, 64) + b"a" * 1048576 + for policy in policies: + result = subprocess.run(command(binary, {"budget": 131072, "slots": 1}, policy), + input=large, capture_output=True, timeout=15) + m = metrics(result.stderr) + assert m["status"] == ("ok" if policy["policy"] == "project" else "budget_too_small") + checks.append("frame larger than buffer budget: bounded refusal versus projection") + if Path("/dev/full").exists(): + with open("/dev/full", "wb") as sink: + result = subprocess.run(command(binary, {"hold_us": 1000}, policies[0]), input=data, + stdout=sink, stderr=subprocess.PIPE, timeout=15) + assert result.returncode == 2 and metrics(result.stderr)["status"] == "output_failed" + checks.append("real output failure through /dev/full, pending buffers released") + # Real signal cancellation after allocation, while waiting for the delayed sink. + with test.open("rb") as source, open(os.devnull, "wb") as sink: + proc = subprocess.Popen(command(binary, {"hold_us": 500000}, policies[0]), + stdin=source, stdout=sink, stderr=subprocess.PIPE) + time.sleep(0.08) + proc.send_signal(signal.SIGTERM) + _, error = proc.communicate(timeout=5) + assert proc.returncode == 3 and metrics(error)["status"] == "cancelled", error + checks.append("SIGTERM interrupts delayed sink and releases buffers") + return checks + + +def sample(binary, source, output, case, policy): + with source.open("rb") as inp, output.open("wb") as out: + result = subprocess.run(command(binary, case, policy), stdin=inp, stdout=out, + stderr=subprocess.PIPE, timeout=30) + m = metrics(result.stderr) + assert result.returncode == 0 and m["status"] == "ok", result.stderr + return m + + +def summarize(rows): + result = [] + names = ["elapsed_ns", "cpu_ns", "peak_reserved_bytes", "peak_active_bytes", "rss_peak_bytes", + "copy_bytes", "active_byte_seconds", "latency_mean_ns", "latency_p99_upper_ns", + "budget_waits", "allocations", "reuses"] + groups = sorted({(r["case"], r["variant"]) for r in rows}) + for case, variant in groups: + group = [r for r in rows if (r["case"], r["variant"]) == (case, variant)] + value = {"case": case, "variant": variant, "samples": len(group)} + for name in names: + values = [r[name] for r in group] + value[name] = {"median": statistics.median(values), "min": min(values), "max": max(values)} + result.append(value) + return result + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--check-only", action="store_true") + parser.add_argument("--output", type=Path) + args = parser.parse_args() + if not args.check_only and args.output is None: + parser.error("provide --check-only or --output (new directory)") + if args.output: + args.output.mkdir(parents=True, exist_ok=False) + cc = shlex.split(os.environ.get("CC", "cc")) + plan = json.loads((ROOT / "plan.json").read_text()) + affinity = None + if hasattr(os, "sched_getaffinity"): + affinity = min(os.sched_getaffinity(0)) + os.sched_setaffinity(0, {affinity}) + with tempfile.TemporaryDirectory(prefix="pure-retention-") as folder: + tmp = Path(folder) + native, sanitized = tmp / "bench", tmp / "bench-sanitized" + builds = [build(cc, native), build(cc, sanitized, True)] + verification = {"native": check(native, tmp, plan["policies"]), + "asan_ubsan": check(sanitized, tmp, plan["policies"])} + print(f'Verified {sum(map(len, verification.values()))} scenario groups (native + ASan/UBSan).', flush=True) + if args.check_only: + return + sources = {name: digest(ROOT / name) for name in ["benchmark.c", "run.py", "plan.json"]} + environment = {"platform": platform.platform(), "machine": platform.machine(), + "python": platform.python_version(), "cpu_affinity": affinity, + "compiler": subprocess.check_output([*cc, "--version"], text=True), + "builds": builds, "source_sha256": sources, + "clock": "CLOCK_MONOTONIC; CPU = CLOCK_PROCESS_CPUTIME_ID", + "timing": "includes file IO, scheduling, deliberate sleep, metrics, buffer cleanup; excludes startup/build/fixture/verification", + "limitations": "shared host; unregulated thermals; cached files; output page cache, no fsync; synthetic closed-loop arrivals", + "started_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} + if Path("/proc/cpuinfo").exists(): + environment["cpuinfo"] = Path("/proc/cpuinfo").read_text() + dump(args.output / "environment.json", environment) + dump(args.output / "verification.json", verification) + dump(args.output / "plan.json", plan) + fixtures, rows = [], [] + rng = random.Random(plan["seed"]) + for case in plan["cases"]: + source, output = tmp / "input.frames", tmp / "output.frames" + info = fixture(source, case["shape"], case["records"], plan["seed"]) + info["case"] = case["name"] + fixtures.append(info) + # Warm each strategy once. The following measurements use fresh processes. + for policy in plan["policies"]: + sample(native, source, output, case, policy) + assert digest(output) == info["expected_output_sha256"] + for repeat in range(plan["repeats"]): + order = list(plan["policies"]) + rng.shuffle(order) + for policy in order: + m = sample(native, source, output, case, policy) + assert m["accepted"] == m["emitted"] == case["records"] + assert m["input_bytes"] == info["input_bytes"] + assert m["output_bytes"] == info["expected_output_bytes"] + assert digest(output) == info["expected_output_sha256"] + rows.append({"case": case["name"], "variant": policy["name"], "repeat": repeat, **m}) + dump(args.output / "raw.json", rows) + dump(args.output / "fixtures.json", fixtures) + print(f'Completed {case["name"]}: {len(plan["policies"])} variants x {plan["repeats"]} trials.', flush=True) + dump(args.output / "summary.json", summarize(rows)) + with (args.output / "raw.csv").open("w") as out: + writer = csv.DictWriter(out, fieldnames=list(rows[0])) + writer.writeheader() + writer.writerows(rows) + # Detect accidental source edits during a measurement run. + assert sources == {name: digest(ROOT / name) for name in sources} + dump(args.output / "complete.json", {"samples": len(rows), "all_output_hashes_match": True, + "finished_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}) + print(args.output.resolve()) + + +if __name__ == "__main__": + main() |