Run TPC-C on Pure and on SQLite, and compare the states they leave

5 files changed, 1316 insertions, 0 deletions

diff --git a/benchmarks/tpcc/data.py b/benchmarks/tpcc/data.py
new file mode 100644
index 0000000..d8a4ccb
--- /dev/null
+++ b/benchmarks/tpcc/data.py
@@ -0,0 +1,68 @@
+"""TPC-C's initial database (clause 4.3.3.1) for W warehouses, as rows of the
+records in tpcc.aup (each composite key packed into one number, as there),
+and the same rows for SQLite's tables with their own keys. Texts the
+transactions only carry along are left out (see tpcc.aup)."""
+import random
+import struct
+
+ITEMS, CUSTOMERS, ORDERS, DELIVERED = 100000, 3000, 3000, 2100
+
+# Each table's fields in record order (tpcc.aup), all numbers.
+FIELDS = {
+    'warehouses': ('id', 'tax', 'ytd'),
+    'districts': ('id', 'tax', 'ytd', 'next_order', 'next_delivery'),
+    'customers': ('id', 'discount', 'credit', 'balance', 'ytd_payment', 'payments', 'deliveries', 'last_order'),
+    'items': ('id', 'price'),
+    'stock': ('id', 'quantity', 'ytd', 'orders', 'remote'),
+    'orders': ('id', 'customer', 'entry', 'carrier', 'lines', 'local'),
+    'new_orders': ('id',),
+    'lines': ('id', 'item', 'supply', 'delivery', 'quantity', 'amount'),
+}
+
+
+def district_key(w, d):
+    return w * 10 + d
+
+
+def customer_key(w, d, c):
+    return district_key(w, d) * 3000 + c
+
+
+def order_key(w, d, o):
+    return district_key(w, d) * 10000000 + o
+
+
+def generate(warehouses, seed=1, now=1.7e12):
+    """The tables as lists of tuples in FIELDS order."""
+    rng = random.Random(seed)
+    tables = {name: [] for name in FIELDS}
+    tables['items'] = [(i, round(rng.uniform(1, 100), 2)) for i in range(1, ITEMS + 1)]
+    for w in range(1, warehouses + 1):
+        tables['warehouses'].append((w, round(rng.uniform(0, 0.2), 4), 300000.0))
+        tables['stock'].extend((w * 100000 + i, rng.randint(10, 100), 0, 0, 0) for i in range(1, ITEMS + 1))
+        for d in range(1, 11):
+            tables['districts'].append((district_key(w, d), round(rng.uniform(0, 0.2), 4), 30000.0, ORDERS + 1,
+                                        DELIVERED + 1))
+            owners = list(range(1, CUSTOMERS + 1))
+            rng.shuffle(owners)
+            last = {}
+            for o in range(1, ORDERS + 1):
+                key, count, delivered = order_key(w, d, o), rng.randint(5, 15), o <= DELIVERED
+                tables['orders'].append((key, customer_key(w, d, owners[o - 1]), now,
+                                         rng.randint(1, 10) if delivered else 0, count, 1))
+                last[owners[o - 1]] = key
+                if not delivered:
+                    tables['new_orders'].append((key,))
+                for n in range(1, count + 1):
+                    tables['lines'].append((key * 16 + n, rng.randint(1, ITEMS), w, now if delivered else 0, 5,
+                                            0 if delivered else round(rng.uniform(0.01, 9999.99), 2)))
+            for c in range(1, CUSTOMERS + 1):
+                tables['customers'].append((customer_key(w, d, c), round(rng.uniform(0, 0.5), 4),
+                                            0 if rng.random() < 0.1 else 1, -10.0, 10.0, 1, 0, last[c]))
+    return tables
+
+
+def packed(rows):
+    """Rows in a list's stored form (each number's binary64, little-endian)."""
+    width = len(rows[0])
+    return struct.pack(f'<{width * len(rows)}d', *(v for row in rows for v in row))
diff --git a/benchmarks/tpcc/run.py b/benchmarks/tpcc/run.py
new file mode 100644
index 0000000..f24126f
--- /dev/null
+++ b/benchmarks/tpcc/run.py
@@ -0,0 +1,332 @@
+#!/usr/bin/env python3
+"""TPC-C on Pure and on SQLite (docs/research/092-tpcc.md).
+
+tpcc.aup is TPC-C's tables and five transactions as an aup app; data.py makes
+the initial database of clause 4.3.3.1 for --warehouses warehouses. The same
+rows go into pure-serve (in bulk, as binary rows) and into SQLite (the
+specification's tables with their composite primary keys, WITHOUT ROWID,
+WAL), both kept in --directory. Then:
+
+1. The same transactions on both: one terminal, --check transactions drawn
+   from one seed by tpcc-load (over HTTP) and by sqlite-tpcc (in process),
+   and every row of every table compared (times aside), Pure's read through
+   queries registered for the purpose.
+2. Throughput: --seconds of the whole mix, Pure's from each count of
+   --terminals connections (tpcc-load on core 3, the server on cores 0-2),
+   SQLite's from each count of --sqlite-threads threads, each with its own
+   connection, on cores 0-2, synchronous --sync (FULL syncs each commit
+   before it is answered, as Pure does).
+3. The consistency conditions of clause 3.3.2 on both after each run.
+
+Prints one JSON line a step.
+"""
+import argparse
+import http.client
+import json
+import os
+import select
+import shutil
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[1]
+sys.path.insert(0, str(HERE))
+import data  # noqa: E402
+
+SCHEMA = '''
+CREATE TABLE warehouse (w_id INTEGER PRIMARY KEY, w_tax REAL, w_ytd REAL);
+CREATE TABLE district (d_w_id INTEGER, d_id INTEGER, d_tax REAL, d_ytd REAL, d_next_o_id INTEGER, d_next_delivery INTEGER,
+  PRIMARY KEY (d_w_id, d_id)) WITHOUT ROWID;
+CREATE TABLE customer (c_w_id INTEGER, c_d_id INTEGER, c_id INTEGER, c_discount REAL, c_credit INTEGER, c_balance REAL,
+  c_ytd_payment REAL, c_payment_cnt INTEGER, c_delivery_cnt INTEGER, c_last_order INTEGER,
+  PRIMARY KEY (c_w_id, c_d_id, c_id)) WITHOUT ROWID;
+CREATE TABLE item (i_id INTEGER PRIMARY KEY, i_price REAL);
+CREATE TABLE stock (s_w_id INTEGER, s_i_id INTEGER, s_quantity INTEGER, s_ytd INTEGER, s_order_cnt INTEGER,
+  s_remote_cnt INTEGER, PRIMARY KEY (s_w_id, s_i_id)) WITHOUT ROWID;
+CREATE TABLE orders (o_w_id INTEGER, o_d_id INTEGER, o_id INTEGER, o_c_id INTEGER, o_entry_d REAL, o_carrier_id INTEGER,
+  o_ol_cnt INTEGER, o_all_local INTEGER, PRIMARY KEY (o_w_id, o_d_id, o_id)) WITHOUT ROWID;
+CREATE TABLE new_order (no_w_id INTEGER, no_d_id INTEGER, no_o_id INTEGER,
+  PRIMARY KEY (no_w_id, no_d_id, no_o_id)) WITHOUT ROWID;
+CREATE TABLE order_line (ol_w_id INTEGER, ol_d_id INTEGER, ol_o_id INTEGER, ol_number INTEGER, ol_i_id INTEGER,
+  ol_supply_w_id INTEGER, ol_delivery_d REAL, ol_quantity INTEGER, ol_amount REAL,
+  PRIMARY KEY (ol_w_id, ol_d_id, ol_o_id, ol_number)) WITHOUT ROWID;
+CREATE TABLE history (h_c_w_id INTEGER, h_c_d_id INTEGER, h_c_id INTEGER, h_w_id INTEGER, h_d_id INTEGER, h_date REAL,
+  h_amount REAL);
+'''
+
+ORDERS = 10000000  # an order's key is its district's times this, plus its number
+
+
+def district_of(key):
+    """A district key's warehouse and district number."""
+    return (key - 1) // 10, (key - 1) % 10 + 1
+
+
+def sqlite_rows(tables):
+    """data.generate's rows as the rows of SQLite's tables."""
+    def customer(key):
+        return (*district_of((key - 1) // 3000), (key - 1) % 3000 + 1)
+
+    def order(key):
+        return (*district_of(key // ORDERS), key % ORDERS)
+
+    yield 'warehouse', tables['warehouses']
+    yield 'district', [(*district_of(k), tax, ytd, next_order, next_delivery)
+                       for k, tax, ytd, next_order, next_delivery in tables['districts']]
+    yield 'customer', [(*customer(k), discount, credit, balance, ytd, payments, deliveries, last % ORDERS)
+                       for k, discount, credit, balance, ytd, payments, deliveries, last in tables['customers']]
+    yield 'item', tables['items']
+    yield 'stock', [((k - 1) // 100000, (k - 1) % 100000 + 1, *rest) for k, *rest in tables['stock']]
+    yield 'orders', [(*order(k), (c - 1) % 3000 + 1, *rest) for k, c, *rest in tables['orders']]
+    yield 'new_order', [order(k) for k, in tables['new_orders']]
+    yield 'order_line', [(*order(k // 16), k % 16, *rest) for k, *rest in tables['lines']]
+
+
+def load_sqlite(path, tables):
+    db = sqlite3.connect(str(path))
+    db.executescript('PRAGMA journal_mode=WAL; PRAGMA synchronous=OFF;' + SCHEMA)
+    with db:
+        for table, rows in sqlite_rows(tables):
+            db.executemany(f'INSERT INTO {table} VALUES ({", ".join("?" * len(rows[0]))})', rows)
+    db.execute('PRAGMA wal_checkpoint(TRUNCATE)')
+    db.close()
+
+
+class Pure:
+    """pure-serve on cores 0-2 with the app's store in directory."""
+
+    def __init__(self, serve, module, store, threads):
+        self.process = subprocess.Popen(['taskset', '-c', '0-2', str(serve), '--module', str(module), '--store', str(store),
+                                         '--port', '0', '--threads', str(threads), '--timeout-ms', '600000'],
+                                        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        select.select([self.process.stdout], [], [], 30)
+        self.port = json.loads(self.process.stdout.readline())['port']
+        self.connection = http.client.HTTPConnection('127.0.0.1', self.port, timeout=600)
+
+    def call(self, method, path, body=None, kind='application/json'):
+        self.connection.request(method, path, body, {'Content-Type': kind} if body is not None else {})
+        answer = self.connection.getresponse()
+        return answer.status, answer.read()
+
+    def view(self, name):
+        status, body = self.call('GET', '/views/' + name)
+        assert status == 200, (name, status, body[:200])
+        return json.loads(body)
+
+    def load(self, tables):
+        for name, rows in tables.items():
+            if name == 'history':
+                continue
+            chunk = max(1, (1400000 // 8) // len(data.FIELDS[name]) - 1)
+            for at in range(0, len(rows), chunk):
+                status, body = self.call('POST', f'/inputs/load_{name}', data.packed(rows[at:at + chunk]),
+                                         'application/octet-stream')
+                assert status == 200, (name, status, body[:200])
+
+    def cpu_seconds(self):
+        fields = Path(f'/proc/{self.process.pid}/stat').read_text().rsplit(')', 1)[1].split()
+        return (int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK')
+
+    def resident_mib(self):
+        for line in Path(f'/proc/{self.process.pid}/status').read_text().splitlines():
+            if line.startswith('VmRSS:'):
+                return round(int(line.split()[1]) / 1024, 1)
+        return None
+
+    def stop(self):
+        self.connection.close()
+        self.process.terminate()
+        self.process.communicate(timeout=120)
+
+
+# Each table's rows, times aside, as Pure's queries and SQLite's SELECTs name them (keys packed as tpcc.aup packs them).
+DUMPS = {
+    'warehouses': ('each w in warehouses {\n  id = w.id\n  ytd = w.ytd\n}',
+                   'SELECT w_id, w_ytd FROM warehouse'),
+    'districts': ('each d in districts {\n  id = d.id\n  ytd = d.ytd\n  next_order = d.next_order\n'
+                  '  next_delivery = d.next_delivery\n}',
+                  'SELECT d_w_id * 10 + d_id, d_ytd, d_next_o_id, d_next_delivery FROM district'),
+    'customers': ('each c in customers {\n  id = c.id\n  balance = c.balance\n  ytd_payment = c.ytd_payment\n'
+                  '  payments = c.payments\n  deliveries = c.deliveries\n  last_order = c.last_order\n}',
+                  'SELECT (c_w_id * 10 + c_d_id) * 3000 + c_id, c_balance, c_ytd_payment, c_payment_cnt, c_delivery_cnt, '
+                  '(c_w_id * 10 + c_d_id) * 10000000 + c_last_order FROM customer'),
+    'stock': ('each s in stock {\n  id = s.id\n  quantity = s.quantity\n  ytd = s.ytd\n  orders = s.orders\n'
+              '  remote = s.remote\n}',
+              'SELECT s_w_id * 100000 + s_i_id, s_quantity, s_ytd, s_order_cnt, s_remote_cnt FROM stock'),
+    'orders': ('each o in orders {\n  id = o.id\n  customer = o.customer\n  carrier = o.carrier\n  lines = o.lines\n'
+               '  local = o.local\n}',
+               'SELECT (o_w_id * 10 + o_d_id) * 10000000 + o_id, (o_w_id * 10 + o_d_id) * 3000 + o_c_id, o_carrier_id, '
+               'o_ol_cnt, o_all_local FROM orders'),
+    'new_orders': ('each n in new_orders {\n  id = n.id\n}',
+                   'SELECT (no_w_id * 10 + no_d_id) * 10000000 + no_o_id FROM new_order'),
+    'lines': ('each l in lines {\n  id = l.id\n  item = l.item\n  supply = l.supply\n  quantity = l.quantity\n'
+              '  amount = l.amount\n  delivery = l.delivery\n}',
+              'SELECT ((ol_w_id * 10 + ol_d_id) * 10000000 + ol_o_id) * 16 + ol_number, ol_i_id, ol_supply_w_id, '
+              'ol_quantity, ol_amount, ol_delivery_d FROM order_line'),
+    'history': ('each h in history {\n  customer = h.customer\n  district = h.district\n  amount = h.amount\n}',
+                'SELECT (h_c_w_id * 10 + h_c_d_id) * 3000 + h_c_id, h_w_id * 10 + h_d_id, h_amount FROM history'),
+}
+
+# Clause 3.3.2.2 and 3.3.2.3, a district a row: its next order, its newest order, and its oldest and newest new orders
+# and how many (orders and new orders by their numbers in the district).
+DISTRICTS = ('each d in districts {\n  id = d.id\n  next_order = d.next_order\n'
+             '  newest = max(o.id for o in orders[d.id * 10000000 .. d.id * 10000000 + 10000000]) - d.id * 10000000\n'
+             '  newest_new = max(n.id for n in new_orders[d.id * 10000000 .. d.id * 10000000 + 10000000])'
+             ' - d.id * 10000000\n'
+             '  oldest_new = min(n.id for n in new_orders[d.id * 10000000 .. d.id * 10000000 + 10000000])'
+             ' - d.id * 10000000\n'
+             '  new_orders = count(n in new_orders[d.id * 10000000 .. d.id * 10000000 + 10000000])\n}',
+             'SELECT d_w_id * 10 + d_id, d_next_o_id, '
+             '(SELECT max(o_id) FROM orders WHERE o_w_id = d_w_id AND o_d_id = d_id), '
+             '(SELECT max(no_o_id) FROM new_order WHERE no_w_id = d_w_id AND no_d_id = d_id), '
+             '(SELECT min(no_o_id) FROM new_order WHERE no_w_id = d_w_id AND no_d_id = d_id), '
+             '(SELECT count(*) FROM new_order WHERE no_w_id = d_w_id AND no_d_id = d_id) FROM district')
+
+
+def register(pure, aup, work, name, view):
+    source = work / f'{name}.aup'
+    source.write_text(f'view {name} = {view}\n')
+    subprocess.run([str(aup), 'query', str(HERE / 'tpcc.aup'), str(source), '-o', str(work / f'{name}.pmod')], check=True)
+    status, body = pure.call('PUT', f'/queries/{name}', (work / f'{name}.pmod').read_bytes(), 'application/octet-stream')
+    assert status == 200, (name, status, body[:200])
+
+
+def asked(pure, name):
+    status, body = pure.call('GET', f'/queries/{name}')
+    assert status == 200, (name, status, body[:200])
+    return [tuple(row.values()) for row in json.loads(body)['items']]
+
+
+def same(a, b):
+    return a == b or abs(a - b) <= 1e-9 * max(1.0, abs(a), abs(b))
+
+
+def compare(table, pure_rows, sqlite_rows_):
+    """Rows that differ between the two (a table's rows in key order; history's sorted)."""
+    if table in ('lines',):  # a line's delivery time differs; whether it has one must not
+        pure_rows = [(*r[:-1], r[-1] > 0) for r in pure_rows]
+        sqlite_rows_ = [(*r[:-1], r[-1] > 0) for r in sqlite_rows_]
+    pure_rows, sqlite_rows_ = sorted(pure_rows), sorted(sqlite_rows_)
+    if len(pure_rows) != len(sqlite_rows_):
+        return [('count', len(pure_rows), len(sqlite_rows_))]
+    return [(p, s) for p, s in zip(pure_rows, sqlite_rows_) if not all(same(x, y) for x, y in zip(p, s))]
+
+
+def consistency(pure, db):
+    """Clause 3.3.2.1-4 on both: the conditions each fails (none when consistent)."""
+    failures = []
+    for system, ytd, district_ytd, districts, ordered, lines in (
+            ('pure', pure.view('warehouse_ytd'), pure.view('district_ytd'), asked(pure, 'district_check'),
+             pure.view('ordered_lines'), pure.view('line_count')),
+            ('sqlite', *db.execute('SELECT (SELECT sum(w_ytd) FROM warehouse), (SELECT sum(d_ytd) FROM district)').fetchone(),
+             db.execute(DISTRICTS[1]).fetchall(), *db.execute(
+                 'SELECT (SELECT sum(o_ol_cnt) FROM orders), (SELECT count(*) FROM order_line)').fetchone())):
+        if not same(ytd, district_ytd):
+            failures.append((system, '3.3.2.1', ytd, district_ytd))
+        for key, next_order, newest, newest_new, oldest_new, new_orders in districts:
+            if not next_order - 1 == newest == newest_new:
+                failures.append((system, '3.3.2.2', key, next_order, newest, newest_new))
+            if newest_new - oldest_new + 1 != new_orders:
+                failures.append((system, '3.3.2.3', key, oldest_new, newest_new, new_orders))
+        if ordered != lines:
+            failures.append((system, '3.3.2.4', ordered, lines))
+    return failures
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--aup', type=Path, default=ROOT.parent / 'aup' / 'build' / 'aup')
+    parser.add_argument('--serve', type=Path, default=ROOT / 'build' / 'pure-serve')
+    parser.add_argument('--directory', type=Path, default=Path('/dev/shm'), help='where the store and database live')
+    parser.add_argument('--warehouses', type=int, default=1)
+    parser.add_argument('--seed', type=int, default=1)
+    parser.add_argument('--check', type=int, default=20000, help='transactions run on both and compared (0: none)')
+    parser.add_argument('--seconds', type=float, default=20)
+    parser.add_argument('--terminals', default='16,64', help="Pure's connections, a run each")
+    parser.add_argument('--threads', default='3', help="pure-serve's --threads")
+    parser.add_argument('--sqlite-threads', default='1,4,16', help="SQLite's threads, a run each")
+    parser.add_argument('--sync', default='full', choices=('full', 'normal', 'off'))
+    args = parser.parse_args()
+    work = Path(tempfile.mkdtemp(prefix='pure-tpcc-'))
+    place = Path(tempfile.mkdtemp(prefix='pure-tpcc-', dir=args.directory))
+    pure = None
+    try:
+        module = work / 'tpcc.pmod'
+        subprocess.run([str(args.aup), 'build', str(HERE / 'tpcc.aup'), '-o', str(module)], check=True)
+        subprocess.run(['cc', '-O2', '-o', str(work / 'tpcc-load'), str(HERE / 'tpcc-load.c')], check=True)
+        subprocess.run(['cc', '-O2', '-o', str(work / 'sqlite-tpcc'), str(HERE / 'sqlite-tpcc.c'), '-lsqlite3', '-lpthread'],
+                       check=True)
+        began = time.perf_counter()
+        tables = data.generate(args.warehouses, args.seed)
+        rows = sum(len(t) for t in tables.values())
+        generated = time.perf_counter() - began
+        pure = Pure(args.serve, module, place / 'store', args.threads)
+        began = time.perf_counter()
+        pure.load(tables)
+        pure_load = time.perf_counter() - began
+        database = place / 'tpcc.db'
+        began = time.perf_counter()
+        load_sqlite(database, tables)
+        sqlite_load = time.perf_counter() - began
+        counts = {'order_count': len(tables['orders']), 'new_order_count': len(tables['new_orders']),
+                  'line_count': len(tables['lines'])}
+        assert {v: pure.view(v) for v in counts} == counts
+        print(json.dumps({'step': 'load', 'directory': str(args.directory), 'warehouses': args.warehouses, 'rows': rows,
+                          'generate_s': round(generated, 2), 'pure_load_s': round(pure_load, 2),
+                          'sqlite_load_s': round(sqlite_load, 2), 'pure_resident_mib': pure.resident_mib(),
+                          'sqlite_file_mib': round(database.stat().st_size / 2**20, 1)}), flush=True)
+        db = sqlite3.connect(str(database))
+        register(pure, args.aup, work, 'district_check', DISTRICTS[0])
+        if args.check:
+            pure_run = json.loads(subprocess.run(
+                [str(work / 'tpcc-load'), str(pure.port), '1', '3600', str(args.warehouses), str(args.seed),
+                 str(args.check)], capture_output=True, text=True, check=True).stdout)
+            sqlite_run = json.loads(subprocess.run(
+                [str(work / 'sqlite-tpcc'), str(database), str(args.warehouses), '1', '3600', args.sync, str(args.seed),
+                 str(args.check)], capture_output=True, text=True, check=True).stdout)
+            differing = {}
+            for table, (view, sql) in DUMPS.items():
+                register(pure, args.aup, work, f'dump_{table}', view)
+                found = compare(table, asked(pure, f'dump_{table}'), db.execute(sql).fetchall())
+                if found:
+                    differing[table] = [len(found), found[:3]]
+            kinds = ('new_order', 'payment', 'order_status', 'delivery', 'stock_level')
+            print(json.dumps({'step': 'same transactions', 'transactions': args.check,
+                              'pure': {k: pure_run[k]['done'] for k in kinds},
+                              'sqlite': {k: sqlite_run[k]['done'] for k in kinds},
+                              'undone_new_orders': [pure_run['undone_new_orders'], sqlite_run['undone_new_orders']],
+                              'failed': [sum(pure_run[k]['failed'] for k in kinds),
+                                         sum(sqlite_run[k]['failed'] for k in kinds)],
+                              'differing': differing, 'consistency_failures': consistency(pure, db)}), flush=True)
+        for terminals in (int(t) for t in args.terminals.split(',')):
+            used = pure.cpu_seconds()
+            result = json.loads(subprocess.run(
+                ['taskset', '-c', '3', str(work / 'tpcc-load'), str(pure.port), str(terminals), str(args.seconds),
+                 str(args.warehouses), str(args.seed + terminals)], capture_output=True, text=True, check=True).stdout)
+            used = pure.cpu_seconds() - used
+            done = sum(result[k]['done'] for k in ('new_order', 'payment', 'order_status', 'delivery', 'stock_level'))
+            print(json.dumps({'step': 'run', **result, 'server_threads': args.threads,
+                              'cpu_us_per_transaction': round(used / max(done, 1) * 1e6, 1),
+                              'resident_mib': pure.resident_mib(), 'consistency_failures': consistency(pure, db)}),
+                  flush=True)
+        for threads in (int(t) for t in args.sqlite_threads.split(',')):
+            result = json.loads(subprocess.run(
+                ['taskset', '-c', '0-2', str(work / 'sqlite-tpcc'), str(database), str(args.warehouses), str(threads),
+                 str(args.seconds), args.sync, str(args.seed + threads)], capture_output=True, text=True,
+                check=True).stdout)
+            print(json.dumps({'step': 'run', **result, 'consistency_failures': consistency(pure, db)}), flush=True)
+        db.close()
+    finally:
+        if pure:
+            pure.stop()
+        shutil.rmtree(place, ignore_errors=True)
+        shutil.rmtree(work, ignore_errors=True)
+
+
+if __name__ == '__main__':
+    main()
diff --git a/benchmarks/tpcc/sqlite-tpcc.c b/benchmarks/tpcc/sqlite-tpcc.c
new file mode 100644
index 0000000..62fb465
--- /dev/null
+++ b/benchmarks/tpcc/sqlite-tpcc.c
@@ -0,0 +1,389 @@
+#define _GNU_SOURCE
+/* TPC-C's five transactions on SQLite, the same work tpcc.aup does
+ * (benchmarks/tpcc/run.py builds the database: TPC-C's tables with their
+ * composite primary keys, WITHOUT ROWID, and TPC-C's initial rows):
+ *
+ *     sqlite-tpcc DATABASE WAREHOUSES THREADS SECONDS full|normal|off [SEED [TRANSACTIONS]]
+ *
+ * THREADS terminals, each with its own connection and a home warehouse,
+ * run the mix of clause 5.2.3 with the inputs of clause 2 for SECONDS, each
+ * transaction BEGIN IMMEDIATE ... COMMIT (WAL, synchronous as given).
+ * Departures as in tpcc.aup: customers by number, no texts carried along,
+ * a customer's last order kept in c_last_order, Delivery a district a
+ * transaction. With TRANSACTIONS each terminal stops after that many; the
+ * inputs are drawn in tpcc-load's order. Prints one JSON line as tpcc-load
+ * does. */
+#include <pthread.h>
+#include <sqlite3.h>
+#include <stdatomic.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+
+enum { BUCKETS = 64 * 16, KINDS = 5 };
+enum { NEW_ORDER, PAYMENT, ORDER_STATUS, DELIVERY, STOCK_LEVEL };
+static const char *const names[KINDS] = {"new_order", "payment", "order_status", "delivery", "stock_level"};
+
+static const char *database, *synchronous;
+static int warehouses;
+static double seconds;
+static uint64_t seed, each;
+static atomic_uint_fast64_t histogram[KINDS][BUCKETS], done[KINDS], failed[KINDS], undone;
+
+static uint64_t now_us(void) {
+    struct timespec t;
+    clock_gettime(CLOCK_MONOTONIC, &t);
+    return (uint64_t)t.tv_sec * 1000000 + (uint64_t)t.tv_nsec / 1000;
+}
+static unsigned bucket(uint64_t us) {
+    if (us < 16) return (unsigned)us;
+    unsigned power = 63 - (unsigned)__builtin_clzll(us);
+    unsigned index = (power - 3) * 16 + (unsigned)((us >> (power - 4)) & 15);
+    return index < BUCKETS ? index : BUCKETS - 1;
+}
+static uint64_t bucket_value(unsigned index) {
+    if (index < 16) return index;
+    unsigned power = index / 16 + 3;
+    return ((uint64_t)16 + index % 16) << (power - 4);
+}
+static uint64_t percentile(atomic_uint_fast64_t *counts, uint64_t total, double p) {
+    uint64_t want = (uint64_t)((double)total * p), seen = 0;
+    for (unsigned i = 0; i < BUCKETS; ++i) {
+        seen += counts[i];
+        if (seen > want) return bucket_value(i);
+    }
+    return 0;
+}
+
+static uint64_t next_random(uint64_t *state) {
+    uint64_t x = *state;
+    x ^= x << 13, x ^= x >> 7, x ^= x << 17;
+    return *state = x;
+}
+static int uniform(uint64_t *state, int low, int high) { return low + (int)(next_random(state) % (uint64_t)(high - low + 1)); }
+static int nurand(uint64_t *state, int a, int low, int high, int c) {
+    return ((uniform(state, 0, a) | uniform(state, low, high)) + c) % (high - low + 1) + low;
+}
+
+static sqlite3_stmt *prepare(sqlite3 *db, const char *sql) {
+    sqlite3_stmt *statement;
+    if (sqlite3_prepare_v2(db, sql, -1, &statement, NULL) != SQLITE_OK) {
+        fprintf(stderr, "%s: %s\n", sql, sqlite3_errmsg(db));
+        exit(1);
+    }
+    return statement;
+}
+
+/* A statement stepped without error: run to its end and reset (so it can
+ * be bound again), or, given row, stepped once for its first row, which the
+ * caller reads before resetting it. The first few errors are shown. */
+static atomic_uint shown;
+static bool step(sqlite3_stmt *statement, bool *row) {
+    int result = sqlite3_step(statement);
+    if (row) *row = result == SQLITE_ROW;
+    while (result == SQLITE_ROW && !row) result = sqlite3_step(statement);
+    bool ok = result == SQLITE_DONE || result == SQLITE_ROW;
+    if (!ok && shown++ < 5) fprintf(stderr, "%s: %s\n", sqlite3_sql(statement), sqlite3_errmsg(sqlite3_db_handle(statement)));
+    if (!row) sqlite3_reset(statement);
+    return ok;
+}
+
+typedef struct {
+    sqlite3 *db;
+    sqlite3_stmt *begin, *begin_read, *commit, *rollback;
+    sqlite3_stmt *district_next, *district_advance, *customer_of, *item_price, *stock_of, *stock_take, *line_add, *order_add,
+                 *new_order_add, *customer_last;
+    sqlite3_stmt *warehouse_pay, *district_pay, *customer_pay, *history_add;
+    sqlite3_stmt *customer_status, *lines_of;
+    sqlite3_stmt *district_delivery, *delivery_advance, *new_order_remove, *order_carrier, *order_customer, *lines_deliver,
+                 *lines_total, *customer_deliver;
+    sqlite3_stmt *district_order, *stock_level;
+} Terminal;
+
+static void open_terminal(Terminal *t) {
+    char pragmas[128];
+    if (sqlite3_open(database, &t->db) != SQLITE_OK) { fprintf(stderr, "%s\n", sqlite3_errmsg(t->db)); exit(1); }
+    sqlite3_busy_timeout(t->db, 60000);
+    snprintf(pragmas, sizeof(pragmas), "PRAGMA journal_mode=WAL; PRAGMA synchronous=%s;", synchronous);
+    if (sqlite3_exec(t->db, pragmas, NULL, NULL, NULL) != SQLITE_OK) { fprintf(stderr, "%s\n", sqlite3_errmsg(t->db)); exit(1); }
+    sqlite3 *db = t->db;
+    t->begin = prepare(db, "BEGIN IMMEDIATE");
+    t->begin_read = prepare(db, "BEGIN");
+    t->commit = prepare(db, "COMMIT");
+    t->rollback = prepare(db, "ROLLBACK");
+    t->district_next = prepare(db, "SELECT d_next_o_id FROM district WHERE d_w_id = ?1 AND d_id = ?2");
+    t->district_advance = prepare(db, "UPDATE district SET d_next_o_id = d_next_o_id + 1 WHERE d_w_id = ?1 AND d_id = ?2");
+    t->customer_of = prepare(db, "SELECT c_id FROM customer WHERE c_w_id = ?1 AND c_d_id = ?2 AND c_id = ?3");
+    t->item_price = prepare(db, "SELECT i_price FROM item WHERE i_id = ?1");
+    t->stock_of = prepare(db, "SELECT s_quantity FROM stock WHERE s_w_id = ?1 AND s_i_id = ?2");
+    t->stock_take = prepare(db, "UPDATE stock SET s_quantity = ?3, s_ytd = s_ytd + ?4, s_order_cnt = s_order_cnt + 1, "
+                                "s_remote_cnt = s_remote_cnt + ?5 WHERE s_w_id = ?1 AND s_i_id = ?2");
+    t->line_add = prepare(db, "INSERT INTO order_line (ol_w_id, ol_d_id, ol_o_id, ol_number, ol_i_id, ol_supply_w_id, "
+                              "ol_delivery_d, ol_quantity, ol_amount) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0, ?7, ?8)");
+    t->order_add = prepare(db, "INSERT INTO orders (o_w_id, o_d_id, o_id, o_c_id, o_entry_d, o_carrier_id, o_ol_cnt, o_all_local) "
+                               "VALUES (?1, ?2, ?3, ?4, ?5, 0, ?6, ?7)");
+    t->new_order_add = prepare(db, "INSERT INTO new_order (no_w_id, no_d_id, no_o_id) VALUES (?1, ?2, ?3)");
+    t->customer_last = prepare(db, "UPDATE customer SET c_last_order = ?4 WHERE c_w_id = ?1 AND c_d_id = ?2 AND c_id = ?3");
+    t->warehouse_pay = prepare(db, "UPDATE warehouse SET w_ytd = w_ytd + ?2 WHERE w_id = ?1");
+    t->district_pay = prepare(db, "UPDATE district SET d_ytd = d_ytd + ?3 WHERE d_w_id = ?1 AND d_id = ?2");
+    t->customer_pay = prepare(db, "UPDATE customer SET c_balance = c_balance - ?4, c_ytd_payment = c_ytd_payment + ?4, "
+                                  "c_payment_cnt = c_payment_cnt + 1 WHERE c_w_id = ?1 AND c_d_id = ?2 AND c_id = ?3");
+    t->history_add = prepare(db, "INSERT INTO history (h_c_w_id, h_c_d_id, h_c_id, h_w_id, h_d_id, h_date, h_amount) "
+                                 "VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)");
+    t->customer_status = prepare(db, "SELECT c_balance, c_last_order FROM customer WHERE c_w_id = ?1 AND c_d_id = ?2 AND c_id = ?3");
+    t->lines_of = prepare(db, "SELECT ol_i_id, ol_supply_w_id, ol_quantity, ol_amount, ol_delivery_d FROM order_line "
+                              "WHERE ol_w_id = ?1 AND ol_d_id = ?2 AND ol_o_id = ?3");
+    t->district_delivery = prepare(db, "SELECT d_next_delivery, d_next_o_id FROM district WHERE d_w_id = ?1 AND d_id = ?2");
+    t->delivery_advance = prepare(db, "UPDATE district SET d_next_delivery = d_next_delivery + 1 WHERE d_w_id = ?1 AND d_id = ?2");
+    t->new_order_remove = prepare(db, "DELETE FROM new_order WHERE no_w_id = ?1 AND no_d_id = ?2 AND no_o_id = ?3");
+    t->order_carrier = prepare(db, "UPDATE orders SET o_carrier_id = ?4 WHERE o_w_id = ?1 AND o_d_id = ?2 AND o_id = ?3");
+    t->order_customer = prepare(db, "SELECT o_c_id FROM orders WHERE o_w_id = ?1 AND o_d_id = ?2 AND o_id = ?3");
+    t->lines_deliver = prepare(db, "UPDATE order_line SET ol_delivery_d = ?4 WHERE ol_w_id = ?1 AND ol_d_id = ?2 AND ol_o_id = ?3");
+    t->lines_total = prepare(db, "SELECT sum(ol_amount) FROM order_line WHERE ol_w_id = ?1 AND ol_d_id = ?2 AND ol_o_id = ?3");
+    t->customer_deliver = prepare(db, "UPDATE customer SET c_balance = c_balance + ?4, c_delivery_cnt = c_delivery_cnt + 1 "
+                                      "WHERE c_w_id = ?1 AND c_d_id = ?2 AND c_id = ?3");
+    t->district_order = prepare(db, "SELECT d_next_o_id FROM district WHERE d_w_id = ?1 AND d_id = ?2");
+    t->stock_level = prepare(db, "SELECT count(DISTINCT s_i_id) FROM order_line, stock WHERE ol_w_id = ?1 AND ol_d_id = ?2 "
+                                 "AND ol_o_id < ?3 AND ol_o_id >= ?3 - 20 AND s_w_id = ?1 AND s_i_id = ol_i_id AND s_quantity < ?4");
+}
+
+static double now_ms(void) {
+    struct timespec t;
+    clock_gettime(CLOCK_REALTIME, &t);
+    return (double)t.tv_sec * 1e3 + (double)(t.tv_nsec / 1000000);
+}
+
+#define RUN(s) step(s, NULL)
+
+/* A read-only transaction: Order-Status or Stock-Level, each on one snapshot. */
+static bool read_only(Terminal *t, int kind, int w, int d, uint64_t *r) {
+    bool row;
+    if (kind == ORDER_STATUS) {
+        sqlite3_stmt *s = t->customer_status;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int(s, 3, nurand(r, 1023, 1, 3000, 259));
+        if (!step(s, &row) || !row) return false;
+        sqlite3_int64 order = sqlite3_column_int64(s, 1);
+        (void)sqlite3_column_double(s, 0);
+        sqlite3_reset(s);
+        s = t->lines_of;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, order);
+        int result;
+        while ((result = sqlite3_step(s)) == SQLITE_ROW) (void)sqlite3_column_double(s, 3);
+        sqlite3_reset(s);
+        return result == SQLITE_DONE;
+    }
+    sqlite3_stmt *s = t->district_order;
+    sqlite3_reset(s);
+    sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d);
+    if (!step(s, &row) || !row) return false;
+    sqlite3_int64 next = sqlite3_column_int64(s, 0);
+    sqlite3_reset(s);
+    s = t->stock_level;
+    sqlite3_reset(s);
+    sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, next), sqlite3_bind_int(s, 4, uniform(r, 10, 20));
+    bool ok = step(s, &row) && row;
+    if (ok) (void)sqlite3_column_int64(s, 0);
+    sqlite3_reset(s);
+    return ok;
+}
+
+/* One transaction of kind; false when it failed (an unused item's New-Order is undone and counts in undone). */
+static bool transact(Terminal *t, int kind, int w, int d, uint64_t *r) {
+    bool row;
+    if (kind == ORDER_STATUS || kind == STOCK_LEVEL) {
+        if (!RUN(t->begin_read)) return false;
+        bool ok = read_only(t, kind, w, d, r);
+        return RUN(t->commit) && ok;
+    }
+    if (!RUN(t->begin)) return false;
+    bool ok = true;
+    if (kind == NEW_ORDER) {
+        int c = nurand(r, 1023, 1, 3000, 259), count = uniform(r, 5, 15), local = 1;
+        bool unused = uniform(r, 1, 100) == 1;
+        sqlite3_stmt *s = t->district_next;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d);
+        ok = step(s, &row) && row;
+        sqlite3_int64 order = ok ? sqlite3_column_int64(s, 0) : 0;
+        sqlite3_reset(s);
+        sqlite3_bind_int(t->district_advance, 1, w), sqlite3_bind_int(t->district_advance, 2, d);
+        ok = ok && RUN(t->district_advance);
+        s = t->customer_of;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int(s, 3, c);
+        ok = ok && step(s, &row) && row;
+        sqlite3_reset(s);
+        for (int n = 1; ok && n <= count; ++n) {
+            int item = unused && n == count ? 100001 : nurand(r, 8191, 1, 100000, 7911);
+            int supply = warehouses > 1 && uniform(r, 1, 100) == 1 ? uniform(r, 1, warehouses) : w, quantity = uniform(r, 1, 10);
+            s = t->item_price;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, item);
+            if (!step(s, &row) || !row) { /* an unused item: the order is undone */
+                sqlite3_reset(s);
+                RUN(t->rollback);
+                if (unused) ++undone;
+                return unused;
+            }
+            double price = sqlite3_column_double(s, 0);
+            sqlite3_reset(s);
+            s = t->stock_of;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, supply), sqlite3_bind_int(s, 2, item);
+            ok = step(s, &row) && row;
+            int stocked = ok ? sqlite3_column_int(s, 0) : 0;
+            sqlite3_reset(s);
+            int remote = supply != w;
+            if (remote) local = 0;
+            s = t->stock_take;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, supply), sqlite3_bind_int(s, 2, item);
+            sqlite3_bind_int(s, 3, stocked >= quantity + 10 ? stocked - quantity : stocked - quantity + 91);
+            sqlite3_bind_int(s, 4, quantity), sqlite3_bind_int(s, 5, remote);
+            ok = ok && RUN(s);
+            s = t->line_add;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, order), sqlite3_bind_int(s, 4, n);
+            sqlite3_bind_int(s, 5, item), sqlite3_bind_int(s, 6, supply), sqlite3_bind_int(s, 7, quantity);
+            sqlite3_bind_double(s, 8, quantity * price);
+            ok = ok && RUN(s);
+        }
+        s = t->order_add;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, order), sqlite3_bind_int(s, 4, c);
+        sqlite3_bind_double(s, 5, now_ms()), sqlite3_bind_int(s, 6, count), sqlite3_bind_int(s, 7, local);
+        ok = ok && RUN(s);
+        s = t->new_order_add;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, order);
+        ok = ok && RUN(s);
+        s = t->customer_last;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int(s, 3, c), sqlite3_bind_int64(s, 4, order);
+        ok = ok && RUN(s);
+    } else if (kind == PAYMENT) {
+        int cw = w, cd = d, c = nurand(r, 1023, 1, 3000, 259);
+        if (warehouses > 1 && uniform(r, 1, 100) > 85) {
+            do cw = uniform(r, 1, warehouses); while (cw == w);
+            cd = uniform(r, 1, 10);
+        }
+        double amount = uniform(r, 100, 500000) / 100.0;
+        sqlite3_stmt *s = t->warehouse_pay;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_double(s, 2, amount);
+        ok = RUN(s);
+        s = t->district_pay;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_double(s, 3, amount);
+        ok = ok && RUN(s);
+        s = t->customer_pay;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, cw), sqlite3_bind_int(s, 2, cd), sqlite3_bind_int(s, 3, c), sqlite3_bind_double(s, 4, amount);
+        ok = ok && RUN(s) && sqlite3_changes(t->db) == 1;
+        s = t->history_add;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, cw), sqlite3_bind_int(s, 2, cd), sqlite3_bind_int(s, 3, c), sqlite3_bind_int(s, 4, w);
+        sqlite3_bind_int(s, 5, d), sqlite3_bind_double(s, 6, now_ms()), sqlite3_bind_double(s, 7, amount);
+        ok = ok && RUN(s);
+    } else { /* DELIVERY, one district */
+        int carrier = uniform(r, 1, 10);
+        sqlite3_stmt *s = t->district_delivery;
+        sqlite3_reset(s);
+        sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d);
+        ok = step(s, &row) && row;
+        sqlite3_int64 order = ok ? sqlite3_column_int64(s, 0) : 0, next = ok ? sqlite3_column_int64(s, 1) : 0;
+        sqlite3_reset(s);
+        if (ok && order < next) {
+            sqlite3_bind_int(t->delivery_advance, 1, w), sqlite3_bind_int(t->delivery_advance, 2, d);
+            ok = RUN(t->delivery_advance);
+            s = t->new_order_remove;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, order);
+            ok = ok && RUN(s) && sqlite3_changes(t->db) == 1;
+            s = t->order_carrier;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, order), sqlite3_bind_int(s, 4, carrier);
+            ok = ok && RUN(s);
+            s = t->order_customer;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, order);
+            ok = ok && step(s, &row) && row;
+            int c = ok ? sqlite3_column_int(s, 0) : 0;
+            sqlite3_reset(s);
+            s = t->lines_deliver;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, order), sqlite3_bind_double(s, 4, now_ms());
+            ok = ok && RUN(s);
+            s = t->lines_total;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int64(s, 3, order);
+            ok = ok && step(s, &row) && row;
+            double total = ok ? sqlite3_column_double(s, 0) : 0;
+            sqlite3_reset(s);
+            s = t->customer_deliver;
+            sqlite3_reset(s);
+            sqlite3_bind_int(s, 1, w), sqlite3_bind_int(s, 2, d), sqlite3_bind_int(s, 3, c), sqlite3_bind_double(s, 4, total);
+            ok = ok && RUN(s);
+        }
+    }
+    if (ok && RUN(t->commit)) return true;
+    RUN(t->rollback);
+    return false;
+}
+
+static void *terminal(void *argument) {
+    int index = (int)(uintptr_t)argument;
+    uint64_t random = 0x9e3779b97f4a7c15u * (uint64_t)(index + 1) ^ seed * 0xbf58476d1ce4e5b9u;
+    Terminal t;
+    open_terminal(&t);
+    int home = index % warehouses + 1;
+    uint64_t stop = now_us() + (uint64_t)(seconds * 1e6);
+    for (uint64_t left = each; left && now_us() < stop; --left) {
+        int d = uniform(&random, 1, 10), roll = uniform(&random, 1, 100);
+        int kind = roll <= 45 ? NEW_ORDER : roll <= 88 ? PAYMENT : roll <= 92 ? ORDER_STATUS : roll <= 96 ? DELIVERY : STOCK_LEVEL;
+        uint64_t started = now_us();
+        bool ok = transact(&t, kind, home, d, &random);
+        ++done[kind];
+        if (!ok) ++failed[kind];
+        ++histogram[kind][bucket(now_us() - started)];
+    }
+    for (sqlite3_stmt *s; (s = sqlite3_next_stmt(t.db, NULL));) sqlite3_finalize(s);
+    sqlite3_close(t.db);
+    return NULL;
+}
+
+int main(int argc, char **argv) {
+    if (argc < 6) {
+        fputs("sqlite-tpcc DATABASE WAREHOUSES THREADS SECONDS full|normal|off [SEED [TRANSACTIONS]]\n", stderr);
+        return 2;
+    }
+    database = argv[1];
+    warehouses = atoi(argv[2]);
+    int threads = atoi(argv[3]);
+    seconds = atof(argv[4]);
+    synchronous = argv[5];
+    seed = argc > 6 ? strtoull(argv[6], NULL, 10) : 1;
+    each = argc > 7 ? strtoull(argv[7], NULL, 10) : UINT64_MAX;
+    pthread_t *thread = calloc((size_t)threads, sizeof(*thread));
+    uint64_t start = now_us();
+    for (int i = 0; i < threads; ++i) pthread_create(&thread[i], NULL, terminal, (void *)(uintptr_t)i);
+    for (int i = 0; i < threads; ++i) pthread_join(thread[i], NULL);
+    double elapsed = (double)(now_us() - start) / 1e6;
+    uint64_t total = 0;
+    for (int k = 0; k < KINDS; ++k) total += done[k];
+    printf("{\"system\":\"sqlite %s\",\"synchronous\":\"%s\",\"warehouses\":%d,\"terminals\":%d,\"transactions_per_second\":%.0f,"
+           "\"new_orders_per_minute\":%.0f,\"undone_new_orders\":%llu",
+           sqlite3_libversion(), synchronous, warehouses, threads, (double)total / elapsed,
+           (double)(done[NEW_ORDER] - undone) * 60 / elapsed, (unsigned long long)undone);
+    for (int k = 0; k < KINDS; ++k)
+        printf(",\"%s\":{\"done\":%llu,\"failed\":%llu,\"p50_us\":%llu,\"p99_us\":%llu}", names[k], (unsigned long long)done[k],
+               (unsigned long long)failed[k], (unsigned long long)percentile(histogram[k], done[k], 0.5),
+               (unsigned long long)percentile(histogram[k], done[k], 0.99));
+    puts("}");
+    return 0;
+}
diff --git a/benchmarks/tpcc/tpcc-load.c b/benchmarks/tpcc/tpcc-load.c
new file mode 100644
index 0000000..176ec1e
--- /dev/null
+++ b/benchmarks/tpcc/tpcc-load.c
@@ -0,0 +1,261 @@
+#define _GNU_SOURCE
+/* TPC-C's terminals for benchmarks/tpcc/tpcc.aup over HTTP/1.1: N
+ * connections on one epoll thread, each a terminal of a home warehouse
+ * (connection mod W, plus one) sending one transaction at a time, picked
+ * by the mix of clause 5.2.3 (New-Order 45%, Payment 43%, Order-Status,
+ * Delivery and Stock-Level 4% each) with the inputs of clause 2 (NURand
+ * customers and items, 1% of New-Orders naming an unused item, 1% of lines
+ * and 15% of Payments from another warehouse when there is one). Delivery
+ * is one district a request, as tpcc.aup takes it.
+ *
+ * tpcc-load PORT CONNECTIONS SECONDS WAREHOUSES [SEED [TRANSACTIONS]]
+ *
+ * With TRANSACTIONS each terminal stops after that many. The inputs come in
+ * the order sqlite-tpcc draws them, so one terminal and a seed make the same
+ * transactions there (run.py compares the states they leave).
+ *
+ * Prints one JSON line: transactions a second, New-Orders a minute, and for
+ * each kind its count, failures (a New-Order undone for its unused item is
+ * not one) and latency percentiles (µs). */
+#include <arpa/inet.h>
+#include <errno.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <stdbool.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/epoll.h>
+#include <sys/socket.h>
+#include <time.h>
+#include <unistd.h>
+
+enum { BUCKETS = 64 * 16, RESPONSE = 1 << 20, REQUEST = 4096, KINDS = 5 };
+enum { NEW_ORDER, PAYMENT, ORDER_STATUS, DELIVERY, STOCK_LEVEL };
+static const char *const names[KINDS] = {"new_order", "payment", "order_status", "delivery", "stock_level"};
+
+typedef struct {
+    int fd, home, kind;
+    bool unused; /* a New-Order naming an unused item: undone, as it should be */
+    char request[REQUEST];
+    size_t request_size, sent;
+    char *response;
+    size_t received;
+    uint64_t started, random, left;
+} Connection;
+
+static uint64_t now_us(void) {
+    struct timespec t;
+    clock_gettime(CLOCK_MONOTONIC, &t);
+    return (uint64_t)t.tv_sec * 1000000u + (uint64_t)t.tv_nsec / 1000u;
+}
+
+/* Log-linear histograms: 16 buckets a power of two, one a kind. */
+static uint64_t histogram[KINDS][BUCKETS], done[KINDS], failed[KINDS], undone;
+static unsigned bucket(uint64_t us) {
+    if (us < 16) return (unsigned)us;
+    unsigned power = 63 - (unsigned)__builtin_clzll(us);
+    unsigned index = (power - 3) * 16 + (unsigned)((us >> (power - 4)) & 15);
+    return index < BUCKETS ? index : BUCKETS - 1;
+}
+static uint64_t bucket_value(unsigned index) {
+    if (index < 16) return index;
+    unsigned power = index / 16 + 3;
+    return ((uint64_t)16 + index % 16) << (power - 4);
+}
+static uint64_t percentile(const uint64_t *counts, uint64_t total, double p) {
+    uint64_t want = (uint64_t)((double)total * p), seen = 0;
+    for (unsigned i = 0; i < BUCKETS; ++i) {
+        seen += counts[i];
+        if (seen > want) return bucket_value(i);
+    }
+    return 0;
+}
+
+static int port, warehouses;
+static int epoll_fd;
+
+/* xorshift64: each terminal its own sequence. */
+static uint64_t next_random(uint64_t *state) {
+    uint64_t x = *state;
+    x ^= x << 13, x ^= x >> 7, x ^= x << 17;
+    return *state = x;
+}
+static int uniform(uint64_t *state, int low, int high) { return low + (int)(next_random(state) % (uint64_t)(high - low + 1)); }
+/* NURand (2.1.6) with a run's constants. */
+static int nurand(uint64_t *state, int a, int low, int high, int c) {
+    return ((uniform(state, 0, a) | uniform(state, low, high)) + c) % (high - low + 1) + low;
+}
+
+static long district_key(int w, int d) { return w * 10L + d; }
+static long customer_key(int w, int d, int c) { return district_key(w, d) * 3000 + c; }
+
+/* The terminal's next transaction, as a request. */
+static void prepare(Connection *c) {
+    uint64_t *r = &c->random;
+    int w = c->home, d = uniform(r, 1, 10), roll = uniform(r, 1, 100);
+    char body[REQUEST - 256];
+    int length = 0;
+    const char *path = NULL;
+    c->kind = roll <= 45 ? NEW_ORDER : roll <= 88 ? PAYMENT : roll <= 92 ? ORDER_STATUS : roll <= 96 ? DELIVERY : STOCK_LEVEL;
+    c->unused = false;
+    if (c->kind == NEW_ORDER) {
+        int customer = nurand(r, 1023, 1, 3000, 259), count = uniform(r, 5, 15);
+        c->unused = uniform(r, 1, 100) == 1;
+        length = snprintf(body, sizeof(body), "{\"w\":%d,\"d\":%d,\"c\":%d,\"lines\":[", w, d, customer);
+        for (int i = 0; i < count; ++i) {
+            int item = c->unused && i == count - 1 ? 100001 : nurand(r, 8191, 1, 100000, 7911);
+            int supply = warehouses > 1 && uniform(r, 1, 100) == 1 ? uniform(r, 1, warehouses) : w;
+            length += snprintf(body + length, sizeof(body) - (size_t)length, "%s{\"item\":%d,\"supply\":%d,\"quantity\":%d}",
+                               i ? "," : "", item, supply, uniform(r, 1, 10));
+        }
+        length += snprintf(body + length, sizeof(body) - (size_t)length, "]}");
+        path = "/inputs/new_order";
+    } else if (c->kind == PAYMENT) {
+        int customer = nurand(r, 1023, 1, 3000, 259), cw = w, cd = d;
+        if (warehouses > 1 && uniform(r, 1, 100) > 85) {
+            do cw = uniform(r, 1, warehouses); while (cw == w);
+            cd = uniform(r, 1, 10);
+        }
+        double amount = uniform(r, 100, 500000) / 100.0;
+        length = snprintf(body, sizeof(body), "{\"w\":%d,\"d\":%d,\"cw\":%d,\"cd\":%d,\"c\":%d,\"amount\":%.2f}", w, d, cw,
+                          cd, customer, amount);
+        path = "/inputs/payment";
+    } else if (c->kind == DELIVERY) {
+        length = snprintf(body, sizeof(body), "{\"w\":%d,\"d\":%d,\"carrier\":%d}", w, d, uniform(r, 1, 10));
+        path = "/inputs/delivery";
+    }
+    if (path)
+        c->request_size = (size_t)snprintf(c->request, sizeof(c->request),
+            "POST %s HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: %d\r\n\r\n%s",
+            path, length, body);
+    else if (c->kind == ORDER_STATUS)
+        c->request_size = (size_t)snprintf(c->request, sizeof(c->request),
+            "GET /views/order_status?customer=%ld HTTP/1.1\r\nHost: localhost\r\n\r\n", customer_key(w, d, nurand(r, 1023, 1, 3000, 259)));
+    else
+        c->request_size = (size_t)snprintf(c->request, sizeof(c->request),
+            "GET /views/stock_level?w=%d&d=%d&threshold=%d HTTP/1.1\r\nHost: localhost\r\n\r\n", w, d, uniform(r, 10, 20));
+    c->sent = c->received = 0;
+    c->started = now_us();
+}
+
+static void open_connection(Connection *c) {
+    c->fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
+    int one = 1;
+    setsockopt(c->fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
+    struct sockaddr_in address = {.sin_family = AF_INET, .sin_port = htons((uint16_t)port),
+                                  .sin_addr.s_addr = htonl(INADDR_LOOPBACK)};
+    connect(c->fd, (struct sockaddr *)&address, sizeof(address));
+    struct epoll_event event = {.events = EPOLLOUT | EPOLLIN, .data.ptr = c};
+    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, c->fd, &event);
+}
+
+static void reopen(Connection *c) {
+    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, c->fd, NULL);
+    close(c->fd);
+    open_connection(c);
+}
+
+/* A complete response in c->response: its status and length, or false. */
+static bool complete(Connection *c, int *status, size_t *length) {
+    char *end = memmem(c->response, c->received, "\r\n\r\n", 4);
+    if (!end) return false;
+    size_t header = (size_t)(end - c->response) + 4, content = 0;
+    char *field = memmem(c->response, header, "Content-Length:", 15);
+    if (!field) field = memmem(c->response, header, "content-length:", 15);
+    if (field) content = strtoul(field + 15, NULL, 10);
+    if (c->received < header + content) return false;
+    *status = atoi(c->response + 9);
+    *length = header + content;
+    return true;
+}
+
+int main(int argc, char **argv) {
+    if (argc < 5) {
+        fputs("tpcc-load PORT CONNECTIONS SECONDS WAREHOUSES [SEED [TRANSACTIONS]]\n", stderr);
+        return 2;
+    }
+    port = atoi(argv[1]);
+    int count = atoi(argv[2]);
+    double seconds = atof(argv[3]);
+    warehouses = atoi(argv[4]);
+    uint64_t seed = argc > 5 ? strtoull(argv[5], NULL, 10) : 1, each = argc > 6 ? strtoull(argv[6], NULL, 10) : UINT64_MAX;
+    epoll_fd = epoll_create1(0);
+    Connection *all = calloc((size_t)count, sizeof(*all));
+    for (int i = 0; i < count; ++i) {
+        all[i].home = i % warehouses + 1;
+        all[i].random = 0x9e3779b97f4a7c15u * (uint64_t)(i + 1) ^ seed * 0xbf58476d1ce4e5b9u;
+        all[i].left = each;
+        all[i].response = malloc(RESPONSE);
+        open_connection(&all[i]);
+        prepare(&all[i]);
+    }
+    uint64_t start = now_us(), stop = start + (uint64_t)(seconds * 1e6), errors = 0;
+    int finished = 0;
+    struct epoll_event events[256];
+    while (now_us() < stop && finished < count) {
+        int ready = epoll_wait(epoll_fd, events, 256, 100);
+        for (int e = 0; e < ready; ++e) {
+            Connection *c = events[e].data.ptr;
+            if (events[e].events & (EPOLLERR | EPOLLHUP) && !(events[e].events & EPOLLIN)) {
+                ++errors;
+                reopen(c);
+                prepare(c);
+                continue;
+            }
+            if (events[e].events & EPOLLOUT) {
+                while (c->sent < c->request_size) {
+                    ssize_t n = send(c->fd, c->request + c->sent, c->request_size - c->sent, MSG_NOSIGNAL);
+                    if (n <= 0) break;
+                    c->sent += (size_t)n;
+                }
+                if (c->sent == c->request_size) {
+                    struct epoll_event event = {.events = EPOLLIN, .data.ptr = c};
+                    epoll_ctl(epoll_fd, EPOLL_CTL_MOD, c->fd, &event);
+                }
+            }
+            if (events[e].events & EPOLLIN) {
+                ssize_t n = recv(c->fd, c->response + c->received, RESPONSE - c->received, 0);
+                if (n <= 0) {
+                    if (n < 0 && errno == EAGAIN) continue;
+                    ++errors;
+                    reopen(c);
+                    prepare(c);
+                    continue;
+                }
+                c->received += (size_t)n;
+                int status;
+                size_t length;
+                if (!complete(c, &status, &length)) continue;
+                bool expected = status == 200 || (c->unused && status == 404);
+                ++done[c->kind];
+                undone += c->unused && status == 404;
+                failed[c->kind] += !expected;
+                ++histogram[c->kind][bucket(now_us() - c->started)];
+                if (!--c->left) { /* this terminal's transactions are done */
+                    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, c->fd, NULL);
+                    close(c->fd);
+                    ++finished;
+                    continue;
+                }
+                struct epoll_event event = {.events = EPOLLOUT | EPOLLIN, .data.ptr = c};
+                epoll_ctl(epoll_fd, EPOLL_CTL_MOD, c->fd, &event);
+                prepare(c);
+            }
+        }
+    }
+    double elapsed = (double)(now_us() - start) / 1e6;
+    uint64_t total = 0;
+    for (int k = 0; k < KINDS; ++k) total += done[k];
+    printf("{\"system\":\"pure\",\"warehouses\":%d,\"terminals\":%d,\"transactions_per_second\":%.0f,"
+           "\"new_orders_per_minute\":%.0f,\"undone_new_orders\":%llu,\"errors\":%llu",
+           warehouses, count, (double)total / elapsed, (double)(done[NEW_ORDER] - undone) * 60 / elapsed,
+           (unsigned long long)undone, (unsigned long long)errors);
+    for (int k = 0; k < KINDS; ++k)
+        printf(",\"%s\":{\"done\":%llu,\"failed\":%llu,\"p50_us\":%llu,\"p99_us\":%llu}", names[k], (unsigned long long)done[k],
+               (unsigned long long)failed[k], (unsigned long long)percentile(histogram[k], done[k], 0.5),
+               (unsigned long long)percentile(histogram[k], done[k], 0.99));
+    puts("}");
+    return 0;
+}
diff --git a/benchmarks/tpcc/tpcc.aup b/benchmarks/tpcc/tpcc.aup
new file mode 100644
index 0000000..b3e75aa
--- /dev/null
+++ b/benchmarks/tpcc/tpcc.aup
@@ -0,0 +1,266 @@
+# TPC-C (revision 5.11): its tables and its five transactions, the clause 2
+# profiles, each composite key packed into one number: a district is
+# w * 10 + d, a customer district * 3000 + c, an order district * 10^7 + o,
+# an order line order * 16 + number, a stock item w * 100000 + i.
+#
+# Where this departs from the specification: a customer is always chosen by
+# number (never by last name), texts the transactions only carry along
+# (names, addresses, data) are left out, a customer keeps its last order,
+# and Delivery is an event a district (clause 2.7.4 lets the ten districts
+# be separate transactions).
+
+record Warehouse {
+  id = 0
+  tax = 0
+  ytd = 0
+}
+record District {
+  id = 0
+  tax = 0
+  ytd = 0
+  next_order = 3001
+  next_delivery = 2101
+}
+record Customer {
+  id = 0
+  discount = 0
+  credit = 1
+  balance = -10
+  ytd_payment = 10
+  payments = 1
+  deliveries = 0
+  last_order = 0
+}
+record Item {
+  id = 0
+  price = 0
+}
+record Stock {
+  id = 0
+  quantity = 0
+  ytd = 0
+  orders = 0
+  remote = 0
+}
+record Order {
+  id = 0
+  customer = 0
+  entry = 0
+  carrier = 0
+  lines = 0
+  local = 1
+}
+record NewOrder {
+  id = 0
+}
+record Line {
+  id = 0
+  item = 0
+  supply = 0
+  delivery = 0
+  quantity = 0
+  amount = 0
+}
+record History {
+  customer = 0
+  district = 0
+  date = 0
+  amount = 0
+}
+record Ordered {
+  item = 0
+  supply = 0
+  quantity = 0
+}
+
+state warehouses: list of Warehouse by id
+state districts: list of District by id
+state customers: list of Customer by id
+state items: list of Item by id
+state stock: list of Stock by id
+state orders: list of Order by id
+state new_orders: list of NewOrder by id
+state lines: list of Line by id
+state history: list of History
+
+# Loading: a table's rows in bulk.
+input load_warehouses = message {
+  rows: list of Warehouse
+}
+input load_districts = message {
+  rows: list of District
+}
+input load_customers = message {
+  rows: list of Customer
+}
+input load_items = message {
+  rows: list of Item
+}
+input load_stock = message {
+  rows: list of Stock
+}
+input load_orders = message {
+  rows: list of Order
+}
+input load_new_orders = message {
+  rows: list of NewOrder
+}
+input load_lines = message {
+  rows: list of Line
+}
+on load_warehouses {
+  for r in load_warehouses.rows {
+    warehouses.add(Warehouse { id: r.id, tax: r.tax, ytd: r.ytd })
+  }
+}
+on load_districts {
+  for r in load_districts.rows {
+    districts.add(District { id: r.id, tax: r.tax, ytd: r.ytd, next_order: r.next_order, next_delivery: r.next_delivery })
+  }
+}
+on load_customers {
+  for r in load_customers.rows {
+    customers.add(Customer { id: r.id, discount: r.discount, credit: r.credit, balance: r.balance,
+                             ytd_payment: r.ytd_payment, payments: r.payments, deliveries: r.deliveries,
+                             last_order: r.last_order })
+  }
+}
+on load_items {
+  for r in load_items.rows {
+    items.add(Item { id: r.id, price: r.price })
+  }
+}
+on load_stock {
+  for r in load_stock.rows {
+    stock.add(Stock { id: r.id, quantity: r.quantity, ytd: r.ytd, orders: r.orders, remote: r.remote })
+  }
+}
+on load_orders {
+  for r in load_orders.rows {
+    orders.add(Order { id: r.id, customer: r.customer, entry: r.entry, carrier: r.carrier, lines: r.lines, local: r.local })
+  }
+}
+on load_new_orders {
+  for r in load_new_orders.rows {
+    new_orders.add(NewOrder { id: r.id })
+  }
+}
+on load_lines {
+  for r in load_lines.rows {
+    lines.add(Line { id: r.id, item: r.item, supply: r.supply, delivery: r.delivery, quantity: r.quantity,
+                     amount: r.amount })
+  }
+}
+
+# New-Order (2.4): an order of 5 to 15 lines, each taking stock; an unused
+# item (1% of orders) fails the whole order, which is then undone.
+input new_order = message {
+  w: u32
+  d: u32
+  c: u32
+  lines: list of Ordered
+}
+on new_order {
+  let district = districts[new_order.w * 10 + new_order.d]
+  let customer = customers[district.id * 3000 + new_order.c]
+  let order = district.id * 10000000 + district.next_order
+  district.next_order = district.next_order + 1
+  let local = 1
+  let count = 0
+  for l in new_order.lines {
+    let price = items[l.item].price
+    let s = stock[l.supply * 100000 + l.item]
+    if s.quantity >= l.quantity + 10 {
+      s.quantity = s.quantity - l.quantity
+    } else {
+      s.quantity = s.quantity - l.quantity + 91
+    }
+    s.ytd = s.ytd + l.quantity
+    s.orders = s.orders + 1
+    if l.supply != new_order.w {
+      s.remote = s.remote + 1
+      local = 0
+    }
+    count = count + 1
+    lines.add(Line { id: order * 16 + count, item: l.item, supply: l.supply, quantity: l.quantity,
+                     amount: l.quantity * price })
+  }
+  orders.add(Order { id: order, customer: customer.id, entry: now, lines: count, local: local })
+  new_orders.add(NewOrder { id: order })
+  customer.last_order = order
+}
+
+# Payment (2.5): a payment to a district, from one of its customers or
+# another warehouse's.
+input payment = message {
+  w: u32
+  d: u32
+  cw: u32
+  cd: u32
+  c: u32
+  amount: number
+}
+on payment {
+  let warehouse = warehouses[payment.w]
+  warehouse.ytd = warehouse.ytd + payment.amount
+  let district = districts[payment.w * 10 + payment.d]
+  district.ytd = district.ytd + payment.amount
+  let customer = customers[(payment.cw * 10 + payment.cd) * 3000 + payment.c]
+  customer.balance = customer.balance - payment.amount
+  customer.ytd_payment = customer.ytd_payment + payment.amount
+  customer.payments = customer.payments + 1
+  history.append(History { customer: customer.id, district: district.id, date: now, amount: payment.amount })
+}
+
+# Delivery (2.7), a district an event: its oldest undelivered order is
+# delivered, its lines dated, and their amount charged to its customer.
+input delivery = message {
+  w: u32
+  d: u32
+  carrier: u32
+}
+on delivery {
+  let district = districts[delivery.w * 10 + delivery.d]
+  if district.next_delivery < district.next_order {
+    let order = district.id * 10000000 + district.next_delivery
+    district.next_delivery = district.next_delivery + 1
+    new_orders.remove(new_orders[order])
+    let o = orders[order]
+    o.carrier = delivery.carrier
+    let total = 0
+    for l in lines[order * 16 .. order * 16 + 16] {
+      l.delivery = now
+      total = total + l.amount
+    }
+    let customer = customers[o.customer]
+    customer.balance = customer.balance + total
+    customer.deliveries = customer.deliveries + 1
+  }
+}
+
+# Order-Status (2.6): a customer's balance, and the lines of its last order.
+view order_status(customer: u32) = each l in lines where l.id >= customers[customer].last_order * 16 && l.id < customers[customer].last_order * 16 + 16 {
+  balance = customers[customer].balance
+  item = l.item
+  supply = l.supply
+  quantity = l.quantity
+  amount = l.amount
+  delivery = l.delivery
+}
+
+# Stock-Level (2.8): how many of the items of a district's last 20 orders
+# are below a threshold.
+view stock_level(w: u32, d: u32, threshold: number) = distinct(l.item for l in lines[
+    (districts[w * 10 + d].id * 10000000 + districts[w * 10 + d].next_order - 20) * 16 ..
+    (districts[w * 10 + d].id * 10000000 + districts[w * 10 + d].next_order) * 16]
+  where stock[w * 100000 + l.item].quantity < threshold)
+
+# Consistency (3.3.2): each view has its twin in SQL.
+view warehouse_ytd = sum(w.ytd for w in warehouses)
+view district_ytd = sum(d.ytd for d in districts)
+view order_count = orders.count
+view new_order_count = new_orders.count
+view line_count = lines.count
+view ordered_lines = sum(o.lines for o in orders)
+view delivered = count(o in orders where o.carrier > 0)
+view history_count = history.count