diff --git a/benchmarks/src/main/java/io/hstore/bench/Comparison.java b/benchmarks/src/main/java/io/hstore/bench/Comparison.java index fb5656c..4622f95 100644 --- a/benchmarks/src/main/java/io/hstore/bench/Comparison.java +++ b/benchmarks/src/main/java/io/hstore/bench/Comparison.java @@ -8,6 +8,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; @@ -18,6 +19,7 @@ import java.util.function.IntToLongFunction; import java.util.function.LongSupplier; import java.util.function.Supplier; +import java.util.concurrent.locks.LockSupport; public final class Comparison { @@ -61,6 +63,10 @@ private interface Slice { private static final int BATCH = 1_000; private static final int COMMITS = 5_000; + private static final long MIXED_NANOS = 10_000_000_000L; + private static final int MIXED_BATCH = 10; + private static final int MIXED_UPDATES_PER_SECOND = 1_000; + private static final int SAMPLES_PER_THREAD = 200_000; private static final long SEED = 20_260_904L; private Comparison() { @@ -140,6 +146,7 @@ private static void run(Map options) throws Exception { store.update(dataset.updates(), i, i + 1, round[0]++); return 1; }, false))); + results.addAll(mixed(store, dataset, threads)); int largeSize = dataset.nodes(); results.add(writing(store, () -> measure("large.ingest", "create one hyperedge with %,d members".formatted(largeSize), largeSize, () -> { store.largeEdge(largeSize); @@ -225,6 +232,65 @@ private static Result latency(String workload, String description, int operation .using(JvmUsage.now().since(usage)).with("latency", Latency.of(nanos).json()); } + private static List mixed(Store store, Dataset dataset, int threads) throws Exception { + int[] probes = dataset.probeNodes(); + int[] updates = dataset.updates(); + int readers = Math.max(1, threads - 1); + Reservoir[] reads = new Reservoir[readers]; + Reservoir commits = new Reservoir(SAMPLES_PER_THREAD, SEED); + long interval = 1_000_000_000L * MIXED_BATCH / MIXED_UPDATES_PER_SECOND; + long before = store.bytesWritten(); + JvmUsage usage = JvmUsage.now(); + long started; + long finished; + long committed = 0; + try (ExecutorService pool = Executors.newFixedThreadPool(readers)) { + started = System.nanoTime(); + long deadline = started + MIXED_NANOS; + List> futures = new ArrayList<>(); + for (int t = 0; t < readers; t++) { + Reservoir reservoir = reads[t] = new Reservoir(SAMPLES_PER_THREAD, SEED + t + 1); + int offset = t * BATCH; + futures.add(pool.submit(() -> { + for (int i = offset; System.nanoTime() < deadline; i = (i + 1) % probes.length) { + long begin = System.nanoTime(); + store.incidence(probes, i, i + 1); + reservoir.add(System.nanoTime() - begin); + } + })); + } + for (int k = 0; ; k++) { + long intended = started + k * interval; + if (intended >= deadline || System.nanoTime() >= deadline) { + break; + } + for (long wait = intended - System.nanoTime(); wait > 0; wait = intended - System.nanoTime()) { + LockSupport.parkNanos(wait); + } + int from = k * MIXED_BATCH % updates.length; + store.update(updates, from, Math.min(updates.length, from + MIXED_BATCH), 1_000_000 + k); + commits.add(System.nanoTime() - intended); + committed += Math.min(updates.length, from + MIXED_BATCH) - from; + } + for (Future future : futures) { + future.get(); + } + finished = System.nanoTime(); + } + JvmUsage used = JvmUsage.now().since(usage); + store.flush(); + double millis = (finished - started) / 1e6; + long readCount = Arrays.stream(reads).mapToLong(Reservoir::seen).sum(); + String load = "%,d readers while one writer commits %d updates at a time, targeting %,d updates/s" + .formatted(readers, MIXED_BATCH, MIXED_UPDATES_PER_SECOND); + Result read = new Result("mixed.read", "single incidence lookups from " + load, readCount, millis, 0) + .using(used).with("latency", Reservoir.merge(reads).json()).with("checked", new Json.Bool(false)); + Result write = new Result("mixed.write", "updates committed by the writer alongside the readers", committed, millis, 0) + .with("latency", Reservoir.merge(commits).json()).with("checked", new Json.Bool(false)) + .with("bytesWritten", number(store.bytesWritten() - before)); + return List.of(read, write); + } + private static Result concurrent(Store store, Dataset dataset, int threads) throws Exception { int[] probes = dataset.probeNodes(); try (ExecutorService pool = Executors.newFixedThreadPool(threads)) { diff --git a/benchmarks/src/main/java/io/hstore/bench/Report.java b/benchmarks/src/main/java/io/hstore/bench/Report.java index c0c5a59..55dc299 100644 --- a/benchmarks/src/main/java/io/hstore/bench/Report.java +++ b/benchmarks/src/main/java/io/hstore/bench/Report.java @@ -57,11 +57,12 @@ static String markdown(String[] args) { double[] mine = values(subjects, workload, disk, single); double[] theirs = values(baselines, workload, disk, single); double ratio = disk || single ? median(theirs) / median(mine) : median(mine) / median(theirs); + boolean checked = !(entry.getValue().get("checked") instanceof Json.Bool(boolean value)) || value; boolean agree = disk || workload.startsWith("ingest") || workload.equals("reopen") || Stream.concat(subjects.stream(), baselines.stream()) .map(run -> number(run.results().get(workload).get("checksum"))).distinct().count() == 1; out.append("| `%s`: %s | %s | %s | **%.2f×** | %s |%n".formatted(workload, text(entry.getValue().get("description")), - cell(mine, disk, single), cell(theirs, disk, single), ratio, agree ? "yes" : "**no**")); + cell(mine, disk, single), cell(theirs, disk, single), ratio, !checked ? "n/a" : agree ? "yes" : "**no**")); } out.append("%nRatios above 1 favour %s: throughput ratios divide %s by %s; latency and size ratios divide %s by %s.%n" .formatted(subject.store(), subject.store(), baseline.store(), baseline.store(), subject.store())); diff --git a/benchmarks/src/main/java/io/hstore/bench/Reservoir.java b/benchmarks/src/main/java/io/hstore/bench/Reservoir.java new file mode 100644 index 0000000..1566071 --- /dev/null +++ b/benchmarks/src/main/java/io/hstore/bench/Reservoir.java @@ -0,0 +1,43 @@ +package io.hstore.bench; + +import java.util.Arrays; +import java.util.SplittableRandom; +import java.util.stream.LongStream; + +final class Reservoir { + + private final long[] sample; + private final SplittableRandom random; + private long seen; + private long max; + + Reservoir(int capacity, long seed) { + this.sample = new long[capacity]; + this.random = new SplittableRandom(seed); + } + + void add(long nanos) { + max = Math.max(max, nanos); + if (seen < sample.length) { + sample[(int) seen] = nanos; + } else { + long slot = random.nextLong(seen + 1); + if (slot < sample.length) { + sample[(int) slot] = nanos; + } + } + seen++; + } + + long seen() { + return seen; + } + + static Latency merge(Reservoir... reservoirs) { + return Latency.of(Arrays.stream(reservoirs) + .flatMapToLong(reservoir -> LongStream.concat( + Arrays.stream(reservoir.sample, 0, (int) Math.min(reservoir.seen, reservoir.sample.length)), + LongStream.of(reservoir.max))) + .toArray()); + } +} diff --git a/docs/benchmarks.md b/docs/benchmarks.md index e109a7b..f9ea77a 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -49,6 +49,8 @@ taken from a common hyperedge (so co-membership counts are non-zero), and 20,000 | `latency.read` | one incidence lookup per read transaction, each timed on its own | `Reader.incident` | `getIncidenceSet(h).getSearchResult()` | | `write.update` | replace a node's value, 1,000 per transaction | `Writer.set` | `graph.replace` | | `latency.commit` | 5,000 write transactions of one update each, each timed on its own | `Writer.set` | `graph.replace` | +| `mixed.read` | single incidence lookups from `threads − 1` threads for 10 s while `mixed.write` runs | `Reader.incident` | `getIncidenceSet(h).getSearchResult()` | +| `mixed.write` | one thread committing 10 updates at a time on a fixed schedule of 1,000 updates/s | `Writer.set` | `graph.replace` | | `large.ingest` | one hyperedge containing every node | `Writer.load` | one `HGPlainLink` | | `large.scan` | enumerate the large hyperedge twenty times | `Reader.members` | `getTargetAt` loop | | `large.probe` | test whether a node belongs to the large hyperedge | `View.incidence(edge, node)` (membership-tree lookup) | `getIncidenceSet(node).contains(edge)` | @@ -64,6 +66,13 @@ maximum, taking the median of each percentile across runs. Throughput hides occa group commit, checkpoints, compaction or GC; these percentiles show them. Under `sync` durability, `latency.commit` is effectively the cost of one durable commit. +**Reads during writes.** In `mixed`, the readers run in a closed loop for 10 s, and a single writer commits on a +fixed schedule (open loop). Each commit's latency is measured from when it was *scheduled* to start, not from +when it actually started. If the engine stalls, the commits queued behind the stall show the delay, instead of +the writer quietly slowing down. The writer stops at the deadline even when it is behind, so an engine that can't +keep up shows an achieved rate below 1,000 updates/s. Readers keep a uniform random sample of 200,000 latencies +each, plus the exact maximum. Both results depend on timing, so *Results agree* shows "n/a". + **Bytes written.** For every write workload the report also shows the bytes each engine wrote per operation, in a *Resources* table, plus the total for the whole run. Both engines count the bytes they write themselves: