From c4e5df267a1f05d56f1675eeb6715f431dc44e0d Mon Sep 17 00:00:00 2001 From: Alexandre Felipe Date: Wed, 26 Aug 2026 08:29:48 +0100 Subject: [PATCH-v3 2/9] Multi-processing benchmark Improve the previous benchmark by using multiple processe and adding sub groups in each experiment, separating e.g. when a lock waited or not. --- src/test/modules/microbench/Makefile | 3 +- src/test/modules/microbench/bufmap/bench.c | 226 +++++++++++++ .../modules/microbench/bufmap/install.sql | 10 + src/test/modules/microbench/bufmap/query.sql | 35 ++ src/test/modules/microbench/lwlock/bench.c | 154 +++++---- .../modules/microbench/lwlock/install.sql | 6 +- src/test/modules/microbench/lwlock/query.sql | 28 +- src/test/modules/microbench/meson.build | 2 + .../modules/microbench/microbench--1.0.sql | 93 ++++++ .../microbench/microbench-head--1.0.sql | 47 ++- src/test/modules/microbench/multiprocessing.c | 312 ++++++++++++++++++ src/test/modules/microbench/multiprocessing.h | 30 ++ .../modules/microbench/scripts/run-test.sh | 13 +- src/test/modules/microbench/timing-magic.h | 65 +++- 14 files changed, 925 insertions(+), 99 deletions(-) create mode 100644 src/test/modules/microbench/bufmap/bench.c create mode 100644 src/test/modules/microbench/bufmap/install.sql create mode 100644 src/test/modules/microbench/bufmap/query.sql create mode 100644 src/test/modules/microbench/microbench--1.0.sql create mode 100644 src/test/modules/microbench/multiprocessing.c create mode 100644 src/test/modules/microbench/multiprocessing.h diff --git a/src/test/modules/microbench/Makefile b/src/test/modules/microbench/Makefile index 4cdbe4ff0c7..4db6fce35e8 100644 --- a/src/test/modules/microbench/Makefile +++ b/src/test/modules/microbench/Makefile @@ -14,6 +14,7 @@ MICROBENCH_SQL_BUILT = microbench--1.0.sql OBJS = \ $(WIN32RES) \ + multiprocessing.o \ $(addsuffix /bench.o,$(MICROBENCH_TESTS)) # Generated extension script; must be DATA_built so all/install depend on it. @@ -43,7 +44,7 @@ $(MICROBENCH_SQL_BUILT): $(addprefix $(srcdir)/,$(MICROBENCH_SQL_HEAD)) \ # Incremental compile of a single benchmark folder, e.g. make lwlock $(MICROBENCH_TESTS): %: %/bench.o -%/bench.o: %/bench.c randomize.h timing-magic.h +%/bench.o: %/bench.c randomize.h timing-magic.h multiprocessing.h $(COMPILE.c) -I. -o $@ $< .PHONY: run tests list $(MICROBENCH_TESTS) diff --git a/src/test/modules/microbench/bufmap/bench.c b/src/test/modules/microbench/bufmap/bench.c new file mode 100644 index 00000000000..d3f508b0879 --- /dev/null +++ b/src/test/modules/microbench/bufmap/bench.c @@ -0,0 +1,226 @@ +/*------------------------------------------------------------------------- + * + * bufmap/bench.c + * Micro-benchmark BufTable insert/lookup/delete with synthetic tags. + * No anchor table required - uses free buffer slots directly. + * + * IDENTIFICATION + * src/test/modules/microbench/bufmap/bench.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/table.h" +#include "catalog/namespace.h" +#include "fmgr.h" +#include "funcapi.h" +#include "miscadmin.h" +#include "storage/buf_internals.h" +#include "storage/bufmgr.h" +#include "storage/lwlock.h" +#include "utils/builtins.h" +#include "utils/rel.h" +#include "utils/tuplestore.h" + +#include "multiprocessing.h" +#include "randomize.h" +#include "timing-magic.h" + +/* synthetic relfilenodes for bench tags — unlikely to collide with anything real */ +#define BENCH_SPC_OID 0xB0B0 +#define BENCH_DB_OID 0xB1B1 +#define BENCH_REL_PRESENT ((RelFileNumber) 0x7E570001) +#define BENCH_REL_ABSENT ((RelFileNumber) 0x7E570002) + +PG_FUNCTION_INFO_V1(bench_bufmap); + +static void +run_bufmap_bench(int proc_id, int n_parallel, int rounds, int iterations, + ReturnSetInfo *rsinfo) +{ + intptr_t *blks; + BufferTag *ptags; + BufferTag *atags; + intptr_t *bufids; + pg_prng_state rng; + int nfree = 0; + LWLock *arbitrary_lock; + volatile int64 sink PG_USED_FOR_ASSERTS_ONLY = 0; + int start_buf; + RelFileLocator rp = {.spcOid = BENCH_SPC_OID, .dbOid = BENCH_DB_OID, .relNumber = BENCH_REL_PRESENT}; + RelFileLocator ra = {.spcOid = BENCH_SPC_OID, .dbOid = BENCH_DB_OID, .relNumber = BENCH_REL_ABSENT}; + + blks = palloc(sizeof(intptr_t) * iterations); + ptags = palloc0(sizeof(BufferTag) * iterations); + atags = palloc0(sizeof(BufferTag) * iterations); + bufids = palloc(sizeof(intptr_t) * iterations); + pg_prng_seed(&rng, 0xB0FF0A00 ^ (uint64)proc_id); + + /* + * Each worker collects buffer IDs from a distinct range to avoid spinlock + * contention. Worker 1 starts at buffer 0, worker 2 at iterations, etc. + * We need n_parallel * iterations total free buffers. + */ + start_buf = (proc_id - 1) * iterations; + + elog(LOG, "Worker %d: starting buffer collection from %d", proc_id, start_buf); + + arbitrary_lock = BufMappingPartitionLock(0); + LWLockAcquire(arbitrary_lock, LW_EXCLUSIVE); + for (int i = start_buf; i < NBuffers && nfree < iterations; i++) + { + BufferDesc *desc = GetBufferDescriptor(i); + uint64 state = pg_atomic_read_u64(&desc->state); + + if (!(state & (BM_TAG_VALID | BM_LOCKED | BUF_REFCOUNT_MASK))) + { + state = LockBufHdr(desc); + UnlockBufHdrExt(desc, state, 0, 0, 1); + bufids[nfree++] = (intptr_t) i; + } + } + LWLockRelease(arbitrary_lock); + elog(LOG, "Worker %d: collected %d buffers", proc_id, nfree); + + if (nfree < iterations) + goto teardown; + + for (int i = 0; i < iterations; ++i) + { + int idx = n_parallel * i + proc_id - 1; + blks[i] = (intptr_t) i; + + InitBufferTag(&ptags[i], &rp, MAIN_FORKNUM, (BlockNumber) idx); + InitBufferTag(&atags[i], &ra, MAIN_FORKNUM, (BlockNumber) idx); + } + + if (!timing_initialized) + pg_initialize_timing(); + + elog(LOG, "Worker %d: setup complete, entering timing loops", proc_id); + + for (int64 r = 0; r < rounds; r++) + { + INIT_TIMING_SCOPE(); + shuffle_pointers(&rng, (void **) blks, iterations); + shuffle_pointers(&rng, (void **) bufids, iterations); + + BEGIN_GROUPED_TIMING("insert", iterations, 2) + { + BufferTag *tag = &ptags[blks[i]]; + uint32 hash; + LWLock *lock; + + hash = BufTableHashCode(tag); + lock = BufMappingPartitionLock(hash); + group_id = LWLockAcquire(lock, LW_EXCLUSIVE) ? 0 : 1; + sink += BufTableInsert(tag, hash, (Buffer)bufids[i]); + LWLockRelease(lock); + } + END_GROUPED_TIMING; + + BEGIN_GROUPED_TIMING("hit", iterations, 2) + { + BufferTag *tag = &ptags[blks[i]]; + uint32 hash; + LWLock *lock; + + hash = BufTableHashCode(tag); + lock = BufMappingPartitionLock(hash); + group_id = LWLockAcquire(lock, LW_SHARED) ? 0 : 1; + BufTableLookup(tag, hash); + LWLockRelease(lock); + } + END_GROUPED_TIMING; + + BEGIN_GROUPED_TIMING("miss", iterations, 2) + { + BufferTag *tag = &atags[blks[i]]; + uint32 hash; + LWLock *lock; + + hash = BufTableHashCode(tag); + lock = BufMappingPartitionLock(hash); + group_id = LWLockAcquire(lock, LW_SHARED) ? 0 : 1; + BufTableLookup(tag, hash); + LWLockRelease(lock); + } + END_GROUPED_TIMING; + + BEGIN_GROUPED_TIMING("delete", iterations, 2) + { + BufferTag *tag = &ptags[blks[i]]; + uint32 hash; + LWLock *lock; + + hash = BufTableHashCode(tag); + lock = BufMappingPartitionLock(hash); + group_id = LWLockAcquire(lock, LW_EXCLUSIVE) ? 0 : 1; + BufTableDelete(tag, hash); + LWLockRelease(lock); + } + END_GROUPED_TIMING; + + BEGIN_GROUPED_TIMING("LWLock", iterations, 2) + { + BufferTag *tag = &ptags[blks[i]]; + uint32 hash; + LWLock *lock; + + hash = BufTableHashCode(tag); + lock = BufMappingPartitionLock(hash); + group_id = LWLockAcquire(lock, LW_EXCLUSIVE) ? 0 : 1; + LWLockRelease(lock); + } + END_GROUPED_TIMING; + } + +teardown: + for (int i = 0; i < nfree; ++i){ + BufferDesc *desc = GetBufferDescriptor(bufids[i]); + uint64 state = LockBufHdr(desc); + UnlockBufHdrExt(desc, state, 0, 0, -1); + } + + pfree(bufids); + pfree(ptags); + pfree(atags); + pfree(blks); + (void) sink; + if(nfree < iterations) + elog(ERROR, "Couldn't get enough free buffers"); +} + +static void +microbench_parallel_work(int proc_id, int n_parallel, int rounds, int iterations) +{ + run_bufmap_bench(proc_id, n_parallel, rounds, iterations, NULL); +} + +/* + * bench_bufmap(n_parallel, rounds, iterations) -> SETOF microbench_sample + */ +Datum +bench_bufmap(PG_FUNCTION_ARGS) +{ + int n_parallel = PG_ARGISNULL(0) ? 1 : PG_GETARG_INT32(0); + int rounds = PG_ARGISNULL(1) ? 1 : PG_GETARG_INT64(1); + int iterations = PG_ARGISNULL(2) ? 128 : PG_GETARG_INT64(2); + ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; + int proc_id; + + if (n_parallel <= 0 || rounds <= 0 || iterations <= 0) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("n_parallel, rounds, and iterations must be positive"))); + + InitMaterializedSRF(fcinfo, 0); + + proc_id = replicate_backend(n_parallel, rounds, iterations, 2, + microbench_parallel_work); + run_bufmap_bench(proc_id, n_parallel, rounds, iterations, rsinfo); + microbench_mp_leave(rsinfo); + + return (Datum) 0; +} diff --git a/src/test/modules/microbench/bufmap/install.sql b/src/test/modules/microbench/bufmap/install.sql new file mode 100644 index 00000000000..4d5fa39e096 --- /dev/null +++ b/src/test/modules/microbench/bufmap/install.sql @@ -0,0 +1,10 @@ +CREATE FUNCTION bench_bufmap( + IN n_parallel int4 DEFAULT 1, + IN rounds int8 DEFAULT 1, + IN iterations int8 DEFAULT 128 +) +RETURNS SETOF microbench_sample +AS 'MODULE_PATHNAME', 'bench_bufmap' +LANGUAGE C; + +REVOKE ALL ON FUNCTION bench_bufmap(int4, int8, int8) FROM PUBLIC; diff --git a/src/test/modules/microbench/bufmap/query.sql b/src/test/modules/microbench/bufmap/query.sql new file mode 100644 index 00000000000..29b50260006 --- /dev/null +++ b/src/test/modules/microbench/bufmap/query.sql @@ -0,0 +1,35 @@ +-- bufmap micro-benchmark query +-- +-- 1 row = 1 buffer: pad is STORAGE PLAIN and larger than half a page. +-- Grow anchor to 90% of shared_buffers if needed. iterations stays the +-- runner value so n_parallel * iterations fits in that heap. + +\pset format aligned + +\if :{?rounds} +\else +\set rounds 1000 +\endif +\if :{?iterations} +\else +\set iterations 128 +\endif +\if :{?max_parallel} +\else +\set max_parallel 16 +\endif + +\timing on + +SELECT format($q$ +SELECT %s AS workers, +op || coalesce(' / ' || "group"::text, '') as "op / wait" +, avg, q1, med, q3, count +FROM format_microbench_with_count( + (SELECT array_agg(s ORDER BY s.id)s + FROM bench_bufmap(%s, %s::int8, (%s / %s)::int8) AS s) +) bench_stats +ORDER BY 1,2; +$q$, i, i, :rounds, :iterations, i) +FROM generate_series(1, :max_parallel) AS i +\gexec diff --git a/src/test/modules/microbench/lwlock/bench.c b/src/test/modules/microbench/lwlock/bench.c index 33d75770a0d..101cd7423c1 100644 --- a/src/test/modules/microbench/lwlock/bench.c +++ b/src/test/modules/microbench/lwlock/bench.c @@ -18,52 +18,29 @@ #include "utils/builtins.h" #include "utils/tuplestore.h" +#include "multiprocessing.h" #include "randomize.h" #include "timing-magic.h" + PG_MODULE_MAGIC; PG_FUNCTION_INFO_V1(bench_lwlock); -/* - * bench_lwlock(n, rounds, randomize) -> SETOF - * (op text, avg_ns float8, batch_size int8, id int8) - */ -Datum -bench_lwlock(PG_FUNCTION_ARGS) +static void +run_lwlock_bench(int proc_id, int n_parallel, int rounds, int iterations, + ReturnSetInfo *rsinfo) { - int64 n = PG_GETARG_INT64(0); - int64 rounds = PG_ARGISNULL(1) ? 1 : PG_GETARG_INT64(1); - bool randomize = PG_ARGISNULL(2) ? true : PG_GETARG_BOOL(2); - ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; - LWLock **locks; - volatile int64 sink PG_USED_FOR_ASSERTS_ONLY = 0; - - if (n <= 0 || rounds <= 0) - ereport(ERROR, - (errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("n and rounds must be positive"))); + LWLock **locks; + pg_prng_state rng; + volatile int64 sink PG_USED_FOR_ASSERTS_ONLY = 0; - if (n > NUM_BUFFER_PARTITIONS) - n = NUM_BUFFER_PARTITIONS; + (void) n_parallel; - InitMaterializedSRF(fcinfo, 0); - - locks = palloc(sizeof(LWLock *) * n); - if (randomize) - { - pg_prng_state rng; - - pg_prng_seed(&rng, 0xDA7ABA5E); - for (int i = 0; i < n; i++) - locks[i] = BufMappingPartitionLock(i); - shuffle_pointers(&rng, (void **) locks, (int) n); - } - else - { - for (int i = 0; i < n; i++) - locks[i] = BufMappingPartitionLock(0); - } + locks = palloc(sizeof(LWLock *) * iterations); + pg_prng_seed(&rng, 0xDA7ABA5E ^ (uint64) proc_id * 0x5EED); + for (int i = 0; i < iterations; i++) + locks[i] = BufMappingPartitionLock(i); if (!timing_initialized) pg_initialize_timing(); @@ -71,53 +48,88 @@ bench_lwlock(PG_FUNCTION_ARGS) for (int64 r = 0; r < rounds; r++) { INIT_TIMING_SCOPE(); - { - BufferDesc *buf_desc = GetBufferDescriptor(1); - BEGIN_TIMING("spin-lock", n) - { - LockBufHdr(buf_desc); - UnlockBufHdr(buf_desc); - } - END_TIMING; - } - BEGIN_TIMING("LWLock-ex", n) + /* + * Shuffling every round to reduce collision distribution bias + */ + shuffle_pointers(&rng, (void **) locks, iterations); { - LWLock *lock = locks[i]; + BufferDesc *buf_desc = GetBufferDescriptor(1); - LWLockAcquire(lock, LW_EXCLUSIVE); - LWLockRelease(lock); + BEGIN_GROUPED_TIMING("spin-lock", iterations, 2) + { + LockBufHdr(buf_desc); + UnlockBufHdr(buf_desc); + } + END_GROUPED_TIMING; } - END_TIMING; - BEGIN_TIMING("LWLock-sh", n) - { - LWLock *lock = locks[i]; + BEGIN_GROUPED_TIMING("LWLock-ex", iterations, 2) + { + LWLock *lock = locks[i]; - LWLockAcquire(lock, LW_SHARED); - LWLockRelease(lock); - } - END_TIMING; + group_id = LWLockAcquire(lock, LW_EXCLUSIVE) ? 0 : 1; + LWLockRelease(lock); + } + END_GROUPED_TIMING; - BEGIN_TIMING("LWLock-cond", n) - { - LWLock *lock = locks[i]; - if (!LWLockConditionalAcquire(lock, LW_SHARED)) - elog(ERROR, "Failed to acquire lock"); - LWLockRelease(lock); - } - END_TIMING; + BEGIN_GROUPED_TIMING("LWLock-cond", iterations, 2) + { + LWLock *lock = locks[i]; + + if (LWLockConditionalAcquire(lock, LW_EXCLUSIVE)) + { + group_id = 0; + LWLockRelease(lock); + } + else + group_id = 1; + } + END_GROUPED_TIMING; - BEGIN_TIMING("nop", n) - { - LWLock *lock = locks[i]; + BEGIN_GROUPED_TIMING("nop", iterations, 2) + { + LWLock *lock = locks[i]; - sink += (int64) (uintptr_t) lock; - } - END_TIMING; + sink += (int64) (uintptr_t) lock; + } + END_GROUPED_TIMING; } (void) sink; +} + +static void +microbench_parallel_work(int proc_id, int n_parallel, int rounds, int iterations) +{ + run_lwlock_bench(proc_id, n_parallel, rounds, iterations, NULL); +} + +/* + * bench_lwlock(n_parallel, rounds, iterations) -> SETOF + * (op text, avg_ns float8, batch_size int8, id int8, group int8) + */ +Datum +bench_lwlock(PG_FUNCTION_ARGS) +{ + int n_parallel = PG_ARGISNULL(0) ? 1 : PG_GETARG_INT32(0); + int rounds = PG_ARGISNULL(1) ? 1 : PG_GETARG_INT64(1); + int iterations = PG_ARGISNULL(2) ? 128 : PG_GETARG_INT64(2); + ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; + int proc_id; + + if (n_parallel <= 0 || rounds <= 0 || iterations <= 0) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("n_parallel, rounds, and iterations must be positive"))); + + InitMaterializedSRF(fcinfo, 0); + + proc_id = replicate_backend(n_parallel, rounds, iterations, 2, + microbench_parallel_work); + run_lwlock_bench(proc_id, n_parallel, rounds, iterations, rsinfo); + microbench_mp_leave(rsinfo); + return (Datum) 0; } diff --git a/src/test/modules/microbench/lwlock/install.sql b/src/test/modules/microbench/lwlock/install.sql index 9bf234837b1..2aa02c42101 100644 --- a/src/test/modules/microbench/lwlock/install.sql +++ b/src/test/modules/microbench/lwlock/install.sql @@ -1,10 +1,10 @@ CREATE FUNCTION bench_lwlock( - IN n int8, + IN n_parallel int4 DEFAULT 1, IN rounds int8 DEFAULT 1, - IN random bool DEFAULT true + IN iterations int8 DEFAULT 128 ) RETURNS SETOF microbench_sample AS 'MODULE_PATHNAME', 'bench_lwlock' LANGUAGE C; -REVOKE ALL ON FUNCTION bench_lwlock(int8, int8, bool) FROM PUBLIC; +REVOKE ALL ON FUNCTION bench_lwlock(int4, int8, int8) FROM PUBLIC; diff --git a/src/test/modules/microbench/lwlock/query.sql b/src/test/modules/microbench/lwlock/query.sql index 16c1fd1dfe4..12145d349dd 100644 --- a/src/test/modules/microbench/lwlock/query.sql +++ b/src/test/modules/microbench/lwlock/query.sql @@ -1,8 +1,30 @@ -- lwlock micro-benchmark query +-- +-- :rounds and :iterations are psql variables (run-test.sh sets them). +-- They are not expanded inside dollar-quoted strings. \gexec runs one +-- SELECT per n_parallel so each result set prints as soon as that size +-- finishes. \pset format aligned -SELECT * FROM format_microbench( +\if :{?rounds} +\else +\set rounds 1000 +\endif +\if :{?iterations} +\else +\set iterations 128 +\endif + + +\timing on + +SELECT format($q$ +SELECT %s AS n_parallel, bench_stats.* +FROM format_microbench_with_count( (SELECT array_agg(s ORDER BY s.id) - FROM bench_lwlock(:'n'::int8, :'rounds'::int8, true) AS s) -); + FROM bench_lwlock(%s, %s::int8, %s::int8) AS s) +) bench_stats; +$q$, i, i, :rounds, :iterations) +FROM generate_series(1, 4) AS i +\gexec diff --git a/src/test/modules/microbench/meson.build b/src/test/modules/microbench/meson.build index 9836640a3df..0ac64174e70 100644 --- a/src/test/modules/microbench/meson.build +++ b/src/test/modules/microbench/meson.build @@ -5,6 +5,8 @@ microbench_tests = [ ] microbench_sources = files( + 'multiprocessing.c', +) + files( '@0@/bench.c'.format(test) for test in microbench_tests ) diff --git a/src/test/modules/microbench/microbench--1.0.sql b/src/test/modules/microbench/microbench--1.0.sql new file mode 100644 index 00000000000..7058f2dfa03 --- /dev/null +++ b/src/test/modules/microbench/microbench--1.0.sql @@ -0,0 +1,93 @@ +/* src/test/modules/microbench/microbench--1.0.sql */ +/* Generated from microbench--1.0.sql.head and per-test install.sql files. */ + +-- complain if script is sourced in psql, rather than via CREATE EXTENSION +\echo Use "CREATE EXTENSION microbench" to load this file. \quit + +CREATE DOMAIN microbench_format AS numeric(15, 2); + +CREATE TYPE microbench_sample AS ( + op text, + avg_ns float8, + batch_size int8, + id int8, + "group" int8 +); + +CREATE TYPE microbench_stats AS ( + op text, + "group" int8, + avg microbench_format, + min microbench_format, + q1 microbench_format, + med microbench_format, + q3 microbench_format, + max microbench_format, + std microbench_format +); + + +CREATE TYPE microbench_stats_with_count AS ( + op text, + "group" int8, + avg microbench_format, + min microbench_format, + q1 microbench_format, + med microbench_format, + q3 microbench_format, + max microbench_format, + std microbench_format, + count int8 +); + +CREATE FUNCTION format_microbench(samples microbench_sample[]) +RETURNS SETOF microbench_stats +LANGUAGE sql +STABLE +AS $$ + SELECT + s.op, + s."group", + sum(s.avg_ns * s.batch_size) / NULLIF(sum(s.batch_size), 0), + min(s.avg_ns), + percentile_cont(0.25) WITHIN GROUP (ORDER BY s.avg_ns), + percentile_cont(0.50) WITHIN GROUP (ORDER BY s.avg_ns), + percentile_cont(0.75) WITHIN GROUP (ORDER BY s.avg_ns), + max(s.avg_ns), + stddev(s.avg_ns) + FROM unnest(samples) AS s + GROUP BY s.op, s."group" + ORDER BY min(s.id), s."group" NULLS FIRST; +$$; + + +CREATE FUNCTION format_microbench_with_count(samples microbench_sample[]) +RETURNS SETOF microbench_stats_with_count +LANGUAGE sql +STABLE +AS $$ + SELECT + s.op, + s."group", + sum(s.avg_ns * s.batch_size) / NULLIF(sum(s.batch_size), 0), + min(s.avg_ns), + percentile_cont(0.25) WITHIN GROUP (ORDER BY s.avg_ns), + percentile_cont(0.50) WITHIN GROUP (ORDER BY s.avg_ns), + percentile_cont(0.75) WITHIN GROUP (ORDER BY s.avg_ns), + max(s.avg_ns), + stddev(s.avg_ns), + sum(s.batch_size) + FROM unnest(samples) AS s + GROUP BY s.op, s."group" + ORDER BY min(s.id), s."group" NULLS FIRST; +$$; +CREATE FUNCTION bench_lwlock( + IN n_parallel int4 DEFAULT 1, + IN rounds int8 DEFAULT 1, + IN iterations int8 DEFAULT 128 +) +RETURNS SETOF microbench_sample +AS 'MODULE_PATHNAME', 'bench_lwlock' +LANGUAGE C; + +REVOKE ALL ON FUNCTION bench_lwlock(int4, int8, int8) FROM PUBLIC; diff --git a/src/test/modules/microbench/microbench-head--1.0.sql b/src/test/modules/microbench/microbench-head--1.0.sql index 48c5143843a..d74a6442fca 100644 --- a/src/test/modules/microbench/microbench-head--1.0.sql +++ b/src/test/modules/microbench/microbench-head--1.0.sql @@ -10,11 +10,13 @@ CREATE TYPE microbench_sample AS ( op text, avg_ns float8, batch_size int8, - id int8 + id int8, + "group" int8 ); CREATE TYPE microbench_stats AS ( op text, + "group" int8, avg microbench_format, min microbench_format, q1 microbench_format, @@ -24,6 +26,20 @@ CREATE TYPE microbench_stats AS ( std microbench_format ); + +CREATE TYPE microbench_stats_with_count AS ( + op text, + "group" int8, + avg microbench_format, + min microbench_format, + q1 microbench_format, + med microbench_format, + q3 microbench_format, + max microbench_format, + std microbench_format, + count int8 +); + CREATE FUNCTION format_microbench(samples microbench_sample[]) RETURNS SETOF microbench_stats LANGUAGE sql @@ -31,7 +47,8 @@ STABLE AS $$ SELECT s.op, - avg(s.avg_ns), + s."group", + sum(s.avg_ns * s.batch_size) / NULLIF(sum(s.batch_size), 0), min(s.avg_ns), percentile_cont(0.25) WITHIN GROUP (ORDER BY s.avg_ns), percentile_cont(0.50) WITHIN GROUP (ORDER BY s.avg_ns), @@ -39,6 +56,28 @@ AS $$ max(s.avg_ns), stddev(s.avg_ns) FROM unnest(samples) AS s - GROUP BY s.op - ORDER BY min(s.id); + GROUP BY s.op, s."group" + ORDER BY min(s.id), s."group" NULLS FIRST; +$$; + + +CREATE FUNCTION format_microbench_with_count(samples microbench_sample[]) +RETURNS SETOF microbench_stats_with_count +LANGUAGE sql +STABLE +AS $$ + SELECT + s.op, + s."group", + sum(s.avg_ns * s.batch_size) / NULLIF(sum(s.batch_size), 0), + min(s.avg_ns), + percentile_cont(0.25) WITHIN GROUP (ORDER BY s.avg_ns), + percentile_cont(0.50) WITHIN GROUP (ORDER BY s.avg_ns), + percentile_cont(0.75) WITHIN GROUP (ORDER BY s.avg_ns), + max(s.avg_ns), + stddev(s.avg_ns), + sum(s.batch_size) + FROM unnest(samples) AS s + GROUP BY s.op, s."group" + ORDER BY min(s.id), s."group" NULLS FIRST; $$; diff --git a/src/test/modules/microbench/multiprocessing.c b/src/test/modules/microbench/multiprocessing.c new file mode 100644 index 00000000000..deb1dba399a --- /dev/null +++ b/src/test/modules/microbench/multiprocessing.c @@ -0,0 +1,312 @@ +/*------------------------------------------------------------------------- + * + * multiprocessing.c + * Launch and synchronize micro-benchmark parallel workers. + * + * Follows the parallel btree-build pattern: EnterParallelMode, + * CreateParallelContext, shm_toc, LaunchParallelWorkers, + * WaitForParallelWorkersToFinish. Cooperating backends rendezvous with + * an atomic spin barrier (not BarrierArriveAndWait) so sync stays off + * the kernel path. Worker timing rows are stored in DSM and copied + * into the leader tuplestore at each synchronize and at leave. + * + * IDENTIFICATION + * src/test/modules/microbench/multiprocessing.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include + +#include "access/parallel.h" +#include "access/xact.h" +#include "funcapi.h" +#include "miscadmin.h" +#include "port/atomics.h" +#include "storage/s_lock.h" +#include "utils/builtins.h" +#include "utils/tuplestore.h" + +#include "multiprocessing.h" + +#define PARALLEL_KEY_MICROBENCH_SHARED UINT64CONST(0xA1100001) +#define MICROBENCH_MAX_OPS_PER_ROUND 32 +#define MICROBENCH_OP_LEN 64 + +typedef struct MicrobenchSample +{ + char op[MICROBENCH_OP_LEN]; + double avg_ns; + int64 batch_size; + int64 id; + int64 group; + bool group_isnull; +} MicrobenchSample; + +typedef struct MicrobenchShared +{ + pg_atomic_uint32 ready; + pg_atomic_uint32 nsamples; + pg_atomic_uint32 arrived; + pg_atomic_uint32 generation; + int n_parallel; + int rounds; + int iterations; + int max_group_size; + int max_samples; + ptrdiff_t work_off; + MicrobenchSample samples[FLEXIBLE_ARRAY_MEMBER]; +} MicrobenchShared; + +extern PGDLLEXPORT void microbench_parallel_main(dsm_segment *seg, shm_toc *toc); + +static MicrobenchShared *microbench_mp_state = NULL; +static ParallelContext *microbench_mp_pcxt = NULL; +static int microbench_mp_id = 0; + +static void +microbench_mp_flush_samples(ReturnSetInfo *rsinfo) +{ + MicrobenchShared *shared = microbench_mp_state; + uint32 n; + uint32 i; + Datum values[5]; + bool nulls[5] = {0}; + + if (shared == NULL || rsinfo == NULL || rsinfo->setResult == NULL) + return; + + n = pg_atomic_read_u32(&shared->nsamples); + if (n > (uint32) shared->max_samples) + n = (uint32) shared->max_samples; + + for (i = 0; i < n; i++) + { + MicrobenchSample *s = &shared->samples[i]; + + values[0] = CStringGetTextDatum(s->op); + values[1] = Float8GetDatum(s->avg_ns); + values[2] = Int64GetDatum(s->batch_size); + values[3] = Int64GetDatum(s->id); + values[4] = Int64GetDatum(s->group); + nulls[4] = s->group_isnull; + tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls); + } + if(!pg_atomic_compare_exchange_u32(&shared->nsamples, &n, 0)) + { + elog(FATAL, "Race condition while flushing results."); + } +} + +static void +microbench_mp_teardown(ReturnSetInfo *rsinfo) +{ + if (IsParallelWorker()) + return; + + if (microbench_mp_pcxt != NULL) + { + WaitForParallelWorkersToFinish(microbench_mp_pcxt); + microbench_mp_flush_samples(rsinfo); + DestroyParallelContext(microbench_mp_pcxt); + microbench_mp_pcxt = NULL; + ExitParallelMode(); + } + + microbench_mp_state = NULL; + microbench_mp_id = 0; +} + +bool +microbench_mp_recording(void) +{ + return microbench_mp_state != NULL && microbench_mp_state->n_parallel > 1; +} + +void +microbench_mp_emit_sample(ReturnSetInfo *rsinfo, const char *op, + double avg_ns, int64 batch_size, int64 id, + int64 group, bool group_isnull) +{ + if (rsinfo == NULL) + { + MicrobenchShared *shared = microbench_mp_state; + uint32 slot; + + slot = pg_atomic_fetch_add_u32(&shared->nsamples, 1); + if (slot >= (uint32) shared->max_samples) + elog(ERROR, "microbench sample buffer overflow"); + + strlcpy(shared->samples[slot].op, op, MICROBENCH_OP_LEN); + shared->samples[slot].avg_ns = avg_ns; + shared->samples[slot].batch_size = batch_size; + shared->samples[slot].id = id; + shared->samples[slot].group = group; + shared->samples[slot].group_isnull = group_isnull; + } + else if (rsinfo != NULL && rsinfo->setResult != NULL) + { + Datum values[5]; + bool nulls[5] = {0}; + + values[0] = CStringGetTextDatum(op); + values[1] = Float8GetDatum(avg_ns); + values[2] = Int64GetDatum(batch_size); + values[3] = Int64GetDatum(id); + values[4] = Int64GetDatum(group); + nulls[4] = group_isnull; + tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls); + } +} + +int +replicate_backend(int n_parallel, int rounds, int iterations, int groups, + microbench_parallel_work_fn work) +{ + ParallelContext *pcxt; + MicrobenchShared *shared; + int nworkers = n_parallel - 1; + int max_samples = nworkers * groups; + int shared_size = offsetof(MicrobenchShared, samples) + + max_samples * sizeof(MicrobenchSample); + if (n_parallel <= 0) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("n_parallel must be positive"))); + + if (n_parallel <= 1) + { + microbench_mp_id = 1; + return 1; + } + + if (microbench_mp_id != 0) + return microbench_mp_id; + + EnterParallelMode(); + pcxt = CreateParallelContext("microbench", "microbench_parallel_main", + nworkers); + shm_toc_estimate_chunk(&pcxt->estimator, shared_size); + shm_toc_estimate_keys(&pcxt->estimator, 1); + InitializeParallelDSM(pcxt); + + if (pcxt->seg == NULL) + { + DestroyParallelContext(pcxt); + ExitParallelMode(); + microbench_mp_id = 1; + return 1; + } + + shared = (MicrobenchShared *) shm_toc_allocate(pcxt->toc, shared_size); + memset(shared, 0, shared_size); + pg_atomic_init_u32(&shared->ready, 0); + pg_atomic_init_u32(&shared->nsamples, 0); + pg_atomic_init_u32(&shared->arrived, 0); + pg_atomic_init_u32(&shared->generation, 0); + shared->max_samples = max_samples; + shared->n_parallel = n_parallel; + shared->rounds = rounds; + shared->iterations = iterations; + shared->work_off = (uintptr_t) work - (uintptr_t) microbench_parallel_main; + shm_toc_insert(pcxt->toc, PARALLEL_KEY_MICROBENCH_SHARED, shared); + + LaunchParallelWorkers(pcxt); + + if (pcxt->nworkers_launched < nworkers) + { + int got = pcxt->nworkers_launched; + + DestroyParallelContext(pcxt); + ExitParallelMode(); + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_RESOURCES), + errmsg("could not launch %d parallel workers (got %d)", + nworkers, got), + errhint("Increase max_worker_processes and max_parallel_workers."))); + } + + microbench_mp_state = shared; + microbench_mp_pcxt = pcxt; + microbench_mp_id = 1; + pg_atomic_write_membarrier_u32(&shared->ready, 1); + + WaitForParallelWorkersToAttach(pcxt); + return 1; +} + +static void +wait_eq(volatile pg_atomic_uint32 *ptr, uint32 value) +{ + uint32 spins = 0; + + while (pg_atomic_read_membarrier_u32(ptr) != value) + { + SPIN_DELAY(); + if ((++spins & 0xFFFF) == 0) + CHECK_FOR_INTERRUPTS(); + } +} + +void +synchronize_backends(ReturnSetInfo *rsinfo) +{ + MicrobenchShared *shared = microbench_mp_state; + uint32 gen; + + if (shared == NULL || shared->n_parallel <= 1) + return; + + /* + * Sense-reversing spin barrier. Capture the generation first, then + * announce arrival. The leader waits until everyone is here, copies + * samples, then advances the generation so waiters return together. + */ + gen = pg_atomic_read_u32(&shared->generation); + pg_atomic_add_fetch_u32(&shared->arrived, 1); + + if (!IsParallelWorker()) + { + wait_eq(&shared->arrived, (uint32) shared->n_parallel); + microbench_mp_flush_samples(rsinfo); + pg_atomic_write_u32(&shared->arrived, 0); + pg_atomic_write_membarrier_u32(&shared->generation, gen + 1); + } + else + wait_eq(&shared->generation, gen + 1); +} + +void +microbench_mp_leave(ReturnSetInfo *rsinfo) +{ + microbench_mp_teardown(rsinfo); +} + +/* + * Parallel-worker entry point. Looked up by name in the worker process, + * the same way _bt_parallel_build_main is. + */ +PGDLLEXPORT void +microbench_parallel_main(dsm_segment *seg, shm_toc *toc) +{ + MicrobenchShared *shared; + microbench_parallel_work_fn work; + uint32 spins = 0; + + shared = shm_toc_lookup(toc, PARALLEL_KEY_MICROBENCH_SHARED, false); + + while (pg_atomic_read_u32(&shared->ready) == 0) + { + SPIN_DELAY(); + if ((++spins & 0xFFFF) == 0) + CHECK_FOR_INTERRUPTS(); + } + + microbench_mp_state = shared; + microbench_mp_id = ParallelWorkerNumber + 2; + work = (microbench_parallel_work_fn) + ((uintptr_t) microbench_parallel_main + shared->work_off); + work(microbench_mp_id, shared->n_parallel, + shared->rounds, shared->iterations); +} diff --git a/src/test/modules/microbench/multiprocessing.h b/src/test/modules/microbench/multiprocessing.h new file mode 100644 index 00000000000..4d8c2ac0e01 --- /dev/null +++ b/src/test/modules/microbench/multiprocessing.h @@ -0,0 +1,30 @@ +#ifndef MICROBENCH_MULTIPROCESSING_H +#define MICROBENCH_MULTIPROCESSING_H + +struct ReturnSetInfo; + +typedef void (*microbench_parallel_work_fn) (int proc_id, int n_parallel, + int rounds, int iterations); + +/* + * Launch n-1 parallel workers (same infrastructure as parallel index + * builds) and return 1 in the leader. Workers enter through + * microbench_parallel_main and call the work function passed here. + * + * synchronize_backends() spins until the whole party has arrived, the + * leader flushes new DSM samples, then it releases everyone. It is a + * no-op when n_parallel <= 1. microbench_mp_leave() waits for workers + * and flushes the last batch, then exits parallel mode. + */ +extern int replicate_backend(int n_parallel, int rounds, int iterations, + int samples_per_round, + microbench_parallel_work_fn work); +extern void synchronize_backends(struct ReturnSetInfo *rsinfo); +extern void microbench_mp_leave(struct ReturnSetInfo *rsinfo); +extern bool microbench_mp_recording(void); +extern void microbench_mp_emit_sample(struct ReturnSetInfo *rsinfo, + const char *op, double avg_ns, + int64 batch_size, int64 id, + int64 group, bool group_isnull); + +#endif diff --git a/src/test/modules/microbench/scripts/run-test.sh b/src/test/modules/microbench/scripts/run-test.sh index ee9e29d7b49..24dc72e648d 100755 --- a/src/test/modules/microbench/scripts/run-test.sh +++ b/src/test/modules/microbench/scripts/run-test.sh @@ -4,7 +4,8 @@ # # Usage: run-test.sh TEST # Env: TOP_BUILDDIR, PG_CONFIG, MICROBENCH_PORT (default 55432), -# MICROBENCH_N (default 128), MICROBENCH_ROUNDS (default 1000) +# MICROBENCH_ROUNDS (default 1000), +# MICROBENCH_ITERATIONS or MICROBENCH_N (default 128) # set -euo pipefail @@ -16,8 +17,8 @@ TOP_BUILDDIR=${TOP_BUILDDIR:-$(cd "$MODULE_DIR/../../../.." && pwd)} PORT=${MICROBENCH_PORT:-55432} LOGDIR="$MODULE_DIR/.tmp_check/log" DATADIR="$MODULE_DIR/.tmp_check/data" -N=${MICROBENCH_N:-128} ROUNDS=${MICROBENCH_ROUNDS:-1000} +ITERATIONS=${MICROBENCH_ITERATIONS:-${MICROBENCH_N:-128}} log() { printf '%s\n' "$*" >&2; } @@ -107,14 +108,14 @@ trap cleanup EXIT if ! "$BINDIR/pg_ctl" -D "$DATADIR" status >/dev/null 2>&1; then log "==> starting postgres on port $PORT..." if ! "$BINDIR/pg_ctl" -D "$DATADIR" -l "$LOGDIR/postgres.log" \ - -o "-p $PORT -F -h '' -c shared_buffers=128MB" start \ + -o "-p $PORT -F -h '' -c shared_buffers=128MB -c max_worker_processes=256 -c max_parallel_workers=256" start \ >>"$LOGDIR/pg_ctl.log" 2>&1; then if grep -q 'incompatible with server' "$LOGDIR/postgres.log"; then log "==> postgres rejected datadir; re-initdb..." rm -rf "$DATADIR" ensure_datadir "$BINDIR/pg_ctl" -D "$DATADIR" -l "$LOGDIR/postgres.log" \ - -o "-p $PORT -F -h '' -c shared_buffers=128MB" start \ + -o "-p $PORT -F -h '' -c shared_buffers=128MB -c max_worker_processes=256 -c max_parallel_workers=256" start \ >>"$LOGDIR/pg_ctl.log" 2>&1 else log "pg_ctl start failed; see $LOGDIR/postgres.log and $LOGDIR/pg_ctl.log" @@ -129,7 +130,7 @@ log "==> CREATE EXTENSION microbench" -c "DROP EXTENSION IF EXISTS microbench CASCADE; CREATE EXTENSION microbench;" \ >>"$LOGDIR/psql.log" 2>&1 -log "==> running $TEST/query.sql (n=$N rounds=$ROUNDS)" +log "==> running $TEST/query.sql (rounds=$ROUNDS iterations=$ITERATIONS)" "$BINDIR/psql" -v ON_ERROR_STOP=1 -p "$PORT" -d postgres \ - -v n="$N" -v rounds="$ROUNDS" \ + -v rounds="$ROUNDS" -v iterations="$ITERATIONS" \ -f "$MODULE_DIR/$TEST/query.sql" diff --git a/src/test/modules/microbench/timing-magic.h b/src/test/modules/microbench/timing-magic.h index 54318841f64..15f3978d49c 100644 --- a/src/test/modules/microbench/timing-magic.h +++ b/src/test/modules/microbench/timing-magic.h @@ -1,30 +1,73 @@ #ifndef MICROBENCH_TIMING_MAGIC_H #define MICROBENCH_TIMING_MAGIC_H +#include + #include "portability/instr_time.h" +#include "multiprocessing.h" #define INIT_TIMING_SCOPE() \ int64 timing_operation_id = 0 -#define BEGIN_TIMING(name, n) \ +#define BEGIN_TIMING(name, iterations) \ do { \ instr_time t0, t1, dt; \ - Datum values[4]; \ - bool nulls[4] = {0}; \ - values[0] = CStringGetTextDatum(name); \ - INSTR_TIME_SET_CURRENT_FAST(t0); \ - for (int64 i = 0; i < (n); ++i) \ + const char *timing_name = (name); \ + double avg_ns; \ + synchronize_backends(rsinfo); \ + INSTR_TIME_SET_CURRENT(t0); \ + for (int64 i = 0; i < (iterations); ++i) \ { #define END_TIMING \ } \ - INSTR_TIME_SET_CURRENT_FAST(t1); \ + INSTR_TIME_SET_CURRENT(t1); \ INSTR_TIME_SET_ZERO(dt); \ INSTR_TIME_ACCUM_DIFF(dt, t1, t0); \ - values[1] = Float8GetDatum((double) INSTR_TIME_GET_NANOSEC(dt) / (double) (n)); \ - values[2] = Int64GetDatum(n); \ - values[3] = Int64GetDatum(++timing_operation_id); \ - tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls); \ + avg_ns = (double) INSTR_TIME_GET_NANOSEC(dt) / (double) (iterations); \ + microbench_mp_emit_sample(rsinfo, timing_name, avg_ns, (iterations), \ + ++timing_operation_id, 0, true); \ + } while (0) + +#define BEGIN_GROUPED_TIMING(name, iterations, n_groups) \ + do { \ + int64 _n_groups = (n_groups); \ + int64 _n = (iterations); \ + int64 group_count[n_groups]; \ + instr_time group_dt[n_groups]; \ + instr_time t0, t1; \ + const char *timing_name = (name); \ + int64 g; \ + int64 i; \ + memset(group_count, 0, sizeof(group_count)); \ + for (g = 0; g < _n_groups; g++) \ + INSTR_TIME_SET_ZERO(group_dt[g]); \ + synchronize_backends(rsinfo); \ + for (i = 0; i < _n; ++i) \ + { \ + int64 group_id = 0; \ + INSTR_TIME_SET_CURRENT(t0); + +#define END_GROUPED_TIMING \ + INSTR_TIME_SET_CURRENT(t1); \ + if (group_id < 0 || group_id >= _n_groups) \ + elog(ERROR, \ + "group_id " INT64_FORMAT " out of range [0, " INT64_FORMAT ")", \ + group_id, _n_groups); \ + INSTR_TIME_ACCUM_DIFF(group_dt[group_id], t1, t0); \ + group_count[group_id]++; \ + } \ + for (g = 0; g < _n_groups; g++) \ + { \ + int64 cnt; \ + double avg_ns; \ + cnt = group_count[g]; \ + if (cnt == 0) \ + continue; \ + avg_ns = (double) INSTR_TIME_GET_NANOSEC(group_dt[g]) / (double) cnt; \ + microbench_mp_emit_sample(rsinfo, timing_name, avg_ns, cnt, \ + ++timing_operation_id, g, false); \ + } \ } while (0) #endif -- 2.53.0