From 9d7b059ab2ec6b1df771fbf3d18f141d40184c64 Mon Sep 17 00:00:00 2001 From: Frank McSherry Date: Mon, 9 Nov 2026 23:28:55 -0500 Subject: [PATCH 1/6] DDIR: ledgers, as a model and as programs on the server A ledger is a growing collection of updates, recorded as spans, that actors change only by transactions. interactive/model.md states the model: ledgers and spans, clocks, the invariants, transactions resolved in rounds, and what committing and reading must guarantee. The model runs. model/ledger.ddp is the state, the write frontier and the invariants. model/resolve.ddp is the rules for one round: Create and Append actions, sequence numbers, clocks, and rejections that say where a ledger stands. tests/ledgers.rs checks the two against each other. Both run as ordinary programs on the stock server, shown by three demos under server/demo. bench/ledger/fabric.py drives rounds from outside, makes each durable with a blob and a root swap, rebuilds from that log, and records what it measured in bench/ledger/README.md. The one change to DDIR itself is a set of frontier functions for lambdas: comparison, join, meet and canonical form over lists of lists of ints. Co-Authored-By: Claude Fable 5.2 Claude-Session: https://claude.ai/code/session_01H6wB4EnzBnYpvwz5E2ZGmD --- interactive/bench/ledger/README.md | 146 +++++++ interactive/bench/ledger/fabric.py | 466 +++++++++++++++++++++ interactive/bench/ledger/requests.ddp | 32 ++ interactive/bench/ledger/state.ddp | 22 + interactive/model.md | 225 ++++++++++ interactive/model/ledger.ddp | 80 ++++ interactive/model/resolve.ddp | 284 +++++++++++++ interactive/server/README.md | 8 + interactive/server/demo/ledger_resolver.py | 244 +++++++++++ interactive/server/demo/ledger_rounds.txt | 98 +++++ interactive/server/demo/ledger_views.txt | 58 +++ interactive/src/lambda.rs | 14 +- interactive/src/lambda/frontiers.rs | 323 ++++++++++++++ interactive/tests/ledgers.rs | 177 ++++++++ 14 files changed, 2171 insertions(+), 6 deletions(-) create mode 100644 interactive/bench/ledger/README.md create mode 100644 interactive/bench/ledger/fabric.py create mode 100644 interactive/bench/ledger/requests.ddp create mode 100644 interactive/bench/ledger/state.ddp create mode 100644 interactive/model.md create mode 100644 interactive/model/ledger.ddp create mode 100644 interactive/model/resolve.ddp create mode 100644 interactive/server/demo/ledger_resolver.py create mode 100644 interactive/server/demo/ledger_rounds.txt create mode 100644 interactive/server/demo/ledger_views.txt create mode 100644 interactive/src/lambda/frontiers.rs create mode 100644 interactive/tests/ledgers.rs diff --git a/interactive/bench/ledger/README.md b/interactive/bench/ledger/README.md new file mode 100644 index 000000000..671c5feaf --- /dev/null +++ b/interactive/bench/ledger/README.md @@ -0,0 +1,146 @@ +# Ledger rounds, resolved by DDIR and made durable + +`fabric.py` runs rounds of ledger transactions on a stock `ddir_server`, and makes each round durable. +The rules are [`../../model/resolve.ddp`](../../model/resolve.ddp), and the model they implement is [`../../model.md`](../../model.md). +Everything the server runs is `.ddp`; the driver feeds requests, reads conclusions, writes them to disk, and pumps. + +## How a round goes + +Three kinds of program run on the server. + +- `state`, from [`state.ddp`](state.ddp), stands for the whole run: the manifest, and the three indexes the rules read. +- `gen`, from [`requests.ddp`](requests.ddp), lives for one round: it shapes flat columns of ints into requests. +- `res`, the rules, lives for one round, bound to `state` and `gen`. + +The driver then does five things. + +1. It loads `gen` and `res`, and feeds the round's requests to `gen` as binary frames. +2. It ticks, and the rules decide every request. +3. It reads the round's change to the manifest, as binary frames. +4. It feeds the rules' conclusions into `state`, drops `gen` and `res`, and ticks. +5. It writes the frames as one blob, flushes, swaps the root file, and flushes again. + +A round's outcomes count as reported when step 5 returns. +In `pipelined` mode step 5 runs on another thread while the next round resolves, and rounds that queue behind a commit share the next one. +The log is a chain of blobs, each naming the one before, and a root file naming the last. + +## Run it + +```sh +cargo build --release -p ddir-server +python3 interactive/bench/ledger/fabric.py --dir /tmp/fabric --ledgers 100000 --clocks 1 --spans 0,4096 --check +``` + +`--check` kills the server afterwards and starts another. +It feeds the log into the new server's manifest, derives the indexes from the manifest with [`../../model/ledger.ddp`](../../model/ledger.ddp), and checks that model's invariants. +What it rebuilt must equal what the first server held, and one more round must resolve on it. +Every round of every run also checks that exactly the expected number of actions committed. + +## Recorded results + +Apple M4 Mac mini (mini-03), 2026-11-10, four workers, one million ledgers, means over 25 rounds after 5 of warmup (10 after 2 for the largest). +Every ledger has one time coordinate and every batch is empty, so these are spans of metadata. +Unless said otherwise: every ledger follows one clock, which advances every round; the manifest is resident; commits use `F_FULLFSYNC`. +A span is one committed append, to a ledger or to a clock. + +### Spans per round, from none to many + +| Spans with data per round | Serial ms per round | Serial spans/s | Pipelined ms per round | Pipelined spans/s | Compute ms | Commit ms | Latency p50 ms | +| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| 0 | 16.1 | | 8.1 | | 7.6 | 7.7 | 18.9 | +| 256 | 16.5 | 15.6k | 9.0 | 28.6k | 8.6 | 7.9 | 16.5 | +| 1,024 | 18.3 | 55.9k | 10.6 | 97.1k | 10.2 | 7.9 | 17.9 | +| 4,096 | 25.8 | 159k | 17.9 | 229k | 17.5 | 8.0 | 23.5 | +| 16,384 | 48.4 | 339k | 40.9 | 401k | 40.5 | 8.3 | 38.5 | +| 65,536 | 145.9 | 449k | 136.9 | 479k | 136.4 | 10.9 | 133.4 | +| 262,144 | | | 393.7 | 666k | 391.0 | 25.0 | 432 | + +Compute, commit and latency are from the pipelined runs; latency is from a round's start to its commit's return. + +- **The fixed cost of durability is about 8 ms a commit**, nearly all of it the two flushes: 3.5 ms after the blob and 4 ms after the root. + It does not grow until a round's record passes a megabyte. +- **Compute passes commit within a few hundred spans a round.** + An empty round already computes for 7.6 ms, against 7.7 ms to commit. +- **Past that, rounds are bound by compute**, and pipelining gains little: at 65,536 spans a round, commit is 11 ms of 137. +- **Throughput rises with round size**, from 229k spans/s at 4,096 a round to 666k at 262,144, as fixed costs spread; latency rises with it. + +### Where a round's compute goes + +Pipelined, in ms. + +| Spans per round | Load | Feed | Resolve | Read | Absorb | Total | +| ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| 0 | 5.8 | 0.1 | 0.9 | 0.2 | 0.6 | 7.6 | +| 4,096 | 6.2 | 0.2 | 7.7 | 1.8 | 1.7 | 17.5 | +| 65,536 | 6.6 | 0.9 | 80.8 | 32.8 | 15.3 | 136.4 | + +- **Loading the two programs costs about 6 ms every round**, most of an empty round. The server parses and plans them afresh each time. + Without it, compute would pass commit near 3,000 spans a round, not a few hundred. +- **Reading the conclusions out to the driver is a quarter of a large round.** A server that wrote its own blob would not pay it. +- **Resolving costs about 1.2 µs a span** at 65,536 a round, summed over the round's wall time. + +### What the flush costs + +Serial, in ms. + +| Sync | Commit, empty round | Round, empty | Commit, 16,384 spans | Round, 16,384 spans | +| --- | ---: | ---: | ---: | ---: | +| none | 0.4 | 7.8 | 0.9 | 45.1 | +| fsync | 0.4 | 7.9 | 1.1 | 45.1 | +| full | 8.3 | 16.1 | 9.4 | 48.4 | + +On macOS, fsync alone hands data to the drive without making the drive store it. + +### Clocks + +Pipelined. + +| Spans with data per round | No clocks, ms per round | No clocks, spans/s | One clock, ms per round | One clock, spans/s | +| ---: | ---: | ---: | ---: | ---: | +| 4,096 | 14.1 | 290k | 17.9 | 229k | +| 16,384 | 35.9 | 457k | 40.9 | 401k | +| 65,536 | 114.7 | 571k | 136.9 | 479k | + +An append to a ledger under a clock costs about a fifth more than to a free one. + +With every clock advanced every round, a clock's advance is a span like any other: + +| Clocks | Spans with data per round | ms per round | All spans/s | Spans with data/s | +| ---: | ---: | ---: | ---: | ---: | +| 100 | 0 | 8.7 | 11.5k | 0 | +| 100 | 4,096 | 16.1 | 260k | 254k | +| 100 | 16,384 | 39.2 | 420k | 418k | +| 10,000 | 0 | 24.1 | 415k | 0 | +| 10,000 | 4,096 | 32.1 | 440k | 128k | +| 10,000 | 16,384 | 54.1 | 487k | 303k | +| 100,000 | 0 | 161.8 | 618k | 0 | +| 100,000 | 4,096 | 179.4 | 580k | 22.8k | +| 100,000 | 16,384 | 205.5 | 566k | 79.7k | + +- **Advancing every clock lands 400k to 600k spans/s**, the same rate as appends with data. +- **What a clock saves is the spans it stands for.** Ten thousand clocks advance all million ledgers in 24 ms. + +### Kinds of commit + +Pipelined, one clock, 16,384 appends with data per round unless said. + +| Kind | ms per round | Committed/s | Note | +| --- | ---: | ---: | --- | +| One append per transaction | 40.9 | 401k spans | | +| Four appends per transaction | 40.6 | 404k spans | 101k transactions/s | +| With 8,192 more that repeat an append and are rejected | 53.7 | 305k spans | 457k transactions decided/s | +| Creations under a clock, 50,000 per round | 61 | 810k ledgers | | +| Creations of free ledgers, 50,000 per round | 41 | 1.21M ledgers | | +| Manifest not resident, 65,536 appends per round | 101.6 | 645k spans | 479k with it resident | + +### Recovery + +With one million ledgers and a log of 51 records, the new server rebuilt the manifest, derived the indexes and checked the invariants in 1.2 to 1.4 s. +What it rebuilt matched, no invariant was broken, and it resolved the next round. + +## What these runs do not show + +- Batches with data in them: writing them as blobs, or reading them back. +- A log that is ever shortened: recovery reads every record. +- More than one resolver, or a resolver that commits through a parent. +- A power failure: the server is killed, but the machine stays up. diff --git a/interactive/bench/ledger/fabric.py b/interactive/bench/ledger/fabric.py new file mode 100644 index 000000000..0bc20c25f --- /dev/null +++ b/interactive/bench/ledger/fabric.py @@ -0,0 +1,466 @@ +#!/usr/bin/env python3 +"""Rounds of ledger transactions, resolved by DDIR and made durable by a thin driver. + +The server runs three kinds of program, all `.ddp`: + state state.ddp, standing: the manifest (registrations and spans) and the three indexes the rules read + gen requests.ddp, one per round: shapes flat columns of ints into requests + res ../../model/resolve.ddp, one per round: the rules, bound to `state` and `gen` + +A round, driven from here: + 1. load `gen` and `res`, and feed the round's requests to `gen` as binary frames + 2. tick: the rules decide every request + 3. read the round's change to the manifest, as binary frames + 4. feed the rules' conclusions into `state`, drop `gen` and `res`, and tick + 5. write the frames as one blob, flush, swap the root, flush: the round is durable +Outcomes count as reported when step 5 returns. With `--mode pipelined` step 5 runs on another +thread while the next round resolves, and rounds that queue behind a commit are committed together. + +The durable log is a chain of blobs, each naming the one before it, with a root file naming the last. +`--check` kills the server, rebuilds the manifest from the log on a fresh one, derives the indexes +from it with ../../model/ledger.ddp, checks that model's invariants, and compares what it rebuilt. + +Every ledger here has one time coordinate, and every batch is empty: there is no client data yet. + + cargo build --release -p ddir-server + python3 interactive/bench/ledger/fabric.py --dir /tmp/fabric --ledgers 100000 --clocks 1 --spans 0,2560 --check +""" + +from __future__ import annotations + +import argparse +import fcntl +import hashlib +import json +import os +import queue +import random +import shutil +import socket +import struct +import subprocess +import sys +import threading +import time +from pathlib import Path + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parents[2] +sys.path.insert(0, str(HERE.parent / "ldbc")) +from client import decode, encode_frame # noqa: E402 (the binary frame encoder and decoder) + +FLAT = (("prod", ["int"]), ("prod", ["int"] * 5)) +RULES = "interactive/model/resolve.ddp" +BIND = "input0=state::own input1=gen::requests input2=state::registered input3=state::follows" +INVARIANTS = ["malformed", "orphaned", "forked", "skipped", "early", "unclocked", "mismatched"] + + +class Server: + """A ddir_server process and one session to it.""" + + def __init__(self, binary: Path, workers: int): + ports = [] + for _ in range(2): + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + ports.append(probe.getsockname()[1]) + env = dict(os.environ, DDIR_WORKERS=str(workers), DDIR_BIND=f"127.0.0.1:{ports[0]}", + DDIR_WS_BIND=f"127.0.0.1:{ports[1]}", DDIR_MAX_FEED_BYTES=str(1 << 31)) + self.process = subprocess.Popen([str(binary)], cwd=ROOT, env=env, stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + deadline = time.monotonic() + 15 + while True: + try: + self.stream = socket.create_connection(("127.0.0.1", ports[0]), timeout=1) + break + except OSError: + if self.process.poll() is not None or time.monotonic() > deadline: + raise RuntimeError("the server did not start") + time.sleep(0.01) + self.stream.settimeout(600) + self.reader = self.stream.makefile("rb") + self.count = 0 + + def ask(self, command: str, body: bytes = b"") -> tuple[list[str], list[bytes]]: + """Send one command, with a frame's bytes after it if any. Returns its data lines and data frames.""" + self.count += 1 + reqid = f"q{self.count}" + self.stream.sendall(f"{reqid} {command}\n".encode() + body) + lines, frames = [], [] + while True: + raw = self.reader.readline() + if not raw: + raise RuntimeError(f"the server went away during {command[:60]!r}") + line = raw.decode().rstrip("\r\n") + if not line.startswith(reqid + " "): + continue + reply = line[len(reqid) + 1:] + if reply.startswith("frame "): + kind, size = reply[6:].split(" ") + payload = self.reader.read(int(size)) + if kind == "data": + frames.append(payload) + elif reply.startswith("data "): + lines.append(reply[5:]) + elif reply == "ok" or reply.startswith("ok "): + return lines, frames + else: + raise RuntimeError(f"{command[:80]!r}: {reply}") + + def frames(self, trace: str) -> list[bytes]: + return self.ask(f"peek {trace} as corgi")[1] + + def rows(self, trace: str) -> list: + """A trace's rows, decoded, in a canonical order.""" + return sorted(decode([("data", frame) for frame in self.frames(trace)]), key=repr) + + def feed(self, target: str, frame: bytes) -> None: + self.ask(f"feed {target} frame {len(frame)}", frame) + + def kill(self) -> None: + self.process.kill() + self.process.wait() + + def close(self) -> None: + try: + self.process.stdin.write(b"exit\n") + self.process.stdin.flush() + self.process.wait(timeout=15) + except Exception: + self.kill() + + +def rows_in(frames: list[bytes]) -> int: + """How many updates a peek's data frames hold: each begins with its row count.""" + return sum(struct.unpack_from(" None: + if self.sync == "none": + return + os.fsync(fd) + if flush and self.sync == "full" and hasattr(fcntl, "F_FULLFSYNC"): + fcntl.fcntl(fd, fcntl.F_FULLFSYNC) + + def commit(self, frames: list[tuple[int, bytes]]) -> dict: + """Make these frames durable as the next record of the log. Returns where the time went.""" + clock = time.perf_counter() + record = bytearray(self.head.ljust(64).encode() + struct.pack(" list[list[tuple[int, bytes]]]: + """Every record's frames, oldest record first.""" + records, name = [], self.head + while name: + data = (self.blobs / name).read_bytes() + name, (count,) = data[:64].decode().strip(), struct.unpack_from(" None: + while (item := self.queue.get()) is not None: + batch = [item] + while not self.queue.empty(): + more = self.queue.get() + if more is None: + self.queue.put(None) + break + batch.append(more) + cost = self.store.commit([frame for _, frames, _ in batch for frame in frames]) + now = time.perf_counter() + for number, _, event in batch: + self.done[number], self.costs[number] = now, dict(cost, rounds=len(batch)) + event.set() + + def submit(self, number: int, frames: list[tuple[int, bytes]]) -> threading.Event: + event = threading.Event() + self.queue.put((number, frames, event)) + return event + + def finish(self) -> None: + self.queue.put(None) + self.thread.join() + + +class World: + """The driver's own account of the ledgers, used to write requests that will commit.""" + + def __init__(self, ledgers: int, clocks: int, seed: int): + self.ledgers, self.clocks, self.rng, self.key = ledgers, clocks, random.Random(seed), 0 + self.version, self.upper = [0] * ledgers, [0] * ledgers + self.clock_version, self.clock_upper = [0] * clocks, [0] * clocks + + def keys(self, count: int) -> range: + self.key += count + return range(self.key - count, self.key) + + def creation(self, first: int, count: int) -> tuple[bytes, int]: + """Create ledgers first..first+count, each in its own transaction, following its clock if there are clocks.""" + keys = [(k,) for k in self.keys(count)] + if self.clocks: + vals = [(0, n, 1, n % self.clocks, 1) for n in range(first, first + count)] + else: + vals = [(0, n, 0, 0, 0) for n in range(first, first + count)] + return encode_frame(FLAT, keys, [1] * count, vals), count + + def clock_creation(self) -> tuple[bytes, int]: + keys = [(k,) for k in self.keys(self.clocks)] + return encode_frame(FLAT, keys, [1] * self.clocks, [(1, c, 0, 0, 0) for c in range(self.clocks)]), self.clocks + + def round(self, spans: int, width: int, contend: float, tick: bool) -> dict: + """One round's requests: `spans` appends with (empty) data, `width` to a transaction, and one append per clock.""" + chosen = self.rng.sample(range(self.ledgers), spans) + floors = list(self.clock_upper) + appends = [] + for n in chosen: + lower = max(self.upper[n], floors[n % self.clocks]) if self.clocks else self.upper[n] + appends.append((0, n, lower, lower + 1, self.version[n])) + self.version[n], self.upper[n] = self.version[n] + 1, lower + 1 + singles, grouped, rejected = [], [], 0 + if width == 1: + singles = list(zip(((k,) for k in self.keys(len(appends))), appends)) + # A contender repeats an append under a later key; the rules must reject it. + for row in self.rng.sample(appends, int(contend * len(appends))): + singles.append(((self.keys(1)[0],), row)) + rejected += 1 + else: + for at in range(0, len(appends), width): + key = (self.keys(1)[0],) + grouped += [(key, row) for row in appends[at:at + width]] + ticks = [] + if tick: + for c in range(self.clocks): + ticks.append(((self.keys(1)[0],), (1, c, self.clock_upper[c], self.clock_upper[c] + 1, self.clock_version[c]))) + self.clock_version[c], self.clock_upper[c] = self.clock_version[c] + 1, self.clock_upper[c] + 1 + singles += ticks + frame = lambda rows: encode_frame(FLAT, [k for k, _ in rows], [1] * len(rows), [v for _, v in rows]) if rows else None + transactions = len(singles) + (len(appends) + width - 1) // width * (width > 1) + return dict(singles=frame(singles), grouped=frame(grouped), creates=None, spans=len(appends) + len(ticks), + data=len(appends), ticks=len(ticks), transactions=transactions, rejected=rejected) + + +class Fabric: + """One resolver: a server, its standing state, and the durable log.""" + + def __init__(self, args, store: Store): + self.args, self.store = args, store + self.server = Server(args.binary, args.workers) + self.server.ask(f"load state from {HERE.relative_to(ROOT)}/state.ddp") + self.server.ask("tick") + + def round(self, requests: dict) -> dict: + """Resolve one round and absorb it. Returns the manifest's change as frames, and the timings.""" + s, mark, t = self.server, time.perf_counter(), {} + + def lap(name: str) -> None: + nonlocal mark + now = time.perf_counter() + t[name], mark = now - mark, now + + s.ask(f"load gen from {HERE.relative_to(ROOT)}/requests.ddp") + s.ask(f"load res from {RULES} {BIND}") + lap("load") + for target, name in enumerate(("singles", "grouped", "creates")): + if requests.get(name): + s.feed(f"gen {target}", requests[name]) + lap("feed") + s.ask("tick") + lap("resolve") + registrations, spans = s.frames("res::registrations_add"), s.frames("res::spans_add") + lap("read") + s.ask("feed state 0 from res::own_delta") + s.ask("feed state 1 from res::registered_add") + s.ask("feed state 2 from res::follows_add") + if self.args.manifest == "resident": + s.ask("feed state 3 from res::registrations_add") + s.ask("feed state 4 from res::spans_add") + s.ask("drop res") + s.ask("drop gen") + s.ask("tick") + lap("absorb") + frames = [(3, f) for f in registrations] + [(4, f) for f in spans] + return dict(t, frames=frames, committed=rows_in(registrations) + rows_in(spans)) + + def recover(self) -> dict: + """On a fresh server: the manifest from the log, the indexes from the manifest, the model's invariants.""" + s, begin = self.server, time.perf_counter() + for record in self.store.log(): + for target, frame in record: + s.feed(f"state {target}", frame) + s.ask("tick") + s.ask("load check from interactive/model/ledger.ddp input0=state::registrations input1=state::spans") + s.ask("tick") + broken = {name: rows_in(s.frames(f"check::{name}")) for name in INVARIANTS} + for target, name in enumerate(("own", "registered", "follows")): + s.ask(f"feed state {target} from check::{name}") + s.ask("drop check") + s.ask("tick") + return dict(seconds=time.perf_counter() - begin, broken={k: v for k, v in broken.items() if v}) + + +def percentile(values, q): + ordered = sorted(values) + return ordered[min(len(ordered) - 1, int(q * len(ordered)))] + + +def phase(fabric: Fabric, rounds: list[dict], mode: str, warmup: int) -> dict: + """Run these rounds, committing each; report the rounds after the warmup.""" + committer = Committer(fabric.store) + started, events, results = [], [], [] + for number, requests in enumerate(rounds): + started.append(time.perf_counter()) + result = fabric.round(requests) + expected = requests["spans"] + (requests.get("created") or 0) + if result["committed"] != expected: + raise RuntimeError(f"round {number}: {result['committed']} committed, {expected} expected") + events.append(committer.submit(number, result.pop("frames"))) + results.append(result) + if mode == "serial": + events[-1].wait() + committer.finish() + kept = range(warmup, len(rounds)) + window = committer.done[len(rounds) - 1] - started[warmup] + mean = lambda values: sum(values) / len(kept) + part = lambda name: 1000 * mean([results[i][name] for i in kept]) + cost = lambda name: 1000 * mean([committer.costs[i][name] / committer.costs[i]["rounds"] for i in kept]) + latency = [1000 * (committer.done[i] - started[i]) for i in kept] + count = lambda name: sum(rounds[i].get(name) or 0 for i in kept) + return dict( + mode=mode, rounds=len(kept), ms_per_round=round(1000 * window / len(kept), 2), + spans_per_s=round(count("spans") / window), data_spans_per_s=round(count("data") / window), + creations_per_s=round(count("created") / window), transactions_per_s=round(count("transactions") / window), + rejected_per_round=count("rejected") // len(kept), + resolve_ms=round(part("resolve"), 2), absorb_ms=round(part("absorb"), 2), + load_ms=round(part("load"), 2), feed_ms=round(part("feed"), 2), read_ms=round(part("read"), 2), + compute_ms=round(sum(part(n) for n in ("load", "feed", "resolve", "read", "absorb")), 2), + commit_ms=round(sum(cost(n) for n in ("write", "flush", "root")), 2), + commit_write_ms=round(cost("write"), 2), commit_flush_ms=round(cost("flush"), 2), commit_root_ms=round(cost("root"), 2), + record_kib=round(mean([committer.costs[i]["bytes"] / committer.costs[i]["rounds"] for i in kept]) / 1024, 1), + rounds_per_commit=round(len(kept) / len({committer.done[i] for i in kept}), 2), + latency_p50_ms=round(percentile(latency, 0.5), 1), latency_p95_ms=round(percentile(latency, 0.95), 1)) + + +def main() -> None: + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument("--dir", type=Path, required=True, help="where the durable log goes; must not exist") + p.add_argument("--binary", type=Path, default=ROOT / "target/release/ddir_server") + p.add_argument("--workers", type=int, default=4) + p.add_argument("--ledgers", type=int, default=1_000_000) + p.add_argument("--clocks", type=int, default=1, help="how many clocks the ledgers follow; 0 for none") + p.add_argument("--still", action="store_true", help="do not advance the clocks") + p.add_argument("--spans", default="0,256,2560,25600", help="appends with data per round; one phase for each") + p.add_argument("--width", type=int, default=1, help="appends per transaction") + p.add_argument("--contend", type=float, default=0.0, help="fraction of appends that a second transaction repeats") + p.add_argument("--rounds", type=int, default=30) + p.add_argument("--warmup", type=int, default=5) + p.add_argument("--modes", default="serial,pipelined") + p.add_argument("--sync", choices=["none", "fsync", "full"], default="full") + p.add_argument("--manifest", choices=["resident", "none"], default="resident", + help="whether the server keeps every registration and span resident, or only the indexes") + p.add_argument("--create-batch", type=int, default=50_000, help="creations per round while setting up") + p.add_argument("--check", action="store_true", help="afterwards, kill the server and rebuild from the log") + p.add_argument("--seed", type=int, default=1) + args = p.parse_args() + args.dir.mkdir(parents=True) + store = Store(args.dir, args.sync) + fabric = Fabric(args, store) + world = World(args.ledgers, args.clocks, args.seed) + try: + # Creation: the clocks first, since a ledger may only follow a clock registered before its round. + creating = [] + if args.clocks: + frame, count = world.clock_creation() + creating.append(dict(creates=frame, spans=0, created=count, transactions=count)) + for first in range(0, args.ledgers, args.create_batch): + frame, count = world.creation(first, min(args.create_batch, args.ledgers - first)) + creating.append(dict(creates=frame, spans=0, created=count, transactions=count)) + report = phase(fabric, creating, "pipelined", min(1, len(creating) - 1)) + print(json.dumps(dict(phase="create", ledgers=args.ledgers, clocks=args.clocks, **report)), flush=True) + + for spans in [int(x) for x in args.spans.split(",")]: + for mode in args.modes.split(","): + rounds = [world.round(spans, args.width, args.contend, bool(args.clocks) and not args.still) + for _ in range(args.rounds)] + report = phase(fabric, rounds, mode, args.warmup) + print(json.dumps(dict(phase="append", spans=spans, width=args.width, contend=args.contend, + clocks=args.clocks, manifest=args.manifest, sync=args.sync, **report)), flush=True) + + if args.check: + before = fabric.server.rows("state::own") + fabric.server.kill() + fabric = Fabric(args, Store(args.dir, args.sync)) + recovered = fabric.recover() + after = fabric.server.rows("state::own") + # The rebuilt resolver must also go on: one more round of appends, all of which must commit. + more = world.round(min(1000, args.ledgers), 1, 0.0, bool(args.clocks) and not args.still) + went_on = fabric.round(more)["committed"] == more["spans"] + print(json.dumps(dict(phase="check", records=len(fabric.store.log()), ledgers=len(after), + rebuilt_matches=before == after, invariants_broken=recovered["broken"], + recovery_s=round(recovered["seconds"], 2), resolves_after_recovery=went_on)), flush=True) + if before != after or recovered["broken"] or not went_on: + raise SystemExit("the rebuilt state differs, or an invariant is broken") + finally: + fabric.server.close() + + +if __name__ == "__main__": + main() diff --git a/interactive/bench/ledger/requests.ddp b/interactive/bench/ledger/requests.ddp new file mode 100644 index 000000000..da210b989 --- /dev/null +++ b/interactive/bench/ledger/requests.ddp @@ -0,0 +1,32 @@ +-- Shape flat rows of ints into the requests the rules take (see ../../model/resolve.ddp). +-- A driver can then send its requests as plain columns of ints, and the shaping is done here. +-- Every ledger made this way has one time coordinate, starts at [0], and has an empty shape; +-- every append has an empty batch. +-- +-- input 0, singles : ((tx), (hi, lo, lower, upper, seqno)) one append, alone in its transaction +-- input 1, grouped : ((tx), (hi, lo, lower, upper, seqno)) one append of a transaction with several +-- input 2, creates : ((tx), (hi, lo, clock_hi, clock_lo, following)) one creation; several may share a transaction + +type Batch = Empty | Inline List(int) | Blob List(int); +type Clock = Free | Follows (int, int); +type Action = Create ((int, int), List(int), int, List(List(int)), Clock) + | Append ((int, int), List(List(int)), List(List(int)), Batch, int); + +let singles = input 0 : ((int), (int, int, int, int, int)); +let grouped = input 1 : ((int), (int, int, int, int, int)); +let creates = input 2 : ((int), (int, int, int, int, int)); + +-- Actions that share a transaction are listed in order of (ledger, kind), as the rules require. +let appends = grouped | map((tx, (hi, lo, lower, upper, seqno)) -> + (tx, ((hi, lo), 1, ((hi, lo), lower enlist enlist, upper enlist enlist, () inject Empty, seqno) inject Append))); +let creations = creates | map((tx, (hi, lo, clock_hi, clock_lo, following)) -> + let shape = [] : List(int) in + let clock = (following, (clock_hi, clock_lo) inject Follows, () inject Free) select in + (tx, ((hi, lo), 0, ((hi, lo), shape, 1, 0 enlist enlist, clock) inject Create))); +let together = (appends + creations) | reduce((_, vals) -> + ((vals map (((_, _, action), _) -> action)), 1) enlist); +let alone = singles | map((tx, (hi, lo, lower, upper, seqno)) -> + let action = ((hi, lo), lower enlist enlist, upper enlist enlist, () inject Empty, seqno) inject Append in + (tx, (action enlist))); + +output "requests" = together + alone; diff --git a/interactive/bench/ledger/state.ddp b/interactive/bench/ledger/state.ddp new file mode 100644 index 000000000..a6aae6139 --- /dev/null +++ b/interactive/bench/ledger/state.ddp @@ -0,0 +1,22 @@ +-- What a resolver keeps resident, each as an arrangement others can read. +-- +-- The manifest, which is the record: registrations and spans, as in ../../model/ledger.ddp. +-- Three indexes derived from it, which the rules read: each ledger's version and own upper, +-- its width and start, and the clock it follows. +-- A driver feeds each round's conclusions into all five; after a restart it feeds the manifest +-- from the durable log, and derives the indexes from that. + +type Batch = Empty | Inline List(int) | Blob List(int); +type Clock = Free | Follows (int, int); + +let own = input 0 : (((int, int)), (int, List(List(int)))); +let registered = input 1 : (((int, int)), (int, List(List(int)))); +let follows = input 2 : (((int, int)), ((int, int))); +let registrations = input 3 : (((int, int)), (List(int), int, List(List(int)), Clock)); +let spans = input 4 : (((int, int)), (int, List(List(int)), List(List(int)), List(List(int)), Batch)); + +output "own" = own; +output "registered" = registered; +output "follows" = follows; +output "registrations" = registrations; +output "spans" = spans; diff --git a/interactive/model.md b/interactive/model.md new file mode 100644 index 000000000..f842a5a0f --- /dev/null +++ b/interactive/model.md @@ -0,0 +1,225 @@ +# Ledgers: the model (draft) + +This states what a ledger is, what may change a set of ledgers, and what any implementation must guarantee. +It is a draft, written from the design notes of 9 and 10 November 2026. +The DDIR in it is excerpted from [`model/ledger.ddp`](model/ledger.ddp), which runs on the server. +The rules it describes are [`model/resolve.ddp`](model/resolve.ddp), and [`tests/ledgers.rs`](tests/ledgers.rs) checks each against the other. + +## Ledgers + +A LEDGER is a collection of updates `(data, time, diff)` that only grows. +Its data has a key shape and a value shape, which are Corgi shapes. +A TIME is a list of ints, one per coordinate, and the number of coordinates is the ledger's WIDTH. +Times are ordered coordinate by coordinate, so two times may be incomparable. +A FRONTIER is a set of times, none before another, written as a list of times. +A frontier stands for all the times at or after any of its elements. +The empty frontier stands for no times at all. + +A SPAN is a durable statement about part of a ledger: `(seqno, lower, upper, since, batch)`. + +* `lower` and `upper` are frontiers; the span holds exactly the ledger's updates at times at or after `lower` and not at or after `upper`. +* `since` is a frontier; the span's updates accumulate correctly at any time at or after `since`. +* `batch` says where the updates are: nowhere, because there are none; inline; or in a blob named by its hash. +* `seqno` numbers the ledger's spans: 0, 1, 2 and so on. + It lets a reader tell a span that is missing from a stretch of time that a clock passed over. + +A ledger's VERSION is the number of its spans. +A ledger is identified by an id, which reveals nothing about it. + +## State + +The state of a set of ledgers is two collections. + +``` +type Batch = Empty | Inline List(int) | Blob List(int); +type Clock = Free | Follows (int, int); + +-- registrations : ((id), (shape, width, start, clock)) +-- spans : ((id), (seqno, lower, upper, since, batch)) +let registrations = input 0 : (((int, int)), (List(int), int, List(List(int)), Clock)); +let spans = input 1 : (((int, int)), (int, List(List(int)), List(List(int)), List(List(int)), Batch)); +``` + +A registration records a ledger's shape, its width, the frontier it starts at, and the clock it follows, if any. +Everything else about a ledger is computed from these two collections. +Nothing else is a source of truth. + +## Clocks + +A CLOCK is an ordinary ledger, usually one that holds no data. +A ledger may FOLLOW one clock, and a clock may itself follow a clock. +What a ledger follows is fixed when it is created. +A follower has the same width as its clock. + +A ledger's WRITE FRONTIER is the frontier before which it is complete. +It is the join of the ledger's own upper with the write frontier of the clock it follows. +The ledger's own upper is the upper of its last span, or its start if it has none. + +``` +-- OWN: a ledger's version is the number of its spans, and its own upper is the upper +-- of its last span, or its start if it has none. Marks sort by seqno, the start first. +let marks = (registrations | map((id, (_, _, start, _)) -> (id, (-1, start)))) + + (spans | map((id, (seqno, _, upper, _, _)) -> (id, (seqno, upper)))); +let own = marks | reduce((_, vals) -> + let entries = vals map ((mark, _) -> mark) in + let last = ((entries len, 1) sub enlist, entries) gather in + last map ((seqno, upper) -> (((seqno, 1) add, upper), 1))); + +-- WRITE FRONTIER: the join of a ledger's own upper with the write frontier of the clock +-- it follows, which is in turn joined with that of the clock above it, and so on. +wf: { + let lifted = follows | map(((id), (clock)) -> ((clock), (id))) + | join(write, ((clock), (id), (frontier)) -> ((id), (frontier))); + set write = ((own | map((id, (_, upper)) -> (id, (upper)))) + lifted) | reduce((_, vals) -> + let none = [] : List(List(int)) in + let joined = ((0, none), vals) fold (((seen, sofar), ((frontier), _)) -> + (1, (seen, (sofar, frontier) call frontier_join, frontier) select)) in + ((joined.1), 1) enlist); +} +``` + +One append to a clock therefore advances every ledger below it, with no span for any of them. +After such an advance a follower's spans no longer abut: its next span begins at the new write frontier, past where its last span ended. +The times in between hold no updates. + +This definition does work for every ledger. +An implementation need not: it may compute the write frontier only for the ledgers it is asked about. + +## Invariants + +These hold of every reachable state. +Each is a view in `model/ledger.ddp` that must be empty. + +1. **malformed.** A span's lower is strictly before its upper. +2. **orphaned.** Every span belongs to a registered ledger. +3. **forked, skipped.** A ledger's spans are numbered 0, 1, 2 and so on, with no number used twice and none skipped. +4. **early.** Each span begins no earlier than the span before it ended, and the first no earlier than the ledger's start. +5. **unclocked, mismatched.** A followed clock is registered, and has its follower's width. + +Invariant 3 is what lets a reader holding spans 0 to n know that none is missing. +Invariant 4 does not ask spans to abut, because of clocks. + +## Transactions + +The one way to change the state is a TRANSACTION: a list of actions that all take effect, or none do. + +``` +type Action = Create (id, shape, width, start, clock) + | Append (id, lower, upper, batch, seqno); +-- requests : ((tx), (List(Action))) +``` + +A transaction lists its actions in order of ledger id, a `Create` before an `Append` of the same ledger. +So it names a ledger in at most one `Create` and at most one `Append`, and this can be checked from the request alone. + +* `Create` succeeds if the id is not registered and, when it names a clock, that clock was registered before the round, with the same width. + It adds the registration. +* `Append` succeeds if the ledger is registered, `seqno` equals its version, `lower` equals its write frontier, and `lower` is strictly before `upper`. + It adds the span `(seqno, lower, upper, start, batch)`: a new span's `since` is the ledger's start. + +A transaction COMMITS if every one of its actions succeeds, in order, and otherwise it is REJECTED and changes nothing. +A rejection names the first action that failed and why: a stale append, an id already registered, an id not registered, or a request that is invalid in itself. +A stale append is told the ledger's version and write frontier as the round leaves them, which is where a retry must begin. +A request is invalid in itself if its actions are out of order, a frontier's times differ in width, an append's lower is not strictly before its upper, or its times are not of the ledger's width. + +The writer of a batch vouches for it: its updates have the ledger's shapes, and lie in the span. +The rules do not open a batch. + +### Rounds + +Transactions are resolved in ROUNDS. +Within a round they are taken in the order of their keys, which is the order they arrived in. + +* An action sees the registrations and spans left by the earlier transactions of the round that committed. +* A write frontier in a round is the join of the ledger's own upper, as those earlier transactions left it, with its clock's write frontier as it stood when the round began. + +So an append to a clock takes effect after every other append of its round. +It never causes an append to a follower in the same round to be rejected, and every later round sees it. + +Among independent writers of one ledger, at most one commits in a round, and the others try again. + +This is weaker than running whole transactions one after another. +Two transactions that each append to one ledger and advance the other's clock both commit in one round, though neither order of running them whole would commit both. +What a round guarantees is that each transaction is all or nothing, and that the state it leaves satisfies the invariants. + +## Committing + +A RESOLVER holds the state of some ledgers and resolves rounds of transactions on them. +It is an ordinary DDIR node, and the rules above are a DDIR program it runs. + +* **Writers are dumb.** A writer puts its batches in the blob store, submits transactions to a resolver, and waits for their outcomes. + A writer never commits anything itself. +* **A manifest is a ledger.** A resolver's MANIFEST is the ledger of changes to its registrations and spans. + Its time has one coordinate, the round. + Each round, the resolver writes that round's changes as a blob and appends one span to its manifest. +* **A resolver is a writer too.** It commits its manifest's span by submitting an `Append` to its parent resolver, which treats the batch as opaque. +* **A root ends this.** A ROOT is a resolver whose last step is a compare-and-swap on a durable register. + There may be several roots. +* **A ledger is in one manifest, and a manifest has one parent.** + Several instances may try to advance the same manifest; one wins each round, and the others take up the span that committed. + +Any implementation must guarantee: + +1. An outcome is reported only after it is durable at every level up to a root. +2. A committed transaction survives the failure of any process, and a rejected or lost one leaves no visible effect, even if its blobs were written. +3. Every blob a span names is durable before the span commits. +4. Rounds of one manifest are totally ordered, and a transaction that completed before another was submitted is ordered before it. + +## Reading + +A reader of a ledger needs its registration, its spans, and the write frontiers of the clocks above it. +The ledger's contents at every time not at or after its write frontier are exactly the updates in its spans. + +* Blobs are immutable and named by their contents, so the value of a root's register names one consistent state of everything beneath it. + A reader needs that one value, fresh or stale, and after that reads only blobs. +* A resolver may also serve readers directly: it looks a ledger up by id, and sends subscribers the non-empty spans and the clock advances. +* A reader may begin computing from blobs before their spans are confirmed. + It must not submit transactions built on them until they are. + +## Housekeeping + +None of this is built. + +* **Compaction.** Consecutive spans may be replaced by one span that covers their seqnos, holds their updates consolidated, and has a later `since`. + Adding such a span is always safe, since it is determined by the spans it covers. + Removing the spans it covers is what must be guarded. +* **Holds.** What history must be kept is recorded as data: a ledger of intents, each saying what it still needs. + The compaction allowed is computed from that ledger. +* **Collection.** A blob that no root reaches may be deleted, once no transaction in flight could still name it. + +## Not settled + +1. **Rounds are weaker than serial transactions,** as described under Rounds. + This follows from letting a clock's advance not contend with appends; is it acceptable as stated? + It matters if clocks are to be used for mutual exclusion, where advancing a clock is meant to fence another writer out. +2. **A slow writer behind a busy clock.** An append's lower must equal the write frontier, and a clock's advance moves the write frontier of every follower. + A writer that takes longer than a round to answer a rejection is stale again when it does. + In the resolver demo a slow writer landed nothing while its clock advanced every round, and everything once the clock stopped. + Letting a lower be later than the write frontier, which amounts to an implied empty span, would let such a writer aim ahead of the clock. +3. **Ties.** Among transactions that could each commit alone, the one with the least key wins, every round. + In the resolver demo one of three writers to a shared ledger won 15 rounds running. +4. **Width and clocks.** A follower must have its clock's width, so a follower's width could be taken from its clock instead of stated. +5. **What a root's register holds.** Perhaps the list of the root manifest's spans, or the hash of a blob that holds it. +6. **Whether DDIR's own times should be ints.** Ledger times are ints and DDIR's are unsigned. They match in structure, and the question is put off. +7. **Transactions across manifests, skipping resolution for deterministic ledgers, and commits across independent roots** are deferred or ruled out for now; see the design notes. + +## Where the code stands + +* **The model runs.** `model/ledger.ddp` is the state, the write frontier and the invariants. + `server/demo/ledger_views.txt` shows the write frontiers under a two-level clock tree, and then breaks each invariant once. +* **The rules run.** `model/resolve.ddp` resolves one round as described above. + `server/demo/ledger_rounds.txt` runs three rounds on the stock server. +* **They are checked against each other.** `tests/ledgers.rs` runs rounds of the rules, and after each has the model derive the indexes from the manifest alone; they must agree, and every invariant must hold. +* **Many sessions can feed one resolver.** `server/demo/ledger_resolver.py` has nine client sessions and a tenth that pumps the rounds. +* **Rounds can be made durable, and recovered.** `bench/ledger/fabric.py` writes each round's change to the manifest as a blob and swaps a root file. + With `--check` it kills the server, rebuilds the manifest from the log, derives the indexes with the model, and checks the invariants. + Its measurements are in `bench/ledger/README.md`. +* **Frontier comparisons never fail.** They take times of different widths, reading a missing coordinate as before every int, so a ragged request cannot stop a server. + The rules reject such requests before comparing them. + +What the server does not do itself, and a driver does from outside: + +* commit a round durably, and release its outcomes only afterwards; +* rebuild its state from a manifest when it starts; +* pump the rounds, and assign transaction keys; +* read a ledger's batches from blobs, or write them: every batch so far is empty. diff --git a/interactive/model/ledger.ddp b/interactive/model/ledger.ddp new file mode 100644 index 000000000..972c503e1 --- /dev/null +++ b/interactive/model/ledger.ddp @@ -0,0 +1,80 @@ +-- The state of a set of ledgers, and what is derived from it (see model.md). +-- +-- A time is a list of ints, one per coordinate, and a frontier is a list of times. +-- Ledger ids are two ints. Shapes are opaque bytes. +-- +-- registrations : ((id), (shape, width, start, clock)) +-- spans : ((id), (seqno, lower, upper, since, batch)) +-- +-- The views below are computed from these two collections and nothing else. +-- The outputs after INVARIANTS must all be empty. + +type Batch = Empty | Inline List(int) | Blob List(int); +type Clock = Free | Follows (int, int); + +let registrations = input 0 : (((int, int)), (List(int), int, List(List(int)), Clock)); +let spans = input 1 : (((int, int)), (int, List(List(int)), List(List(int)), List(List(int)), Batch)); + +-- FOLLOWS: each ledger that follows a clock, and the clock it follows. +let follows = registrations + | filter((_, (_, _, _, clock)) -> clock match (Free (_ -> 0), Follows (_ -> 1))) + | map((id, (_, _, _, clock)) -> (id, (clock match (Free (_ -> (0, 0)), Follows (c -> c))))); + +-- OWN: a ledger's version is the number of its spans, and its own upper is the upper +-- of its last span, or its start if it has none. Marks sort by seqno, the start first. +let marks = (registrations | map((id, (_, _, start, _)) -> (id, (-1, start)))) + + (spans | map((id, (seqno, _, upper, _, _)) -> (id, (seqno, upper)))); +let own = marks | reduce((_, vals) -> + let entries = vals map ((mark, _) -> mark) in + let last = ((entries len, 1) sub enlist, entries) gather in + last map ((seqno, upper) -> (((seqno, 1) add, upper), 1))); + +-- WRITE FRONTIER: the join of a ledger's own upper with the write frontier of the clock +-- it follows, which is in turn joined with that of the clock above it, and so on. +wf: { + let lifted = follows | map(((id), (clock)) -> ((clock), (id))) + | join(write, ((clock), (id), (frontier)) -> ((id), (frontier))); + set write = ((own | map((id, (_, upper)) -> (id, (upper)))) + lifted) | reduce((_, vals) -> + let none = [] : List(List(int)) in + let joined = ((0, none), vals) fold (((seen, sofar), ((frontier), _)) -> + (1, (seen, (sofar, frontier) call frontier_join, frontier) select)) in + ((joined.1), 1) enlist); +} + +output "follows" = follows; +output "own" = own; +output "write" = wf::write; +-- REGISTERED: each ledger's width and start. With `own` and `follows`, these are the indexes +-- a resolver keeps beside the manifest, and can rebuild from it. +output "registered" = registrations | map((id, (_, width, start, _)) -> (id, (width, start))); + +-- INVARIANTS: each of these must be empty. + +-- A span's lower is strictly before its upper. +output "malformed" = spans | filter((_, (_, lower, upper, _, _)) -> ((lower, upper) call frontier_lt, 0) eq); + +-- Every span belongs to a registered ledger. +output "orphaned" = spans - (spans | join(registrations, (id, span, _) -> (id, span))); + +-- A ledger's spans are numbered 0, 1, 2, and so on: no number twice, and none skipped. +let numbered = spans | map(((id), (seqno, lower, upper, _, _)) -> ((id, seqno), (lower, upper))); +output "forked" = numbered | reduce((_, vals) -> ((vals len), 1) enlist) | filter((_, (count)) -> (count, 1) gt); +let present = numbered | map((key, _) -> (key, ())) | distinct; +let wanted = numbered | filter(((_, seqno), _) -> (seqno, 0) gt) + | map(((id, seqno), _) -> ((id, (seqno, 1) sub), ())) | distinct; +output "skipped" = wanted - (wanted | join(present, (key, (), ()) -> (key, ()))); + +-- Each span begins no earlier than the span before it ended, and the first no earlier than the start. +-- Spans need not abut: a clock may have advanced the ledger in between. +let ends = (numbered | map(((id, seqno), (_, upper)) -> ((id, (seqno, 1) add), (upper)))) + + (registrations | map(((id), (_, _, start, _)) -> ((id, 0), (start)))); +output "early" = numbered | join(ends, (key, (lower, _), (end)) -> (key, (end, lower))) + | filter((_, (end, lower)) -> ((end, lower) call frontier_le, 0) eq); + +-- A followed clock is registered, and has its follower's width. +let widths = registrations | map((id, (_, width, _, _)) -> (id, (width))); +let clocked = follows | map(((id), (clock)) -> ((clock), (id))) + | join(widths, ((clock), (id), (width)) -> ((id), (clock, width))); +output "unclocked" = follows - (clocked | map((id, (clock, _)) -> (id, (clock)))); +output "mismatched" = clocked | join(widths, (id, (clock, theirs), (ours)) -> (id, (clock, theirs, ours))) + | filter((_, (_, theirs, ours)) -> (theirs, ours) ne); diff --git a/interactive/model/resolve.ddp b/interactive/model/resolve.ddp new file mode 100644 index 000000000..d0b720216 --- /dev/null +++ b/interactive/model/resolve.ddp @@ -0,0 +1,284 @@ +-- The transaction rules of the ledger model, for one round (see ../model.md). +-- +-- A round takes the requests waiting, in the order of their keys, and decides each. +-- A transaction is a list of actions that all take effect or none do. +-- `Create` registers a ledger. `Append` adds a span, and succeeds only if its `seqno` +-- is the ledger's version and its `lower` is the ledger's write frontier. +-- +-- The state these rules read is three derived indexes, all as the round found them: +-- input 0, own : ((id), (version, upper)) a ledger's version and its own last upper +-- input 2, registered : ((id), (width, start)) +-- input 3, follows : ((id), (clock)) only for ledgers that follow a clock +-- Input 1 is the round's requests: ((tx), (actions)). +-- +-- A write frontier is the join of the ledger's own upper, as the round's earlier +-- transactions left it, with the own uppers of the clocks above it as the round found them. +-- So an append to a clock lands after every other append of its round. +-- +-- Actions must be listed in order of ledger id, a `Create` before an `Append` of the same ledger, +-- so that no ledger is created twice or appended twice in one transaction. +-- A `Create` that names a clock needs that clock registered before the round, with the same width. +-- +-- Outputs: +-- results ((tx), (outcome)) every request's outcome +-- registrations_add ((id), (shape, width, start, clock)) the round's change to the manifest, +-- spans_add ((id), (seqno, lower, upper, since, batch)) to be made durable +-- own_delta, registered_add, follows_add the changes to the three indexes +-- taken ((tx), (actions)) the requests this round decided + +type Batch = Empty | Inline List(int) | Blob List(int); +type Clock = Free | Follows (int, int); +type Action = Create ((int, int), List(int), int, List(List(int)), Clock) + | Append ((int, int), List(List(int)), List(List(int)), Batch, int); +-- A rejection for a stale append reports the ledger's version and write frontier as the round left them. +type Failure = Stale (int, List(List(int))) | Exists (int, int) | Unknown (int, int) | Invalid; +type Outcome = Committed | Rejected (int, Failure); +-- What a ledger's group holds, in this order: its registered width, its version and own upper, +-- the own upper of each clock above it, and then the round's actions on it by key. +-- A proposal is (tx, index, kind, seqno, lower, upper, width, eligible); kind 0 creates, 1 appends. +type Row = Reg int + | Own (int, List(List(int))) + | Floor List(List(int)) + | Proposal (int, int, int, int, List(List(int)), List(List(int)), int, int); + +let own = input 0 : (((int, int)), (int, List(List(int)))); +let requests = input 1 : ((int), (List(Action))); +let registered = input 2 : (((int, int)), (int, List(List(int)))); +let follows = input 3 : (((int, int)), ((int, int))); + +-- SHAPE: every time in a frontier has one width. This uses no frontier comparison, so that +-- the comparisons below never meet a ragged frontier. The result is the first misshapen action, or -1. +let sized = requests | map((tx, (actions)) -> + let shapely = actions map (action -> action match ( + Create ((_, _, width, start, _) -> + (width, start) cap_list map ((width, time) -> (time len, width) eq) fold_all), + Append ((_, lower, upper, _, _) -> + let width = lower map (time -> time len) fold_max in + ((lower len, 0) gt, + (width, lower) cap_list map ((width, time) -> (time len, width) eq) fold_all, + (width, upper) cap_list map ((width, time) -> (time len, width) eq) fold_all) all))) in + let first = ((0, -1), shapely) fold (((index, bad), ok) -> + ((index, 1) add, (((bad, 0) lt, (ok, 0) eq) all, index, bad) select)) in + (tx, (actions, first.1))); +let misshapen = sized | filter((_, (_, bad)) -> (bad, 0) ge) | map((tx, (_, bad)) -> (tx, (bad))); + +-- ORDER AND BOUNDS: actions in order of (ledger, kind), and each append's lower strictly before its upper. +-- Frontiers are put in canonical form here, once; equal frontiers are then equal as values. +let ordered = sized | filter((_, (_, bad)) -> (bad, 0) lt) | map((tx, (actions, _)) -> + let actions = actions map (action -> action match ( + Create ((id, shape, width, start, clock) -> + (id, shape, width, start call frontier_normalize, clock) inject Create), + Append ((id, lower, upper, batch, seqno) -> + (id, lower call frontier_normalize, upper call frontier_normalize, batch, seqno) inject Append))) in + let first = ((((0, 0), 0), 0, 0, -1), actions) fold (((previous, seen, index, bad), action) -> + let (place, bounds) = action match ( + Create ((id, _, _, _, _) -> ((id, 0), 1)), + Append ((id, lower, upper, _, _) -> ((id, 1), (lower, upper) call frontier_lt))) in + let ok = ((seen, (previous, place) lt, 1) select, bounds) all in + (place, 1, (index, 1) add, (((bad, 0) lt, (ok, 0) eq) all, index, bad) select)) in + (tx, (actions, first.3))); +let disordered = ordered | filter((_, (_, bad)) -> (bad, 0) ge) | map((tx, (_, bad)) -> (tx, (bad))); +let invalid = misshapen + disordered; +let valid = ordered | filter((_, (_, bad)) -> (bad, 0) lt) | map((tx, (actions, _)) -> (tx, (actions))); + +-- One row per action. The control fields the decision needs are projected once; +-- shapes and batches stay out of the decision. +let acts = valid | flatmap((tx, (actions)) -> actions); +let checks = acts | map((tx, (index, action)) -> + ((tx, index), action) cap_sum match ( + Create (((tx, index), (id, _, width, start, clock)) -> + let (following, clock) = clock match (Free (_ -> (0, (0, 0))), Follows (clock -> (1, clock))) in + (tx, (id, index, 0, 0, start, start, width, following, clock))), + Append (((tx, index), (id, lower, upper, _, seqno)) -> + (tx, (id, index, 1, seqno, lower, upper, lower map (time -> time len) fold_max, 0, (0, 0)))))); + +-- The ledgers this round touches, and every clock above one of them. +let touched = checks | map((_, (id, _, _, _, _, _, _, _, _)) -> ((id), ())) | distinct; +up: { + let next = above | map(((seed), (clock)) -> ((clock), (seed))) + | join(follows, ((clock), (seed), (parent)) -> ((seed), (parent))); + set above = ((touched | join(follows, (seed, (), (clock)) -> (seed, (clock)))) + next) | distinct; +} +let clocks = up::above | map((_, (clock)) -> ((clock), ())) | distinct; +-- The probes of the resident indexes: at the touched ledgers, and at the clocks above them. +let owned = own | join(touched, (id, standing, ()) -> (id, standing)); +let known = registered | join(touched, (id, registration, ()) -> (id, registration)); +let ticking = own | join(clocks, (clock, (_, upper), ()) -> (clock, (upper))); +-- Each touched ledger's floors: the own uppers of the clocks above it. +let floors = up::above | map(((seed), (clock)) -> ((clock), (seed))) + | join(ticking, ((clock), (seed), (upper)) -> ((seed), (upper))); + +-- CREATIONS THAT FOLLOW A CLOCK. The clock must be registered, with the creation's width. +-- The new ledger's write frontier is its start joined with the clock's. +-- The clocks named and those above them are found by their own climb, as small as such creations are few. +let hitched = checks | filter((_, (_, _, kind, _, _, _, _, following, _)) -> ((kind, 0) eq, following) all) + | map(((tx), (id, index, _, _, start, _, width, _, clock)) -> ((clock), (tx, index, id, width, start))); +let named = hitched | map((clock, _) -> (clock, ())) | distinct; +over: { + let next = above | map(((seed), (clock)) -> ((clock), (seed))) + | join(follows, ((clock), (seed), (parent)) -> ((seed), (parent))); + set above = ((named | join(follows, (seed, (), (clock)) -> (seed, (clock)))) + next) | distinct; +} +let clock_frontier = ((own | join(named, (clock, (_, upper), ()) -> (clock, (upper)))) + + (over::above | map(((seed), (clock)) -> ((clock), (seed))) + | join(own, ((clock), (seed), (_, upper)) -> ((seed), (upper))))) + | reduce((_, vals) -> + let none = [] : List(List(int)) in + let joined = ((0, none), vals) fold (((seen, sofar), ((upper), _)) -> + (1, (seen, (sofar, upper) call frontier_join, upper) select)) in + ((joined.1), 1) enlist); +let hitching = hitched | join(registered, ((clock), (tx, index, id, width, start), (theirs, _)) -> + ((clock), (tx, index, id, width, start, theirs))); +-- Failures are ((tx), (index, code, ledger, version, frontier, stands)); only a stale append fills the last three. +let clockless = (hitched | map(((clock), (tx, index, id, _, _)) -> + let none = [] : List(List(int)) in ((tx), (index, 3, clock, 0, none, 0)))) + - (hitching | map(((clock), (tx, index, id, _, _, _)) -> + let none = [] : List(List(int)) in ((tx), (index, 3, clock, 0, none, 0)))); +let unfit = hitching | filter((_, (_, _, _, width, _, theirs)) -> (width, theirs) ne) + | map(((clock), (tx, index, id, _, _, _)) -> + let none = [] : List(List(int)) in ((tx), (index, 4, id, 0, none, 0))); +let unhitched = clockless + unfit; +let doomed = unhitched | map((tx, _) -> (tx, ())) | distinct; +let hitches = hitching | filter((_, (_, _, _, width, _, theirs)) -> (width, theirs) eq) + | join(clock_frontier, ((clock), (tx, index, id, width, start, _), (frontier)) -> + let begins = (start, frontier) call frontier_join in + ((tx), (id, index, 0, 0, begins, begins, width))); + +-- What each action proposes: (tx) -> (id, index, kind, seqno, lower, upper, width). +-- A creation's lower and upper are both the new ledger's write frontier. +let proposed = (checks | filter((_, (_, _, _, _, _, _, _, following, _)) -> (following, 0) eq) + | map((tx, (id, index, kind, seqno, lower, upper, width, _, _)) -> + (tx, (id, index, kind, seqno, lower, upper, width)))) + + hitches; + +-- What each touched ledger brings into the round. +let standing = (known | map((id, (width, _)) -> (id, width inject Reg))) + + (owned | map((id, standing) -> (id, standing inject Own))) + + (floors | map((id, (upper)) -> (id, upper inject Floor))); + +res: { + -- A rejected transaction's actions are still judged, but take no effect. + let excluded = proposed | join(rejected, (tx, proposal, ()) -> (tx, proposal)); + let proposals = ((proposed - excluded) | map(((tx), (id, index, kind, seqno, lower, upper, width)) -> + ((id), (tx, index, kind, seqno, lower, upper, width, 1) inject Proposal))) + + (excluded | map(((tx), (id, index, kind, seqno, lower, upper, width)) -> + ((id), (tx, index, kind, seqno, lower, upper, width, 0) inject Proposal))); + + -- One ledger's round. The state is (known, version, write frontier, width), and then what a + -- creation would have set had its transaction not been rejected: (has, tx, write frontier, width). + -- An append is judged against its own transaction's creation even when that takes no effect, + -- so that it is told its lower was stale, not that the ledger is unknown. + -- Codes: 0 succeeds, 1 stale, 2 exists, 3 unknown, 4 the wrong width. + let bad = (standing + proposals) | reduce(((id), vals) -> + let none = [] : List(List(int)) in + let entries = vals map ((row, _) -> row) in + let final = ((0, 0, none, 0, 0, 0, none, 0), entries) foldscan ((state, row) -> + (state, row) cap_sum match ( + Reg (((_, version, write, _, has, mine, theirs, wide), width) -> + ((1, version, write, width, has, mine, theirs, wide), (0, 0, 0, 0, write))), + Own (((known, _, _, width, has, mine, theirs, wide), (version, upper)) -> + ((known, version, upper, width, has, mine, theirs, wide), (0, 0, 0, 0, upper))), + Floor (((known, version, write, width, has, mine, theirs, wide), floor) -> + ((known, version, (write, floor) call frontier_join, width, has, mine, theirs, wide), (0, 0, 0, 0, floor))), + Proposal (((known, version, write, width, has, mine, theirs, wide), (tx, index, kind, seqno, lower, upper, span, eligible)) -> + let creating = (kind, 0) eq in + let own = (has, (mine, tx) eq) all in + let (k, v, w, d) = ((own, 1, known) select, (own, 0, version) select, + (own, theirs, write) select, (own, wide, width) select) in + let current = ((seqno, v) eq, (lower, w) eq) all in + let code = (creating, + (known, 2, 0) select, + (k, ((span, d) eq, (current, 0, 1) select, 4) select, 3) select) select in + let apply = ((code, 0) eq, eligible) all in + let keep = ((code, 0) eq, creating, (eligible, 0) eq) all in + ((((apply, creating) all, 1, known) select, + (apply, (creating, 0, (version, 1) add) select, version) select, + (apply, upper, write) select, + ((apply, creating) all, span, width) select, + (keep, 1, has) select, (keep, tx, mine) select, + (keep, upper, theirs) select, (keep, span, wide) select), + (tx, index, code, v, w))))) in + let failed = (final.1 map ((_, _, code, _, _) -> (code, 0) ne), final.1) zip filter in + -- A failure reports the ledger as the round leaves it, if it stands then; else as it was judged. + let (stands, version, write, _, _, _, _, _) = final.0 in + ((stands, version, write, id), failed) cap_list map (((stands, version, write, id), (tx, index, code, v, w)) -> + ((tx, index, code, id, (stands, version, v) select, (stands, write, w) select, stands), 1))) + | map((_, (tx, index, code, id, version, write, stands)) -> ((tx), (index, code, id, version, write, stands))); + set rejected = ((bad | map((tx, _) -> (tx, ()))) + doomed) | distinct; +} + +-- The first failed action of each rejected transaction. +let failures = (res::bad + unhitched) | min; +let rejected = res::rejected; +let committed = acts - (acts | join(rejected, (tx, act, ()) -> (tx, act))); + +-- THE ROUND'S CHANGE TO THE MANIFEST. +let created = committed | flatmap((_, (_, action)) -> action match ( + Create ((id, shape, width, start, clock) -> ((id), (shape, width, start, clock)) enlist), + Append (_ -> + let none = [] : List(((int, int), List(int), int, List(List(int)))) in + none map ((id, shape, width, start) -> ((id), (shape, width, start, () inject Free)))))) + | map((_, (_, row)) -> row); +let appended = committed | flatmap((_, (_, action)) -> action match ( + Create (_ -> + let none = [] : List(((int, int), List(List(int)), List(List(int)), int)) in + none map ((id, lower, upper, seqno) -> ((id), (lower, upper, () inject Empty, seqno)))), + Append ((id, lower, upper, batch, seqno) -> ((id), (lower, upper, batch, seqno)) enlist))) + | map((_, (_, row)) -> row); +-- A new span's since is its ledger's start, registered before the round or created in it. +let starts = (known | map((id, (_, start)) -> (id, (start)))) + + (created | map((id, (_, _, start, _)) -> (id, (start)))); +output "registrations_add" = created; +output "spans_add" = appended | join(starts, (id, (lower, upper, batch, seqno), (start)) -> + (id, (seqno, lower, upper, start, batch))); + +-- THE CHANGES TO THE INDEXES. +-- A ledger's new standing is its last committed action's: the start of a creation, or the upper of an append. +let marks = (created | map((id, (_, _, start, _)) -> (id, (-1, start)))) + + (appended | map((id, (_, upper, _, seqno)) -> (id, (seqno, upper)))); +let renewed = marks | reduce((_, vals) -> + let entries = vals map ((mark, _) -> mark) in + let last = ((entries len, 1) sub enlist, entries) gather in + last map ((seqno, upper) -> (((seqno, 1) add, upper), 1))); +let own_delta = renewed - (owned | join(renewed, (id, standing, _) -> (id, standing))); +output "own_delta" = own_delta; +output "registered_add" = created | map((id, (_, width, start, _)) -> (id, (width, start))); +let follows_add = created | filter((_, (_, _, _, clock)) -> clock match (Free (_ -> 0), Follows (_ -> 1))) + | map((id, (_, _, _, clock)) -> (id, (clock match (Free (_ -> (0, 0)), Follows (c -> c))))); +output "follows_add" = follows_add; + +-- OUTCOMES. A stale append is told where its ledger stands as the round leaves it: its version, and its +-- write frontier, which is its own upper joined with the uppers, after this round, of the clocks above it. +-- Those clocks are found by one more climb, from the stale ledgers only: along the clocks they followed +-- before the round, or, for a ledger created in this round, from the clock it was created under. +let stale = failures | filter((_, (_, code, _, _, _, _)) -> (code, 1) eq) + | map(((tx), (index, _, id, version, frontier, stands)) -> ((id), (tx, index, version, frontier, stands))); +let staled = stale | filter((_, (_, _, _, _, stands)) -> stands) | map((id, _) -> (id, ())) | distinct; +beyond: { + let next = above | map(((seed), (clock)) -> ((clock), (seed))) + | join(follows, ((clock), (seed), (parent)) -> ((seed), (parent))); + set above = ((staled | join(follows, (seed, (), (clock)) -> (seed, (clock)))) + + (follows_add | join(staled, (seed, (clock), ()) -> (seed, (clock)))) + + next) | distinct; +} +-- A clock's upper after the round is the later of its resident upper and any this round gave it. +let lifted = beyond::above | map(((seed), (clock)) -> ((clock), (seed))); +let lifts = (lifted | join(own, ((clock), (seed), (_, upper)) -> ((seed), (upper)))) + + (lifted | join(renewed, ((clock), (seed), (_, upper)) -> ((seed), (upper)))); +let left = ((stale | map((id, (_, _, _, frontier, _)) -> (id, (frontier)))) + lifts) + | reduce((_, vals) -> + let none = [] : List(List(int)) in + let joined = ((0, none), vals) fold (((seen, sofar), ((frontier), _)) -> + (1, (seen, (sofar, frontier) call frontier_join, frontier) select)) in + ((joined.1), 1) enlist); +let told = (stale | join(left, (id, (tx, index, version, _, _), (frontier)) -> + ((tx), ((index, (version, frontier) inject Stale) inject Rejected)))) + + (failures | filter((_, (_, code, _, _, _, _)) -> (code, 1) ne) + | map((tx, (index, code, id, _, _, _)) -> + (tx, ((index, ((code, 2) eq, id inject Exists, + ((code, 3) eq, id inject Unknown, () inject Invalid) select) select) inject Rejected)))) + + (invalid | map((tx, (index)) -> (tx, ((index, () inject Invalid) inject Rejected)))); +output "results" = ((valid | map((tx, _) -> (tx, (() inject Committed)))) + - (valid | join(rejected, (tx, _, ()) -> (tx, (() inject Committed))))) + + told; +output "taken" = requests; diff --git a/interactive/server/README.md b/interactive/server/README.md index e44898f5d..7d67c5eb6 100644 --- a/interactive/server/README.md +++ b/interactive/server/README.md @@ -257,6 +257,14 @@ fronting proxy if a deployment ever needs one. ./target/release/ddir_server < interactive/server/demo/txn.txt python3 interactive/server/demo/two_sessions.py # races + size gate over TCP +Ledgers, as [`../model.md`](../model.md) describes them, run as ordinary programs on this server: + + ./target/release/ddir_server < interactive/server/demo/ledger_views.txt # the state, its views, its invariants + ./target/release/ddir_server < interactive/server/demo/ledger_rounds.txt # three rounds of transactions + python3 interactive/server/demo/ledger_resolver.py # nine sessions feed one resolver, a tenth pumps it + +[`../bench/ledger`](../bench/ledger/README.md) makes such rounds durable, and measures them. + ## Benchmarking The [LDBC-derived workload](../bench/ldbc/README.md) runs parameterized interactive diff --git a/interactive/server/demo/ledger_resolver.py b/interactive/server/demo/ledger_resolver.py new file mode 100644 index 000000000..ad19a8374 --- /dev/null +++ b/interactive/server/demo/ledger_resolver.py @@ -0,0 +1,244 @@ +#!/usr/bin/env python3 +"""A resolver fed by many sessions, pumped from outside: ledger transactions on the stock server. + +Nothing here is durable, and no client has data: every batch is empty. The point is the plumbing. + +Standing programs on one server: + state the manifest and the indexes the rules read (interactive/bench/ledger/state.ddp) + intake the transaction requests clients have fed and no round has yet taken + outcomes every transaction's result, for clients to read + +A pump session runs rounds: load the rules (interactive/model/resolve.ddp) against those programs, tick, +feed the conclusions into `state` and the results into `outcomes`, retract the taken requests from +`intake`, drop the rules, tick. + +Client sessions, each on its own connection, feed requests into `intake` and poll `outcomes`: + owners each creates its own ledgers, then appends to all of them in one transaction per step + contenders all append to one shared ledger; a rejection reports where the ledger stands, and they retry + timekeeper creates a clock, then its followers, then advances the clock every step + straggler appends slowly to a follower, learning from rejections where the clock has moved it to + +Run from the repository root, after `cargo build --release -p ddir-server`: + python3 interactive/server/demo/ledger_resolver.py +""" +import os, re, socket, subprocess, sys, threading, time + +HERE = os.path.dirname(os.path.abspath(__file__)) +ROOT = os.path.abspath(os.path.join(HERE, "..", "..", "..")) +BIN = os.path.join(ROOT, "target", "release", "ddir_server") +PORT = 7961 +OWNERS, LEDGERS_EACH, STEPS = 4, 8, 40 # owners, ledgers per owner, appends per ledger +CONTENDERS, WINS = 3, 15 # contenders on one ledger, commits each must land +FOLLOWERS, TICKS = 200, 40 # ledgers following the clock, clock advances + +TYPES = """type Batch = Empty | Inline List(int) | Blob List(int); +type Clock = Free | Follows (int, int); +type Action = Create ((int, int), List(int), int, List(List(int)), Clock) + | Append ((int, int), List(List(int)), List(List(int)), Batch, int); +type Failure = Stale (int, List(List(int))) | Exists (int, int) | Unknown (int, int) | Invalid; +type Outcome = Committed | Rejected (int, Failure); +""" +PROGRAMS = { + "intake": TYPES + """let requests = input 0 : ((int), (List(Action))); +output "requests" = requests; +""", + "outcomes": TYPES + """let results = input 0 : ((int), (Outcome)); +output "results" = results; +""", +} +RULES = ("load res from interactive/model/resolve.ddp input0=state::own input1=intake::requests " + "input2=state::registered input3=state::follows") + + +class Session: + """One TCP connection; every command is sent with an id and waited on.""" + def __init__(self, name): + self.name, self.count, self.buf = name, 0, b"" + for _ in range(200): + try: + self.sock = socket.create_connection(("127.0.0.1", PORT), timeout=30) + break + except OSError: + time.sleep(0.05) + else: + raise RuntimeError("server never came up") + + def ask(self, text): + self.count += 1 + reqid = f"{self.name}{self.count}" + self.sock.sendall(f"{reqid} {text}\n".encode()) + data = [] + while True: + while b"\n" not in self.buf: + chunk = self.sock.recv(1 << 16) + if not chunk: raise RuntimeError("connection closed") + self.buf += chunk + line, self.buf = self.buf.split(b"\n", 1) + toks = line.decode().split(" ", 2) + if toks[0] != reqid: continue + if toks[1] == "data": data.append(toks[2]) + elif toks[1] == "ok": return data + else: raise RuntimeError(f"{text[:60]!r}: {line.decode()}") + + +def frontier(t): return f"list(list({t}))" +def create(ledger, clock=None): + following = f"inject(1,tuple({clock}))" if clock else "inject(0,tuple())" + return f"inject(0,tuple(tuple({ledger}),list(),1,{frontier(0)},{following}))" +def append(ledger, lower, upper, seqno): + return f"inject(1,tuple(tuple({ledger}),{frontier(lower)},{frontier(upper)},inject(0,tuple()),{seqno}))" + + +class Client(Session): + """Feeds one transaction at a time and waits for its outcome.""" + def __init__(self, number): + super().__init__(f"c{number}_") + self.number, self.sent, self.waits = number, 0, [] + + def transact(self, actions): + """Returns None if it committed; (version, frontier) from a stale append's rejection; else ().""" + key = self.number * 10**9 + self.sent + self.sent += 1 + begin = time.perf_counter() + self.ask(f"feed intake 0 {key} val=tuple(list({','.join(actions)}))") + while True: + rows = self.ask(f"peek outcomes::results {key}") + if rows: break + time.sleep(0.0005) + self.waits.append(time.perf_counter() - begin) + if "val=tuple(inject(0," in rows[0]: return None + found = re.search(r"inject\(0,tuple\((\d+),list\(list\((\d+)\)\)\)\)", rows[0]) + return (int(found.group(1)), int(found.group(2))) if found else () + + +def owner(number, done): + c = Client(number) + mine = [f"{10 + number},{j}" for j in range(LEDGERS_EACH)] + assert c.transact([create(l) for l in mine]) is None + for step in range(STEPS): + assert c.transact([append(l, step, step + 1, step) for l in mine]) is None + done[number] = c + + +def contender(number, done): + c = Client(number) + version, frontier, wins, losses = 0, 0, 0, 0 + if number == OWNERS: c.transact([create("1,0")]) # the first contender creates the shared ledger + while wins < WINS: + told = c.transact([append("1,0", frontier, frontier + 1, version)]) + if told is None: version, frontier, wins = version + 1, frontier + 1, wins + 1 + elif told == (): time.sleep(0.002) # the ledger does not exist yet + else: (version, frontier), losses = told, losses + 1 + c.losses = losses + done[number] = c + + +def timekeeper(number, done): + c = Client(number) + assert c.transact([create("9,0")]) is None + # A ledger may follow only a clock registered before its round, so the followers come second. + assert c.transact([create(f"5,{j}", "9,0") for j in range(FOLLOWERS)]) is None + for t in range(TICKS): + assert c.transact([append("9,0", t, t + 1, t)]) is None + done[number] = c + + +def straggler(number, done): + c = Client(number) + version, frontier, wins, losses = 0, 0, 0, 0 + while wins < 10: + time.sleep(0.02) # slow enough for the clock to get ahead + told = c.transact([append("5,7", frontier, frontier + 1, version)]) + if told is None: version, frontier, wins = version + 1, frontier + 1, wins + 1 + elif told == (): time.sleep(0.002) # the ledger does not exist yet + else: (version, frontier), losses = told, losses + 1 + c.losses, c.final, c.version = losses, frontier, version + done[number] = c + + +def pump(stop, rounds): + s = Session("pump") + while not stop.is_set(): + s.ask(RULES) + s.ask("tick") + s.ask("feed outcomes 0 from res::results") + for target, name in enumerate(("own_delta", "registered_add", "follows_add", "registrations_add", "spans_add")): + s.ask(f"feed state {target} from res::{name}") + s.ask("feed intake 0 from res::taken diff=-1") + s.ask("drop res") + s.ask("tick") + rounds.append(time.perf_counter()) + + +def check(label, ok, detail=""): + print(("PASS " if ok else "FAIL ") + label + (f" [{detail}]" if detail else ""), flush=True) + if not ok: sys.exit(1) + + +def main(): + env = dict(os.environ, DDIR_BIND=f"127.0.0.1:{PORT}", DDIR_WS_BIND=f"127.0.0.1:{PORT + 1}", + DDIR_WORKERS=os.environ.get("DDIR_WORKERS", "4")) + server = subprocess.Popen([BIN], cwd=ROOT, env=env, stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=open(os.environ["LEDGER_DEMO_LOG"], "w") if "LEDGER_DEMO_LOG" in os.environ else subprocess.DEVNULL) + try: + setup = Session("setup") + setup.ask("load state from interactive/bench/ledger/state.ddp") + for name, text in PROGRAMS.items(): + setup.ask(f"load {name} begin\n{text}{name} end-load") + setup.ask("tick") + + stop, rounds, done = threading.Event(), [], {} + jobs = [(owner, n) for n in range(OWNERS)] + jobs += [(contender, OWNERS + n) for n in range(CONTENDERS)] + jobs += [(timekeeper, OWNERS + CONTENDERS), (straggler, OWNERS + CONTENDERS + 1)] + threads = [threading.Thread(target=job, args=(n, done)) for job, n in jobs] + pumper = threading.Thread(target=pump, args=(stop, rounds)) + begin = time.perf_counter() + pumper.start() + for t in threads: t.start() + for t in threads: t.join() + elapsed = time.perf_counter() - begin + stop.set(); pumper.join() + check("every client finished", len(done) == len(jobs)) + + current = {} + for row in setup.ask("peek state::own"): + a, b, v, f = re.match(r"tuple\(tuple\((\d+),(\d+)\)\) val=tuple\((\d+),list\(list\((\d+)\)\)\)", row).groups() + current[f"{a},{b}"] = (int(v), int(f)) + owned = [current.get(f"{10 + n},{j}") for n in range(OWNERS) for j in range(LEDGERS_EACH)] + check(f"{OWNERS} owners' {len(owned)} ledgers all reached {STEPS}", owned == [(STEPS, STEPS)] * len(owned)) + shared = CONTENDERS * WINS + check(f"the shared ledger took exactly {shared} appends", current.get("1,0") == (shared, shared)) + check(f"the clock reached {TICKS}", current.get("9,0") == (TICKS, TICKS)) + idle = [current.get(f"5,{j}") for j in range(FOLLOWERS) if j != 7] + check(f"{len(idle)} idle followers have no spans of their own", idle == [(0, 0)] * len(idle)) + s = done[OWNERS + CONTENDERS + 1] + check("the straggler's follower holds its 10 appends", current.get("5,7") == (10, s.final), f"own upper {s.final}") + # The model's own account of the manifest must agree with the indexes, and find nothing wrong with it. + setup.ask("load check from interactive/model/ledger.ddp input0=state::registrations input1=state::spans") + setup.ask("tick") + check("the model's view of the manifest matches the resolver's index", + sorted(setup.ask("peek check::own")) == sorted(setup.ask("peek state::own"))) + broken = [name for name in ("malformed", "orphaned", "forked", "skipped", "early", "unclocked", "mismatched") + if setup.ask(f"peek check::{name}")] + check("the model's invariants hold of the manifest", not broken, ", ".join(broken)) + check("no request is left in the intake", setup.ask("peek intake::requests") == []) + + sent = sum(c.sent for c in done.values()) + results = len(setup.ask("peek outcomes::results")) + check("every transaction has exactly one outcome", results == sent, f"{results} of {sent}") + lost = sum(getattr(c, "losses", 0) for c in done.values()) + waits = sorted(w for c in done.values() for w in c.waits) + print(f"{len(jobs)} client sessions, {sent} transactions ({sent - lost} committed, {lost} rejected and retried), " + f"{len(rounds)} rounds in {elapsed:.2f} s", flush=True) + print(f"contenders' rejections: {[done[OWNERS + n].losses for n in range(CONTENDERS)]}; " + f"straggler's: {s.losses}", flush=True) + print(f"submit-to-outcome: median {1000 * waits[len(waits) // 2]:.1f} ms, " + f"worst {1000 * waits[-1]:.1f} ms; round: {1000 * elapsed / len(rounds):.1f} ms", flush=True) + finally: + server.kill() + + +if __name__ == "__main__": + main() diff --git a/interactive/server/demo/ledger_rounds.txt b/interactive/server/demo/ledger_rounds.txt new file mode 100644 index 000000000..62f335aa5 --- /dev/null +++ b/interactive/server/demo/ledger_rounds.txt @@ -0,0 +1,98 @@ +# Ledger transactions, resolved in rounds on the stock server (see interactive/model/resolve.ddp and model.md). +# +# `state` holds the manifest and the three indexes the rules read. Each round loads the rules against it, +# feeds the round's requests, ticks, reads the outcomes, feeds the conclusions back, and drops the rules. +# A request is a list of actions: inject(0, (id, shape, width, start, clock)) creates a ledger, and +# inject(1, (id, lower, upper, batch, seqno)) appends a span. Nothing here is durable; for that, +# see interactive/bench/ledger/fabric.py. +# +# ./target/release/ddir_server < interactive/server/demo/ledger_rounds.txt + +load state from interactive/bench/ledger/state.ddp +tick +# Round 1. Ledger ids are (0,n); every time has one coordinate. +# 10 creates clock 8. 20 creates ledger 3 and appends [0,1) to it. +# 30 creates ledger 1 under clock 8: rejected, since a clock must be registered before the round. +# 40 gives a start of the wrong width. 50 gives an upper of another width than its lower. +# 60 lists its ledgers out of order. 70 appends [1,2) to ledger 3, as version 1. +# 80 appends at the same place: stale, and told ledger 3 now stands at version 2, frontier [2]. +# 90 appends to a ledger nobody created. +load res from interactive/model/resolve.ddp input0=state::own input2=state::registered input3=state::follows +feed res 1 begin +10 val=tuple(list(inject(0,tuple(tuple(0,8),list(7),1,list(list(0)),inject(0,tuple()))))) +20 val=tuple(list(inject(0,tuple(tuple(0,3),list(7),1,list(list(0)),inject(0,tuple()))),inject(1,tuple(tuple(0,3),list(list(0)),list(list(1)),inject(0,tuple()),0)))) +30 val=tuple(list(inject(0,tuple(tuple(0,1),list(7),1,list(list(0)),inject(1,tuple(0,8)))))) +40 val=tuple(list(inject(0,tuple(tuple(0,4),list(7),2,list(list(0)),inject(0,tuple()))))) +50 val=tuple(list(inject(1,tuple(tuple(0,5),list(list(0)),list(list(1,2)),inject(0,tuple()),0)))) +60 val=tuple(list(inject(1,tuple(tuple(0,3),list(list(1)),list(list(2)),inject(0,tuple()),1)),inject(0,tuple(tuple(0,2),list(7),1,list(list(0)),inject(0,tuple()))))) +70 val=tuple(list(inject(1,tuple(tuple(0,3),list(list(1)),list(list(2)),inject(0,tuple()),1)))) +80 val=tuple(list(inject(1,tuple(tuple(0,3),list(list(1)),list(list(5)),inject(0,tuple()),1)))) +90 val=tuple(list(inject(1,tuple(tuple(0,77),list(list(0)),list(list(1)),inject(0,tuple()),0)))) +end-feed +tick +peek res::results +peek res::registrations_add +peek res::spans_add +feed state 0 from res::own_delta +feed state 1 from res::registered_add +feed state 2 from res::follows_add +feed state 3 from res::registrations_add +feed state 4 from res::spans_add +drop res +tick +peek state::own +# Round 2. +# 10 creates ledger 1 under clock 8 and appends [0,1). 20 creates a follower of another width than its clock. +# 30 advances clock 8 to [5]. 40 creates clock 9, which follows clock 8. +load res from interactive/model/resolve.ddp input0=state::own input2=state::registered input3=state::follows +feed res 1 begin +10 val=tuple(list(inject(0,tuple(tuple(0,1),list(7),1,list(list(0)),inject(1,tuple(0,8)))),inject(1,tuple(tuple(0,1),list(list(0)),list(list(1)),inject(0,tuple()),0)))) +20 val=tuple(list(inject(0,tuple(tuple(0,6),list(7),2,list(list(0,0)),inject(1,tuple(0,8)))))) +30 val=tuple(list(inject(1,tuple(tuple(0,8),list(list(0)),list(list(5)),inject(0,tuple()),0)))) +40 val=tuple(list(inject(0,tuple(tuple(0,9),list(7),1,list(list(0)),inject(1,tuple(0,8)))))) +end-feed +tick +peek res::results +feed state 0 from res::own_delta +feed state 1 from res::registered_add +feed state 2 from res::follows_add +feed state 3 from res::registrations_add +feed state 4 from res::spans_add +drop res +tick +# Round 3. Ledger 1's own upper is [1], but its clock stands at [5], and advances to [7] in this round. +# 10 appends to ledger 1 at [1]: stale, and told it stands at version 2, frontier [7], as the round leaves it. +# 20 creates ledger 2 under clock 9 and appends at [0]: stale, since clock 9 follows clock 8, at [5]. +# 30 advances clock 8 to [7]. 40 appends [5,6) to ledger 1: the clock's advance lands after it. +# 50 creates ledger 2 and appends [5,6). 60 appends to ledger 3 with times of two coordinates. +# 70 creates ledger 12 and appends at [4]: stale against its own creation, which starts at [0]. +load res from interactive/model/resolve.ddp input0=state::own input2=state::registered input3=state::follows +feed res 1 begin +10 val=tuple(list(inject(1,tuple(tuple(0,1),list(list(1)),list(list(2)),inject(0,tuple()),1)))) +20 val=tuple(list(inject(0,tuple(tuple(0,2),list(7),1,list(list(0)),inject(1,tuple(0,9)))),inject(1,tuple(tuple(0,2),list(list(0)),list(list(1)),inject(0,tuple()),0)))) +30 val=tuple(list(inject(1,tuple(tuple(0,8),list(list(5)),list(list(7)),inject(0,tuple()),1)))) +40 val=tuple(list(inject(1,tuple(tuple(0,1),list(list(5)),list(list(6)),inject(0,tuple()),1)))) +50 val=tuple(list(inject(0,tuple(tuple(0,2),list(7),1,list(list(0)),inject(1,tuple(0,9)))),inject(1,tuple(tuple(0,2),list(list(5)),list(list(6)),inject(0,tuple()),0)))) +60 val=tuple(list(inject(1,tuple(tuple(0,3),list(list(2,0)),list(list(3,0)),inject(0,tuple()),2)))) +70 val=tuple(list(inject(0,tuple(tuple(0,12),list(7),1,list(list(0)),inject(0,tuple()))),inject(1,tuple(tuple(0,12),list(list(4)),list(list(5)),inject(0,tuple()),0)))) +end-feed +tick +peek res::results +peek res::spans_add +feed state 0 from res::own_delta +feed state 1 from res::registered_add +feed state 2 from res::follows_add +feed state 3 from res::registrations_add +feed state 4 from res::spans_add +drop res +tick +peek state::own +peek state::follows +# The model's own account of the manifest agrees with the index, and finds nothing wrong with it. +load check from interactive/model/ledger.ddp input0=state::registrations input1=state::spans +tick +peek check::own +peek check::write +peek check::early +peek check::skipped +exit diff --git a/interactive/server/demo/ledger_views.txt b/interactive/server/demo/ledger_views.txt new file mode 100644 index 000000000..58224f3ef --- /dev/null +++ b/interactive/server/demo/ledger_views.txt @@ -0,0 +1,58 @@ +# The state of a set of ledgers and its derived views (see interactive/model/ledger.ddp and interactive/model.md). +# +# First a valid state: the write frontiers under a two-level clock tree, and every invariant view empty. +# Then one row that breaks each invariant, and the views that report it. +# +# ./target/release/ddir_server < interactive/server/demo/ledger_views.txt + +load ledgers from interactive/model/ledger.ddp +# Ledgers 1 and 2 follow clock 9; clock 9 follows clock 8; ledger 3 is free. All have width 1 and start at [0]. +feed ledgers 0 begin +tuple(tuple(0,1)) val=tuple(list(1,2),1,list(list(0)),inject(1,tuple(0,9))) +tuple(tuple(0,2)) val=tuple(list(1,2),1,list(list(0)),inject(1,tuple(0,9))) +tuple(tuple(0,3)) val=tuple(list(1,2),1,list(list(0)),inject(0,tuple())) +tuple(tuple(0,8)) val=tuple(list(1,2),1,list(list(0)),inject(0,tuple())) +tuple(tuple(0,9)) val=tuple(list(1,2),1,list(list(0)),inject(1,tuple(0,8))) +end-feed +# Ledger 2 has spans [0,3) then [5,8), not abutting; clock 9 advanced to [5]; clock 8 to [7]; ledger 3 has [0,1). +feed ledgers 1 begin +tuple(tuple(0,2)) val=tuple(0,list(list(0)),list(list(3)),list(list(0)),inject(0,tuple())) +tuple(tuple(0,2)) val=tuple(1,list(list(5)),list(list(8)),list(list(0)),inject(0,tuple())) +tuple(tuple(0,9)) val=tuple(0,list(list(0)),list(list(5)),list(list(0)),inject(0,tuple())) +tuple(tuple(0,8)) val=tuple(0,list(list(0)),list(list(7)),list(list(0)),inject(0,tuple())) +tuple(tuple(0,3)) val=tuple(0,list(list(0)),list(list(1)),list(list(0)),inject(0,tuple())) +end-feed +tick +peek ledgers::follows +peek ledgers::own +peek ledgers::write +peek ledgers::malformed +peek ledgers::orphaned +peek ledgers::forked +peek ledgers::skipped +peek ledgers::early +peek ledgers::unclocked +peek ledgers::mismatched +# Now break each invariant once: a span with lower at its upper (ledger 1), a span of an unregistered +# ledger (77), a version used twice and one skipped (ledger 3), a span that begins before the last +# ended (clock 8), a clock that is not registered (ledger 4), and a clock of another width (ledger 5). +feed ledgers 0 begin +tuple(tuple(0,4)) val=tuple(list(1,2),1,list(list(0)),inject(1,tuple(0,55))) +tuple(tuple(0,5)) val=tuple(list(1,2),2,list(list(0,0)),inject(1,tuple(0,9))) +end-feed +feed ledgers 1 begin +tuple(tuple(0,1)) val=tuple(0,list(list(4)),list(list(4)),list(list(0)),inject(0,tuple())) +tuple(tuple(0,77)) val=tuple(0,list(list(0)),list(list(1)),list(list(0)),inject(0,tuple())) +tuple(tuple(0,3)) val=tuple(0,list(list(0)),list(list(2)),list(list(0)),inject(0,tuple())) +tuple(tuple(0,3)) val=tuple(3,list(list(2)),list(list(3)),list(list(0)),inject(0,tuple())) +tuple(tuple(0,8)) val=tuple(1,list(list(6)),list(list(9)),list(list(0)),inject(0,tuple())) +end-feed +tick +peek ledgers::malformed +peek ledgers::orphaned +peek ledgers::forked +peek ledgers::skipped +peek ledgers::early +peek ledgers::unclocked +peek ledgers::mismatched +exit diff --git a/interactive/src/lambda.rs b/interactive/src/lambda.rs index 4e1aaca35..a30d8bda2 100644 --- a/interactive/src/lambda.rs +++ b/interactive/src/lambda.rs @@ -33,6 +33,7 @@ use corgi::{ArithOp, Builder, CmpOp, Graph, NumOp, Op, Pred, Scalar, Shape, Valu pub(crate) mod print; pub(crate) mod normalize; +mod frontiers; /// A slot of the IR: a corgi program as written, `let pat = input in body`, and its graph, whose /// input is the tuple of the slot's rows (`(key, val)` for a linear op, `(key, left val, right @@ -446,11 +447,9 @@ pub(crate) fn compile_projection(lambda: &Lambda, rows: &[Shape]) -> Res corgi::Scope { static BASE: std::sync::OnceLock = std::sync::OnceLock::new(); BASE.get_or_init(|| { @@ -465,6 +464,9 @@ pub(crate) fn base_scope() -> corgi::Scope { let input = b.input(); let out = b.add(NumOp::Host(corgi::HostOp(std::sync::Arc::new(BitLength))), vec![input]); scope.word("bit_length", b.finish(out)).unwrap(); + for function in frontiers::functions() { + scope.word(function.name(), function_word(function)).unwrap(); + } scope.declare("enum Option = None () | Some").unwrap(); scope.declare("enum Result = Ok | Err").unwrap(); scope @@ -476,7 +478,7 @@ pub(crate) fn base_scope() -> corgi::Scope { fn is_base_word(name: &str) -> bool { base_words().any(|word| word == name) } fn base_words() -> impl Iterator { - FloatWord::all().map(|(name, _)| name).chain(["bit_length"]) + FloatWord::all().map(|(name, _)| name).chain(["bit_length"]).chain(frontiers::NAMES) } /// The bit length of an Int viewed as a u64: zero gives 0, negative integers give 64. diff --git a/interactive/src/lambda/frontiers.rs b/interactive/src/lambda/frontiers.rs new file mode 100644 index 000000000..0a5dada53 --- /dev/null +++ b/interactive/src/lambda/frontiers.rs @@ -0,0 +1,323 @@ +//! Shared frontier operations over `List(List(int))` antichains of product times. +//! Comparisons ignore order, duplicates, and dominated points. +//! Normalize, join, and meet return sorted minimal antichains, retaining the input time width. +//! Points of different widths may meet in one row: a missing trailing coordinate is the minimum, as in `PointStamp`. +//! No well-shaped input is an error, so a request with ragged times cannot stop a dataflow. +//! Results have the widest width seen in their row; an empty frontier has no width witness. + +use std::sync::{Arc, OnceLock}; + +use corgi::{Bounds, HostKernel, Shape, Value}; +use differential_dataflow::dynamic::pointstamp::PointStamp; +use differential_dataflow::lattice::Lattice; +use timely::progress::frontier::Antichain; + +use super::Function; + +type Time = PointStamp; +type Frontier = Antichain