1 files changed, 119 insertions, 0 deletions
diff --git a/research/experiments/046-commit-replica/prefixes.py b/research/experiments/046-commit-replica/prefixes.py
new file mode 100644
index 0000000..ab228aa
--- /dev/null
+++ b/research/experiments/046-commit-replica/prefixes.py
@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Research 046: what a follower may hold when the owner's whole machine dies (run.py's kills seldom leave one with
+part of a cut: a cut's records travel together). The owner takes EVENTS messages with the shim keeping every op it
+shipped (SHIM_JOURNAL); then the ops are done again in order into a fresh directory, and at SAMPLES points (op
+boundaries drawn at random, within cuts and at their ends, and the whole stream) a copy of what is there is opened by
+pure-run as the owner (and around every rename: a compaction's or an image's install). Each must open without a fault, hold the first `total` messages' entries exactly (summed), hold
+no fewer than at the sample before (a prefix never loses what a shorter one held), and the whole stream every message.
+
+  prefixes.py EVENTS SAMPLES     one JSON line a sample, then one of the totals"""
+import json
+import os
+import random
+import shutil
+import sys
+import time
+from pathlib import Path
+
+import run as harness
+
+OP_CREATE, OP_WRITE, OP_TRUNCATE, OP_ALLOCATE, OP_RENAME, OP_UNLINK, OP_MKDIR, OP_SYNC = range(1, 9)
+
+
+def ops_of(journal):
+    """The ops in order: (kind, path, rest) as bytes, and how many syncs (cuts) came before each."""
+    out, at, cuts = [], 0, 0
+    while at < len(journal):
+        kind, n = journal[at], int.from_bytes(journal[at + 1:at + 3], 'little')
+        path = journal[at + 3:at + 3 + n].decode()
+        p = at + 3 + n
+        if kind == OP_CREATE: end = p + 8
+        elif kind == OP_WRITE: end = p + 12 + int.from_bytes(journal[p + 8:p + 12], 'little')
+        elif kind == OP_TRUNCATE: end = p + 8
+        elif kind == OP_ALLOCATE: end = p + 16
+        elif kind == OP_RENAME: end = p + 2 + int.from_bytes(journal[p:p + 2], 'little')
+        elif kind in (OP_UNLINK, OP_MKDIR): end = p + 4
+        elif kind == OP_SYNC: end = p
+        else: raise SystemExit(f'unknown op {kind} at {at}')
+        out.append((kind, path, journal[p:end], cuts))
+        cuts += kind == OP_SYNC
+        at = end
+    return out
+
+
+def apply(root, op):
+    kind, path, rest, _ = op
+    target = root / path if path else root
+    if kind == OP_CREATE:
+        flags = int.from_bytes(rest[:4], 'little')
+        fd = os.open(target, os.O_WRONLY | flags, int.from_bytes(rest[4:8], 'little'))
+        os.close(fd)
+    elif kind == OP_WRITE:
+        offset, length = int.from_bytes(rest[:8], 'little'), int.from_bytes(rest[8:12], 'little')
+        fd = os.open(target, os.O_WRONLY)
+        os.pwrite(fd, rest[12:12 + length], offset)
+        os.close(fd)
+    elif kind == OP_TRUNCATE:
+        os.truncate(target, int.from_bytes(rest, 'little'))
+    elif kind == OP_ALLOCATE:
+        fd = os.open(target, os.O_WRONLY)
+        os.posix_fallocate(fd, int.from_bytes(rest[:8], 'little'), int.from_bytes(rest[8:16], 'little'))
+        os.close(fd)
+    elif kind == OP_RENAME:
+        n = int.from_bytes(rest[:2], 'little')
+        os.rename(target, root / rest[2:2 + n].decode())
+    elif kind == OP_UNLINK:
+        (os.rmdir if int.from_bytes(rest, 'little') & 0x200 else os.unlink)(target)
+    elif kind == OP_MKDIR:
+        target.mkdir(mode=int.from_bytes(rest, 'little'), exist_ok=True)
+
+
+def main():
+    events, samples = int(sys.argv[1]), int(sys.argv[2])
+    keys = harness.build()
+    journal = harness.WORK / 'journal'
+    harness.ENV['SHIM_JOURNAL'] = str(journal)
+    r = harness.Run(keys)
+    r.feed(events)
+    r.owner.wait()
+    r.reader.join(timeout=10)
+    r.shipper.communicate(timeout=60)
+    r.stop_follower()
+    del harness.ENV['SHIM_JOURNAL']
+    ops = ops_of(journal.read_bytes())
+    near = {i + d for i, op in enumerate(ops) if op[0] == OP_RENAME for d in range(-3, 4) if 0 < i + d <= len(ops)}
+    picks = sorted(set(random.sample(range(1, len(ops)), min(samples, len(ops) - 1))) | near | {len(ops)})
+    root, copy = harness.WORK / 'prefix', harness.WORK / 'prefix-open'
+    shutil.rmtree(root, ignore_errors=True)
+    root.mkdir()
+    done, last_total, failures, within = 0, -1, [], 0
+    started = time.time()
+    for pick in picks:
+        while done < pick:
+            apply(root, ops[done])
+            done += 1
+        mid = ops[pick - 1][0] != OP_SYNC  # the prefix ends inside a cut
+        within += mid
+        shutil.rmtree(copy, ignore_errors=True)
+        shutil.copytree(root, copy)
+        took = harness.opened(copy)
+        total = int(took.get('total', -1)) if took.get('opened') else -1
+        good = (took.get('opened') and took.get('fault') == 'none' and took.get('summed') == harness.summed(total) and
+                total >= last_total and (pick < len(ops) or total == events))
+        if not good:
+            failures.append({'ops': pick, 'cut': ops[pick - 1][3], 'within_cut': mid, 'took': took, 'before': last_total})
+        print(json.dumps({'ops': pick, 'of': len(ops), 'cut': ops[pick - 1][3], 'within_cut': mid, 'total': total,
+                          'passed': bool(good)}), flush=True)
+        last_total = max(last_total, total)
+    renamed = {}
+    for op in ops:
+        if op[0] == OP_RENAME:
+            to = op[2][2:].decode()
+            renamed[to] = renamed.get(to, 0) + 1
+    print(json.dumps({'run': 'prefixes', 'events': events, 'ops': len(ops), 'journal_bytes': journal.stat().st_size,
+                      'renamed_to': renamed, 'samples': len(picks), 'near_renames': len(near), 'within_cuts': within, 'failed': len(failures), 'failures': failures[:5],
+                      'seconds': round(time.time() - started, 1)}), flush=True)
+
+
+if __name__ == '__main__':
+    main()