From 123966fa4a7f65df311960a5d8a42a92ae432408 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Wed, 30 Sep 2026 07:06:52 +0100 Subject: Let TPC-C's SQLite commit in groups as pure-serve does, and say what each run costs The terminals hand their writing transactions to one writer, which runs those waiting in one transaction, each in a savepoint, and commits them with one sync; reads stay on the terminals' connections, as pure-serve shows views from the last save. An undone New-Order is rolled back by its caller now (it rolled back the whole group before, and counted as failed after the split). Each run reports its CPU a transaction, the owner thread's share of a core, its memory, and the bytes a transaction had written, by the process and to the device. SQLite runs with its database mapped, a 64 MiB cache and savepoint journals in memory unless told otherwise. --- benchmarks/tpcc/run.py | 153 +++++++++++++++++++++++++++++++------ benchmarks/tpcc/sqlite-tpcc.c | 173 +++++++++++++++++++++++++++++++++++------- 2 files changed, 276 insertions(+), 50 deletions(-) diff --git a/benchmarks/tpcc/run.py b/benchmarks/tpcc/run.py index 1e18100..5189240 100644 --- a/benchmarks/tpcc/run.py +++ b/benchmarks/tpcc/run.py @@ -15,8 +15,19 @@ WAL), both kept in --directory. Then: --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). + before it is answered, as Pure does). With --sqlite-group, SQLite again + from each count of terminals handing their transactions to one writer + that commits those waiting together (as pure-serve does), on a copy of + the database checked the same way: after the same transactions, every + row equal to the other SQLite's. --sqlite-pragmas are run on each of + SQLite's connections (TUNED unless given), its tables made with pages + of --sqlite-page-size bytes. Each run reports its CPU a transaction + (Pure's server's, and the share of a core its owner thread took), its + memory (Pure's server's resident, SQLite's peak), and the bytes a + transaction had written: by the process, and to the device holding + --directory. 3. The consistency conditions of clause 3.3.2 on both after each run, each side read from one snapshot. +4. At the end, the bytes Pure's states held after its runs (it is stopped before SQLite's), and the files of both. Prints one JSON line a step. """ @@ -61,6 +72,10 @@ CREATE TABLE history (h_c_w_id INTEGER, h_c_d_id INTEGER, h_c_id INTEGER, h_w_id ORDERS = 10000000 # an order's key is its district's times this, plus its number +# SQLite as a careful engineer runs it (docs/research/106): the database mapped, a page cache large enough for it, +# a savepoint's journal in memory. '' is SQLite's defaults, as research 092 ran it. +TUNED = 'PRAGMA mmap_size=1073741824; PRAGMA cache_size=-65536; PRAGMA temp_store=MEMORY;' + def district_of(key): """A district key's warehouse and district number.""" @@ -87,9 +102,9 @@ def sqlite_rows(tables): yield 'order_line', [(*order(k // 16), k % 16, *rest) for k, *rest in tables['lines']] -def load_sqlite(path, tables): +def load_sqlite(path, tables, page_size): db = sqlite3.connect(str(path)) - db.executescript('PRAGMA journal_mode=WAL; PRAGMA synchronous=OFF;' + SCHEMA) + db.executescript(f'PRAGMA page_size={page_size}; 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) @@ -138,10 +153,30 @@ class Pure: return round(int(line.split()[1]) / 1024, 1) return None + def threads_cpu(self): + """The CPU seconds of the server's threads by name (its owner is pure-owner).""" + seconds = {} + for task in Path(f'/proc/{self.process.pid}/task').iterdir(): + try: + name = (task / 'comm').read_text().strip() + fields = (task / 'stat').read_text().rsplit(')', 1)[1].split() + except OSError: # gone meanwhile + continue + seconds[name] = seconds.get(name, 0) + (int(fields[11]) + int(fields[12])) / os.sysconf('SC_CLK_TCK') + return seconds + + def written(self): + for line in Path(f'/proc/{self.process.pid}/io').read_text().splitlines(): + if line.startswith('write_bytes:'): + return int(line.split()[1]) + return 0 + def stop(self): + """The counts the server prints as it stops (state_bytes among them).""" self.connection.close() self.process.terminate() - self.process.communicate(timeout=120) + lines = self.process.communicate(timeout=120)[1].decode().strip().splitlines() + return json.loads(lines[-1]) if lines and lines[-1].startswith('{') else {} # Each table's rows, times aside, as Pure's queries and SQLite's SELECTs name them (keys packed as tpcc.aup packs them). @@ -225,10 +260,30 @@ def compare(table, pure_rows, 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, each read from one snapshot: the conditions failed (none when consistent).""" +def device_written(directory): + """Bytes written so far to the block device holding directory (None on tmpfs).""" + device = os.stat(directory).st_dev + stat = Path(f'/sys/dev/block/{os.major(device)}:{os.minor(device)}/stat') + return int(stat.read_text().split()[6]) * 512 if stat.exists() else None + + +def per(value, count, scale=1): + return None if value is None else round(value / max(count, 1) * scale, 1) + + +def files_mib(place, prefix): + """The files named prefix... in place, and those in directories so named (the store is one).""" + named = list(place.glob(prefix + '*')) + files = [f for p in named for f in ([p] if p.is_file() else p.rglob('*')) if f.is_file()] + return round(sum(f.stat().st_size for f in files) / 2**20, 1) + + +def consistency(pure, db, sqlite_only=False): + """Clause 3.3.2.1-4 on both (or SQLite's alone), each read from one snapshot: the conditions failed (none when + consistent).""" failures = [] - for system, districts in (('pure', asked(pure, 'district_check')), ('sqlite', db.execute(DISTRICTS[1]).fetchall())): + sides = [] if sqlite_only else [('pure', asked(pure, 'district_check'))] + for system, districts in sides + [('sqlite', db.execute(DISTRICTS[1]).fetchall())]: for key, next_order, newest, newest_new, oldest_new, new_orders, ordered, lines, ytd, district_ytd in districts: if not same(ytd, district_ytd): failures.append((system, '3.3.2.1', ytd, district_ytd)) @@ -254,10 +309,15 @@ def main(): 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')) + parser.add_argument('--sqlite-group', default='', help="SQLite's terminals committing together, a run each") + parser.add_argument('--sqlite-page-size', type=int, default=4096, help="SQLite's page size (its default)") + parser.add_argument('--sqlite-pragmas', default=TUNED, help="run on each of SQLite's connections (SQLITE_PRAGMAS)") args = parser.parse_args() work = Path(tempfile.mkdtemp(prefix='pure-tpcc-')) place = Path(tempfile.mkdtemp(prefix='pure-tpcc-', dir=args.directory)) pure = None + sqlite_env = {**os.environ, 'SQLITE_PRAGMAS': args.sqlite_pragmas} + kinds = ('new_order', 'payment', 'order_status', 'delivery', 'stock_level') try: module = work / 'tpcc.pmod' subprocess.run([str(args.aup), 'build', str(HERE / 'tpcc.aup'), '-o', str(module)], check=True) @@ -274,14 +334,18 @@ def main(): pure_load = time.perf_counter() - began database = place / 'tpcc.db' began = time.perf_counter() - load_sqlite(database, tables) + load_sqlite(database, tables, args.sqlite_page_size) sqlite_load = time.perf_counter() - began + grouped = place / 'tpcc-group.db' + if args.sqlite_group: + shutil.copyfile(database, grouped) 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_load_s': round(sqlite_load, 2), 'sqlite_page_size': args.sqlite_page_size, + '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]) @@ -291,39 +355,80 @@ def main(): 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 = {} + str(args.check)], capture_output=True, text=True, check=True, env=sqlite_env).stdout) + differing, group_differing = {}, {} + group_db = None + if args.sqlite_group: + subprocess.run([str(work / 'sqlite-tpcc'), str(grouped), str(args.warehouses), '1', '3600', args.sync, + str(args.seed), str(args.check), 'group'], capture_output=True, text=True, check=True, + env=sqlite_env) + group_db = sqlite3.connect(str(grouped)) 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()) + expected = db.execute(sql).fetchall() + found = compare(table, asked(pure, f'dump_{table}'), expected) if found: differing[table] = [len(found), found[:3]] - kinds = ('new_order', 'payment', 'order_status', 'delivery', 'stock_level') + found = compare(table, group_db.execute(sql).fetchall(), expected) if group_db else [] + if found: + group_differing[table] = [len(found), found[:3]] + if group_db is not None: + group_db.close() 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) + 'differing': differing, 'grouped_sqlite_differing': group_differing, + 'consistency_failures': consistency(pure, db)}), flush=True) for terminals in (int(t) for t in args.terminals.split(',') if t): - used = pure.cpu_seconds() + os.sync() # what loading and copying left is not written during a run + used, threads_used, written, device = pure.cpu_seconds(), pure.threads_cpu(), pure.written(), \ + device_written(place) + began = time.perf_counter() 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) + took = time.perf_counter() - began used = pure.cpu_seconds() - used - done = sum(result[k]['done'] for k in ('new_order', 'payment', 'order_status', 'delivery', 'stock_level')) + share = {name: round((seconds - threads_used.get(name, 0)) / took, 2) + for name, seconds in pure.threads_cpu().items() if name in ('pure-owner', 'pure-checkpoint')} + done = sum(result[k]['done'] for k in kinds) + device = None if device is None else device_written(place) - device 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(',') if t): + 'cpu_us_per_transaction': per(used, done, 1e6), 'core_shares': share, + 'resident_mib': pure.resident_mib(), + 'written_bytes_per_transaction': per(pure.written() - written, done), + 'device_bytes_per_transaction': per(device, done), + 'consistency_failures': consistency(pure, db)}), flush=True) + resident = pure.resident_mib() + counts, pure = pure.stop(), None # its checkpoints are not written during SQLite's runs + runs = [(database, threads, []) for threads in (int(t) for t in args.sqlite_threads.split(',') if t)] + runs += [(grouped, terminals, ['group']) for terminals in (int(t) for t in args.sqlite_group.split(',') if t)] + for path, threads, mode in runs: + os.sync() + device = device_written(place) 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) + ['taskset', '-c', '0-2', str(work / 'sqlite-tpcc'), str(path), str(args.warehouses), str(threads), + str(args.seconds), args.sync, str(args.seed + threads), *mode], capture_output=True, text=True, + check=True, env=sqlite_env).stdout) + device = None if device is None else device_written(place) - device + done = sum(result[k]['done'] for k in kinds) + checked = sqlite3.connect(str(path)) + print(json.dumps({'step': 'run', **result, 'cpu_us_per_transaction': per(result['cpu_seconds'], done, 1e6), + 'peak_resident_mib': round(result['peak_resident_kib'] / 1024, 1), + 'written_bytes_per_transaction': per(result['written_bytes'], done), + 'device_bytes_per_transaction': per(device, done), + 'consistency_failures': consistency(pure, checked, sqlite_only=True)}), flush=True) + checked.close() db.close() + print(json.dumps({'step': 'end', 'pure_resident_mib': resident, + 'pure_state_mib': round(counts.get('state_bytes', 0) / 2**20, 1), + 'pure_store_mib': files_mib(place, 'store'), 'sqlite_files_mib': files_mib(place, 'tpcc.db'), + 'grouped_sqlite_files_mib': files_mib(place, 'tpcc-group.db'), + 'pure_counts': {k: counts.get(k) for k in ('bursts', 'burst_events', 'checkpoints')}}), + flush=True) finally: if pure: pure.stop() diff --git a/benchmarks/tpcc/sqlite-tpcc.c b/benchmarks/tpcc/sqlite-tpcc.c index 62fb465..f497ae5 100644 --- a/benchmarks/tpcc/sqlite-tpcc.c +++ b/benchmarks/tpcc/sqlite-tpcc.c @@ -3,17 +3,26 @@ * (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]] + * sqlite-tpcc DATABASE WAREHOUSES THREADS SECONDS full|normal|off [SEED [TRANSACTIONS]] [group] * * 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). + * With group, as pure-serve does: the terminals hand their writing + * transactions to one writer, which runs those waiting as one transaction, + * each in a SAVEPOINT (a failed one rolled back to it alone), and commits + * them with one sync; each terminal is answered after it. Order-Status and + * Stock-Level are read on the terminal's own connection from the last + * commit, as pure-serve shows views from the last save (docs/research/106). * 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. */ + * inputs are drawn in tpcc-load's order. SQLITE_PRAGMAS, when set, is run on + * each connection after its own (mmap_size, cache_size). Prints one JSON line + * as tpcc-load does, with the CPU seconds of the run, the peak resident + * memory and the bytes written to storage (/proc/self/io). */ #include +#include #include #include #include @@ -21,13 +30,14 @@ #include #include #include +#include #include 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 const char *database, *synchronous, *pragmas; static int warehouses; static double seconds; static uint64_t seed, each; @@ -104,11 +114,14 @@ typedef struct { } Terminal; static void open_terminal(Terminal *t) { - char pragmas[128]; + char own[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); } + snprintf(own, sizeof(own), "PRAGMA journal_mode=WAL; PRAGMA synchronous=%s;", synchronous); + if (sqlite3_exec(t->db, own, NULL, NULL, NULL) != SQLITE_OK || 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"); @@ -192,15 +205,11 @@ static bool read_only(Terminal *t, int kind, int w, int d, uint64_t *r) { 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) { +/* One transaction of kind, in the transaction the caller began; false when it failed. An unused item's + * New-Order sets *undo: the caller rolls it back, and it counts in undone. */ +static bool body(Terminal *t, int kind, int w, int d, uint64_t *r, bool *undo) { 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; + if (kind == ORDER_STATUS || kind == STOCK_LEVEL) return read_only(t, kind, w, d, r); bool ok = true; if (kind == NEW_ORDER) { int c = nurand(r, 1023, 1, 3000, 259), count = uniform(r, 5, 15), local = 1; @@ -226,8 +235,8 @@ static bool transact(Terminal *t, int kind, int w, int d, uint64_t *r) { 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; + *undo = unused; return unused; } double price = sqlite3_column_double(s, 0); @@ -331,9 +340,81 @@ static bool transact(Terminal *t, int kind, int w, int d, uint64_t *r) { ok = ok && RUN(s); } } - if (ok && RUN(t->commit)) return true; + return ok; +} + +/* One transaction of kind, begun and committed (or rolled back) here; false when it failed. */ +static bool transact(Terminal *t, int kind, int w, int d, uint64_t *r) { + bool reading = kind == ORDER_STATUS || kind == STOCK_LEVEL, undo = false; + if (!RUN(reading ? t->begin_read : t->begin)) return false; + bool ok = body(t, kind, w, d, r, &undo); + if (reading) return RUN(t->commit) && ok; + if (ok && !undo && RUN(t->commit)) return true; RUN(t->rollback); - return false; + return ok && undo; +} + +/* group: the terminals' transactions, waiting for the writer. */ +typedef struct Request { + int kind, w, d; + uint64_t *random; + bool ok; + sem_t answered; /* the terminal alone is woken */ + struct Request *next; +} Request; +static bool grouped, stopping; +static pthread_mutex_t queue_lock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t queued = PTHREAD_COND_INITIALIZER; +static Request *queue_head, **queue_tail = &queue_head; +static uint64_t batches, batched; + +static void *writer(void *unused) { + (void)unused; + Terminal t; + open_terminal(&t); + sqlite3_stmt *mark = prepare(t.db, "SAVEPOINT one"), *keep = prepare(t.db, "RELEASE one"), + *back = prepare(t.db, "ROLLBACK TO one"); + for (;;) { + pthread_mutex_lock(&queue_lock); + while (!queue_head && !stopping) pthread_cond_wait(&queued, &queue_lock); + Request *batch = queue_head; + queue_head = NULL, queue_tail = &queue_head; + pthread_mutex_unlock(&queue_lock); + if (!batch) break; + bool begun = RUN(t.begin); + uint64_t count = 0; + for (Request *q = batch; q; q = q->next, ++count) { + bool undo = false; + q->ok = begun && RUN(mark) && body(&t, q->kind, q->w, q->d, q->random, &undo); + if (begun && (!q->ok || undo)) RUN(back); + if (begun) RUN(keep); + } + bool committed = begun && RUN(t.commit); + if (begun && !committed) RUN(t.rollback); + for (Request *q = batch, *next; q; q = next) { + next = q->next; /* q is the terminal's, gone once answered */ + q->ok = q->ok && committed; + sem_post(&q->answered); + } + ++batches, batched += count; + } + sqlite3_finalize(mark), sqlite3_finalize(keep), sqlite3_finalize(back); + for (sqlite3_stmt *s; (s = sqlite3_next_stmt(t.db, NULL));) sqlite3_finalize(s); + sqlite3_close(t.db); + return NULL; +} + +/* A writing transaction the writer runs with the others waiting then; true once committed. */ +static bool handed(int kind, int w, int d, uint64_t *random) { + Request q = {kind, w, d, random, false, .next = NULL}; + sem_init(&q.answered, 0, 0); + pthread_mutex_lock(&queue_lock); + *queue_tail = &q, queue_tail = &q.next; + pthread_cond_signal(&queued); + pthread_mutex_unlock(&queue_lock); + while (sem_wait(&q.answered)) {} /* EINTR */ + sem_destroy(&q.answered); + return q.ok; } static void *terminal(void *argument) { @@ -347,7 +428,8 @@ static void *terminal(void *argument) { 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); + bool ok = grouped && kind != ORDER_STATUS && kind != STOCK_LEVEL ? handed(kind, home, d, &random) + : transact(&t, kind, home, d, &random); ++done[kind]; if (!ok) ++failed[kind]; ++histogram[kind][bucket(now_us() - started)]; @@ -357,9 +439,30 @@ static void *terminal(void *argument) { return NULL; } +/* This process's CPU seconds, all its threads. */ +static double cpu_seconds(void) { + struct rusage u; + getrusage(RUSAGE_SELF, &u); + return (double)(u.ru_utime.tv_sec + u.ru_stime.tv_sec) + (double)(u.ru_utime.tv_usec + u.ru_stime.tv_usec) / 1e6; +} + +/* A field of a /proc/self file (VmHWM of status, in KiB; write_bytes of io), or 0. */ +static long long proc_field(const char *file, const char *name) { + FILE *f = fopen(file, "r"); + char line[256]; + long long value = 0; + size_t length = strlen(name); + while (f && fgets(line, sizeof(line), f)) + if (!strncmp(line, name, length) && line[length] == ':') value = atoll(line + length + 1); + if (f) fclose(f); + return value; +} + int main(int argc, char **argv) { + grouped = argc > 6 && !strcmp(argv[argc - 1], "group"); + argc -= grouped; if (argc < 6) { - fputs("sqlite-tpcc DATABASE WAREHOUSES THREADS SECONDS full|normal|off [SEED [TRANSACTIONS]]\n", stderr); + fputs("sqlite-tpcc DATABASE WAREHOUSES THREADS SECONDS full|normal|off [SEED [TRANSACTIONS]] [group]\n", stderr); return 2; } database = argv[1]; @@ -369,21 +472,39 @@ int main(int argc, char **argv) { 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)); + pragmas = getenv("SQLITE_PRAGMAS") ? getenv("SQLITE_PRAGMAS") : ""; + for (const char *c = pragmas; *c; ++c) + if (*c == '"' || *c == '\\' || (unsigned char)*c < ' ') { fputs("SQLITE_PRAGMAS: no quotes\n", stderr); return 2; } + pthread_t *thread = calloc((size_t)threads, sizeof(*thread)), writing; + if (grouped) pthread_create(&writing, NULL, writer, NULL); + double cpu = cpu_seconds(); + long long written = proc_field("/proc/self/io", "write_bytes"); 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; + if (grouped) { + pthread_mutex_lock(&queue_lock); + stopping = true; + pthread_cond_signal(&queued); + pthread_mutex_unlock(&queue_lock); + pthread_join(writing, NULL); + } + cpu = cpu_seconds() - cpu; + written = proc_field("/proc/self/io", "write_bytes") - written; 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); + printf("{\"system\":\"sqlite %s%s\",\"synchronous\":\"%s\",\"pragmas\":\"%s\",\"warehouses\":%d,\"terminals\":%d," + "\"transactions_per_second\":%.0f,\"batches\":%llu,\"batched\":%llu,\"new_orders_per_minute\":%.0f," + "\"undone_new_orders\":%llu", + sqlite3_libversion(), grouped ? ", grouped" : "", synchronous, pragmas, warehouses, threads, (double)total / elapsed, + (unsigned long long)batches, (unsigned long long)batched, (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("}"); + printf(",\"cpu_seconds\":%.3f,\"peak_resident_kib\":%lld,\"written_bytes\":%lld}\n", cpu, + proc_field("/proc/self/status", "VmHWM"), written); return 0; } -- cgit v1.2.3