From 44740eb93a69a052ee7b4c46f2ccac384c9d4f36 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sun, 27 Sep 2026 23:39:29 +0100 Subject: Run pgbench's transaction on a bank of 100,000 accounts, beside SQLite (090) benchmarks/bank/bank.aup keeps accounts, tellers and branches as keyed lists and a history list; a transfer changes a row of each and appends a history row in one event, as pgbench's TPC-B-like transaction does in SQL. run.py makes a scale's rows and drives transfers through the load generator's new transfer kind; sqlite-bank.c runs the same tables and SQL on SQLite 3.46.1, a connection a thread, WAL. With lists as trees, transfers run at 30.5k a second on 1,000 accounts and 24.5k on 100,000 (RAM store, one connection thread, HTTP and JSON included), where they ran at 3.2k on 30,000 before. Durable on the SD card, Pure makes 1,339 a second and SQLite with synchronous=FULL 84 to 153. SQLite in process with no HTTP makes about 40k in RAM: recorded as it is, with the write amplification of 100,000-row tables on the card and ad hoc queries as what comes next. --- benchmarks/bank/bank.aup | 64 +++++++++++++++ benchmarks/bank/run.py | 117 ++++++++++++++++++++++++++++ benchmarks/bank/sqlite-bank.c | 176 ++++++++++++++++++++++++++++++++++++++++++ benchmarks/http/load.c | 44 +++++++++-- 4 files changed, 393 insertions(+), 8 deletions(-) create mode 100644 benchmarks/bank/bank.aup create mode 100644 benchmarks/bank/run.py create mode 100644 benchmarks/bank/sqlite-bank.c (limited to 'benchmarks') diff --git a/benchmarks/bank/bank.aup b/benchmarks/bank/bank.aup new file mode 100644 index 0000000..acc7938 --- /dev/null +++ b/benchmarks/bank/bank.aup @@ -0,0 +1,64 @@ +# A bank as pgbench's TPC-B-like transaction: accounts, tellers and branches, +# each a list kept in key order, and a history of every transfer. A transfer +# moves a teller's delta into an account, the teller and its branch, and +# records it, all in one event: one transaction over four tables. +record Account { + aid = 0 + bid = 0 + balance = 0 + filler: text +} +record Teller { + tid = 0 + bid = 0 + balance = 0 +} +record Branch { + bid = 0 + balance = 0 +} +record Transfer { + tid = 0 + bid = 0 + aid = 0 + delta = 0 + at = 0 +} + +state accounts: list of Account by aid +state tellers: list of Teller by tid +state branches: list of Branch by bid +state history: list of Transfer + +input branch = message { + bid: u32 +} +input teller = message { + tid: u32 + bid: u32 +} +input account = message { + aid: u32 + bid: u32 +} +input transfer = message { + aid: u32 + tid: u32 + bid: u32 + delta: number +} + +on branch { branches.add(Branch { bid: branch.bid }) } +on teller { tellers.add(Teller { tid: teller.tid, bid: teller.bid }) } +on account { accounts.add(Account { aid: account.aid, bid: account.bid, filler: "" }) } + +on transfer { + accounts[transfer.aid].balance = accounts[transfer.aid].balance + transfer.delta + tellers[transfer.tid].balance = tellers[transfer.tid].balance + transfer.delta + branches[transfer.bid].balance = branches[transfer.bid].balance + transfer.delta + history.append(Transfer { tid: transfer.tid, bid: transfer.bid, aid: transfer.aid, delta: transfer.delta, at: now }) +} + +view accounts_count = accounts.count +view history_count = history.count +view total = sum(b.balance for b in branches) diff --git a/benchmarks/bank/run.py b/benchmarks/bank/run.py new file mode 100644 index 0000000..218c8ec --- /dev/null +++ b/benchmarks/bank/run.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +"""A bank served by pure-serve under pgbench's TPC-B-like transaction. + +bank.aup keeps accounts, tellers and branches as lists in key order and a +history list; a transfer updates one row of each of the three and appends a +history row, in one event. For each --accounts count: a fresh store, the +branches, tellers and accounts of that scale made by posts (timed), then +--seconds of transfers from --connections clients (benchmarks/http/load +transfer), servers on cores 0-2 and the load on core 3. Prints one JSON line +per count: setup time, transfers a second, latency, and server CPU time per +transfer. +""" +import argparse +import http.client +import json +import os +import select +import shutil +import subprocess +import tempfile +import threading +import time +from pathlib import Path + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parents[1] + + +def cpu_seconds(pid): + fields = Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split() + return (int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK') + + +def post_all(port, path, bodies, threads=8): + """Each body posted once, from threads keep-alive connections.""" + failed = [] + + def worker(part): + connection = http.client.HTTPConnection('127.0.0.1', port, timeout=60) + for body in part: + connection.request('POST', path, body, {'Content-Type': 'application/json'}) + answer = connection.getresponse() + answer.read() + if answer.status != 200: + failed.append((body, answer.status)) + connection.close() + + workers = [threading.Thread(target=worker, args=(bodies[i::threads],)) for i in range(threads)] + for w in workers: + w.start() + for w in workers: + w.join() + assert not failed, failed[:3] + + +def run(serve, module, directory, accounts, seconds, connections, threads): + store = directory / f'bank-{accounts}' + shutil.rmtree(store, ignore_errors=True) + process = subprocess.Popen(['taskset', '-c', '0-2', str(serve), '--module', str(module), '--store', str(store), + '--port', '0', '--threads', str(threads)], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + try: + select.select([process.stdout], [], [], 10) + port = json.loads(process.stdout.readline())['port'] + scale = max(1, accounts // 100000) + began = time.perf_counter() + post_all(port, '/inputs/branch', [json.dumps({'bid': b + 1}).encode() for b in range(scale)], 1) + post_all(port, '/inputs/teller', [json.dumps({'tid': t + 1, 'bid': t // 10 + 1}).encode() + for t in range(10 * scale)], 1) + post_all(port, '/inputs/account', [json.dumps({'aid': a + 1, 'bid': a // 100000 + 1}).encode() + for a in range(accounts)]) + setup = time.perf_counter() - began + used = cpu_seconds(process.pid) + load = subprocess.run(['taskset', '-c', '3', str(ROOT / 'benchmarks' / 'http' / 'load'), str(port), + str(connections), str(seconds), 'transfer', str(accounts), 'keep-alive'], + capture_output=True, text=True, check=True) + result = json.loads(load.stdout) + used = cpu_seconds(process.pid) - used + connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30) + connection.request('GET', '/views/history_count') + history = int(connection.getresponse().read()) + connection.close() + # Every transfer answered is in the history, and those in flight when the load stopped may be too. + assert result['2xx'] <= history <= result['2xx'] + connections, (history, result) + return {'accounts': accounts, 'setup_s': round(setup, 2), 'transfers_per_second': result['requests_per_second'], + 'p50_us': result['p50_us'], 'p99_us': result['p99_us'], 'errors': result['errors'], + 'non_2xx': result['completed'] - result['2xx'], + 'cpu_us_per_transfer': round(used / max(result['2xx'], 1) * 1e6, 1)} + finally: + process.terminate() + process.communicate(timeout=60) + shutil.rmtree(store, ignore_errors=True) + + +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')) + parser.add_argument('--accounts', default='1000,10000,30000') + parser.add_argument('--seconds', type=float, default=5) + parser.add_argument('--connections', type=int, default=16) + parser.add_argument('--threads', type=int, default=1) + args = parser.parse_args() + work = Path(tempfile.mkdtemp(prefix='pure-bank-')) + try: + module = work / 'bank.pmod' + subprocess.run([str(args.aup), 'build', str(HERE / 'bank.aup'), '-o', str(module)], check=True) + for accounts in (int(a) for a in args.accounts.split(',')): + print(json.dumps({'directory': str(args.directory), **run(args.serve, module, args.directory, accounts, + args.seconds, args.connections, args.threads)}), + flush=True) + finally: + shutil.rmtree(work) + + +if __name__ == '__main__': + main() diff --git a/benchmarks/bank/sqlite-bank.c b/benchmarks/bank/sqlite-bank.c new file mode 100644 index 0000000..e4ab48f --- /dev/null +++ b/benchmarks/bank/sqlite-bank.c @@ -0,0 +1,176 @@ +#define _GNU_SOURCE +/* pgbench's TPC-B-like transaction on SQLite, the same bank as bank.aup for + * comparing with Pure (benchmarks/bank/run.py): + * + * sqlite-bank DATABASE ACCOUNTS THREADS SECONDS full|normal|off + * + * makes pgbench's tables (a branch, 10 tellers and 100,000 accounts a scale; + * ACCOUNTS, at least one scale), then THREADS threads, each with its own + * connection, run the transaction for SECONDS: BEGIN IMMEDIATE; UPDATE + * accounts; SELECT the balance; UPDATE tellers; UPDATE branches; INSERT into + * history; COMMIT, with random ids and delta as pgbench picks them. WAL, with + * synchronous FULL (each commit synced), NORMAL or OFF. Prints one JSON line. */ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +static const char *database, *synchronous; +static unsigned accounts, scale; +static double seconds; +static atomic_uint_fast64_t done, failed; +enum { BUCKETS = 64 * 16 }; +static atomic_uint_fast64_t histogram[BUCKETS]; + +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; +} + +/* Log-linear buckets: 16 a power of two. */ +static unsigned bucket(uint64_t us) { + if (us < 16) return (unsigned)us; + unsigned power = 63 - (unsigned)__builtin_clzll(us); + return (power - 3) * 16 + (unsigned)((us >> (power - 4)) & 15); +} +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(uint64_t total, double p) { + uint64_t want = (uint64_t)((double)total * p), seen = 0; + for (unsigned i = 0; i < BUCKETS; ++i) { + seen += histogram[i]; + if (seen > want) return bucket_value(i); + } + return 0; +} + +static sqlite3 *open_database(void) { + sqlite3 *db; + char pragmas[160]; + if (sqlite3_open(database, &db) != SQLITE_OK) { fprintf(stderr, "%s\n", sqlite3_errmsg(db)); exit(1); } + sqlite3_busy_timeout(db, 60000); + snprintf(pragmas, sizeof(pragmas), "PRAGMA journal_mode=WAL; PRAGMA synchronous=%s;", synchronous); + if (sqlite3_exec(db, pragmas, NULL, NULL, NULL) != SQLITE_OK) { fprintf(stderr, "%s\n", sqlite3_errmsg(db)); exit(1); } + return db; +} + +static void run(sqlite3 *db, const char *sql) { + char *error = NULL; + if (sqlite3_exec(db, sql, NULL, NULL, &error) != SQLITE_OK) { fprintf(stderr, "%s: %s\n", sql, error); exit(1); } +} + +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\n", sqlite3_errmsg(db)); exit(1); } + return statement; +} + +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 bool step_done(sqlite3_stmt *statement) { + int result = sqlite3_step(statement); + sqlite3_reset(statement); + return result == SQLITE_DONE || result == SQLITE_ROW; +} + +static void *client(void *argument) { + uint64_t random = 0x9e3779b97f4a7c15u * (uintptr_t)argument; + sqlite3 *db = open_database(); + sqlite3_stmt *begin = prepare(db, "BEGIN IMMEDIATE"), *commit = prepare(db, "COMMIT"), + *rollback = prepare(db, "ROLLBACK"), + *account = prepare(db, "UPDATE pgbench_accounts SET abalance = abalance + ?1 WHERE aid = ?2"), + *balance = prepare(db, "SELECT abalance FROM pgbench_accounts WHERE aid = ?1"), + *teller = prepare(db, "UPDATE pgbench_tellers SET tbalance = tbalance + ?1 WHERE tid = ?2"), + *branch = prepare(db, "UPDATE pgbench_branches SET bbalance = bbalance + ?1 WHERE bid = ?2"), + *history = prepare(db, "INSERT INTO pgbench_history (tid, bid, aid, delta, mtime) " + "VALUES (?1, ?2, ?3, ?4, CURRENT_TIMESTAMP)"); + uint64_t stop = now_us() + (uint64_t)(seconds * 1e6); + while (now_us() < stop) { + int aid = (int)(next_random(&random) % accounts) + 1, tid = (int)(next_random(&random) % (10 * scale)) + 1, + bid = (int)(next_random(&random) % scale) + 1, delta = (int)(next_random(&random) % 10001) - 5000; + uint64_t started = now_us(); + sqlite3_bind_int(account, 1, delta), sqlite3_bind_int(account, 2, aid); + sqlite3_bind_int(balance, 1, aid); + sqlite3_bind_int(teller, 1, delta), sqlite3_bind_int(teller, 2, tid); + sqlite3_bind_int(branch, 1, delta), sqlite3_bind_int(branch, 2, bid); + sqlite3_bind_int(history, 1, tid), sqlite3_bind_int(history, 2, bid), sqlite3_bind_int(history, 3, aid), + sqlite3_bind_int(history, 4, delta); + bool ok = step_done(begin) && step_done(account) && step_done(balance) && step_done(teller) && + step_done(branch) && step_done(history) && step_done(commit); + if (!ok) { + step_done(rollback); + ++failed; + continue; + } + ++done; + ++histogram[bucket(now_us() - started)]; + } + sqlite3_stmt *statements[] = {begin, commit, rollback, account, balance, teller, branch, history}; + for (size_t i = 0; i < sizeof(statements) / sizeof(*statements); ++i) sqlite3_finalize(statements[i]); + sqlite3_close(db); + return NULL; +} + +int main(int argc, char **argv) { + if (argc != 6) { + fputs("sqlite-bank DATABASE ACCOUNTS THREADS SECONDS full|normal|off\n", stderr); + return 2; + } + database = argv[1]; + accounts = (unsigned)strtoul(argv[2], NULL, 10); + unsigned threads = (unsigned)strtoul(argv[3], NULL, 10); + seconds = atof(argv[4]); + synchronous = argv[5]; + scale = accounts / 100000 ? accounts / 100000 : 1; + uint64_t began = now_us(); + sqlite3 *db = open_database(); + run(db, "DROP TABLE IF EXISTS pgbench_accounts; DROP TABLE IF EXISTS pgbench_tellers;" + "DROP TABLE IF EXISTS pgbench_branches; DROP TABLE IF EXISTS pgbench_history;" + "CREATE TABLE pgbench_accounts (aid INTEGER PRIMARY KEY, bid INTEGER, abalance INTEGER, filler CHAR(84));" + "CREATE TABLE pgbench_tellers (tid INTEGER PRIMARY KEY, bid INTEGER, tbalance INTEGER, filler CHAR(84));" + "CREATE TABLE pgbench_branches (bid INTEGER PRIMARY KEY, bbalance INTEGER, filler CHAR(88));" + "CREATE TABLE pgbench_history (tid INTEGER, bid INTEGER, aid INTEGER, delta INTEGER, mtime TIMESTAMP," + " filler CHAR(22));"); + run(db, "BEGIN"); + sqlite3_stmt *insert = prepare(db, "INSERT INTO pgbench_branches (bid, bbalance, filler) VALUES (?1, 0, '')"); + for (unsigned b = 1; b <= scale; ++b) sqlite3_bind_int(insert, 1, (int)b), step_done(insert); + sqlite3_finalize(insert); + insert = prepare(db, "INSERT INTO pgbench_tellers (tid, bid, tbalance, filler) VALUES (?1, ?2, 0, '')"); + for (unsigned t = 1; t <= 10 * scale; ++t) sqlite3_bind_int(insert, 1, (int)t), sqlite3_bind_int(insert, 2, (int)((t - 1) / 10 + 1)), step_done(insert); + sqlite3_finalize(insert); + insert = prepare(db, "INSERT INTO pgbench_accounts (aid, bid, abalance, filler) VALUES (?1, ?2, 0, '')"); + for (unsigned a = 1; a <= accounts; ++a) + sqlite3_bind_int(insert, 1, (int)a), sqlite3_bind_int(insert, 2, (int)((a - 1) / 100000 + 1)), step_done(insert); + sqlite3_finalize(insert); + run(db, "COMMIT; PRAGMA wal_checkpoint(TRUNCATE);"); + double setup = (double)(now_us() - began) / 1e6; + pthread_t *thread = calloc(threads, sizeof(*thread)); + uint64_t start = now_us(); + for (unsigned i = 0; i < threads; ++i) pthread_create(&thread[i], NULL, client, (void *)(uintptr_t)(i + 1)); + for (unsigned i = 0; i < threads; ++i) pthread_join(thread[i], NULL); + double elapsed = (double)(now_us() - start) / 1e6; + sqlite3_stmt *count = prepare(db, "SELECT count(*) FROM pgbench_history"); + sqlite3_step(count); + long long history = sqlite3_column_int64(count, 0); + sqlite3_finalize(count); + sqlite3_close(db); + printf("{\"sqlite\":\"%s\",\"synchronous\":\"%s\",\"accounts\":%u,\"threads\":%u,\"setup_s\":%.2f," + "\"transfers_per_second\":%.0f,\"p50_us\":%llu,\"p99_us\":%llu,\"failed\":%llu,\"history\":%lld}\n", + sqlite3_libversion(), synchronous, accounts, threads, setup, (double)done / elapsed, + (unsigned long long)percentile(done, 0.5), (unsigned long long)percentile(done, 0.99), + (unsigned long long)failed, history); + return history == (long long)done ? 0 : 1; +} diff --git a/benchmarks/http/load.c b/benchmarks/http/load.c index d07b2bb..621e99c 100644 --- a/benchmarks/http/load.c +++ b/benchmarks/http/load.c @@ -2,7 +2,7 @@ /* HTTP/1.1 load for Pure's benchmarks: N connections on one epoll thread, * each sending one request at a time and measuring its latency. * - * load PORT CONNECTIONS SECONDS get|put|score|total|page|note|json KEYS [keep-alive|close] [BODY] + * load PORT CONNECTIONS SECONDS get|put|score|total|page|note|json|transfer KEYS [keep-alive|close] [BODY] * get: GET /objects/kI for I in [0, KEYS). * put: connection C edits its own key kC (C mod KEYS), inserting BODY bytes * at 0 with If-Match on the revision it last saw (If-None-Match: * at @@ -11,7 +11,11 @@ * the name as the body), total reads GET /views/total and page GET * /views/leaders?from=0&count=20. note reads GET /views/page (a text state * shown as it is, as in a notes app). json posts {"name":"k","text":"x..."} - * (BODY bytes of text) to POST /inputs/echo as application/json. + * (BODY bytes of text) to POST /inputs/echo as application/json. transfer is + * pgbench's TPC-B-like transaction for benchmarks/bank: POST /inputs/transfer + * {"aid","tid","bid","delta"}, a random account of KEYS, teller and branch of + * the scale (100,000 accounts, 10 tellers and a branch each) and delta in + * [-5000, 5000], as pgbench picks them. * Prints one JSON line: requests/s, statuses, latency percentiles (µs). */ #include #include @@ -37,7 +41,7 @@ typedef struct { size_t request_size, sent, body_size, body_sent; char *response; size_t received; - uint64_t started, revision; + uint64_t started, revision, random; } Connection; static uint64_t now_us(void) { @@ -69,14 +73,36 @@ static uint64_t percentile(uint64_t total, double p) { } static int port, connections_count, keys; -static enum { GET, PUT, SCORE, TOTAL, PAGE, NOTE, JSON } method; +static enum { GET, PUT, SCORE, TOTAL, PAGE, NOTE, JSON, TRANSFER } method; static bool put, keep_alive; static char *body; static size_t body_bytes; static int epoll_fd; +/* xorshift64: each connection 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 void prepare(Connection *c) { const char *closing = keep_alive ? "" : "Connection: close\r\n"; + if (method == TRANSFER) { + unsigned scale = keys / 100000 ? (unsigned)(keys / 100000) : 1; + char json[128]; + int length = snprintf(json, sizeof(json), "{\"aid\":%u,\"tid\":%u,\"bid\":%u,\"delta\":%d}", + (unsigned)(next_random(&c->random) % (uint64_t)keys) + 1, + (unsigned)(next_random(&c->random) % (10u * scale)) + 1, + (unsigned)(next_random(&c->random) % scale) + 1, + (int)(next_random(&c->random) % 10001) - 5000); + c->request_size = (size_t)snprintf(c->request, sizeof(c->request), + "POST /inputs/transfer HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\n" + "Content-Length: %d\r\n%s\r\n%s", length, closing, json); + c->sent = c->received = c->body_sent = c->body_size = 0; + c->started = now_us(); + return; + } if (method == JSON) { c->request_size = (size_t)snprintf(c->request, sizeof(c->request), "POST /inputs/echo HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\n" @@ -151,15 +177,16 @@ static bool complete(Connection *c, int *status, bool *closing, size_t *length) int main(int argc, char **argv) { if (argc < 6) { - fputs("load PORT CONNECTIONS SECONDS get|put|score|total|page|note|json KEYS [keep-alive|close] [BODY]\n", stderr); + fputs("load PORT CONNECTIONS SECONDS get|put|score|total|page|note|json|transfer KEYS [keep-alive|close] [BODY]\n", + stderr); return 2; } port = atoi(argv[1]); connections_count = atoi(argv[2]); double seconds = atof(argv[3]); - static const char *const methods[] = {"get", "put", "score", "total", "page", "note", "json"}; - for (method = GET; method <= JSON && strcmp(argv[4], methods[method]); ++method) {} - if (method > JSON) { fprintf(stderr, "load: unknown method %s\n", argv[4]); return 2; } + static const char *const methods[] = {"get", "put", "score", "total", "page", "note", "json", "transfer"}; + for (method = GET; method <= TRANSFER && strcmp(argv[4], methods[method]); ++method) {} + if (method > TRANSFER) { fprintf(stderr, "load: unknown method %s\n", argv[4]); return 2; } put = method == PUT; keys = atoi(argv[5]); keep_alive = argc > 6 && !strcmp(argv[6], "keep-alive"); @@ -177,6 +204,7 @@ int main(int argc, char **argv) { Connection *all = calloc((size_t)connections_count, sizeof(*all)); for (int i = 0; i < connections_count; ++i) { all[i].key = i % keys; + all[i].random = 0x9e3779b97f4a7c15u * (uint64_t)(i + 1); all[i].response = malloc(RESPONSE); open_connection(&all[i]); prepare(&all[i]); -- cgit v1.2.3