Benchmark reads while a writer is committing

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<String, String> 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<Result> 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<Future<?>> 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)) {

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()));

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());
+    }
+}

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: