diff --git a/benchmarks/run.sh b/benchmarks/run.sh index b5e0a396..0bc2b67a 100755 --- a/benchmarks/run.sh +++ b/benchmarks/run.sh @@ -9,10 +9,18 @@ heap="${HEAP:-4g}" history="${HISTORY:-64}" runs="${RUNS:-3}" cache="${CACHE_MB:-0}" +segment="${PAGES_PER_SEGMENT:-0}" +ratio="${COMPACTION_LIVE_RATIO:-0.5}" results="results/scale-$scale-$durability-history-$history" if [ "$cache" != 0 ]; then results="$results-cache-$cache" fi +if [ "$segment" != 0 ]; then + results="$results-segment-$segment" +fi +if [ "$ratio" != 0.5 ]; then + results="$results-ratio-$ratio" +fi ../mvnw -q -f ../pom.xml -Pbenchmarks -DskipTests install classpath="target/classes:$(cat target/classpath.txt)" @@ -26,7 +34,7 @@ rm -rf "$results" && mkdir -p "$results" for run in $(seq 1 "$runs"); do for store in hstore hypergraphdb; do java "${jvm[@]}" io.hstore.bench.Comparison run --store "$store" --scale "$scale" --durability "$durability" \ - --threads "$threads" --history "$history" --cache-mb "$cache" --out "$results/$store-$run.json" + --threads "$threads" --history "$history" --cache-mb "$cache" --pages-per-segment "$segment" --compaction-live-ratio "$ratio" --out "$results/$store-$run.json" done done java "${jvm[@]}" io.hstore.bench.Comparison report "$results"/hstore-*.json "$results"/hypergraphdb-*.json | tee "$results/report.md" diff --git a/benchmarks/src/main/java/io/hstore/bench/Comparison.java b/benchmarks/src/main/java/io/hstore/bench/Comparison.java index 07593ada..0fe77763 100644 --- a/benchmarks/src/main/java/io/hstore/bench/Comparison.java +++ b/benchmarks/src/main/java/io/hstore/bench/Comparison.java @@ -1,6 +1,7 @@ package io.hstore.bench; import io.hstore.db.value.Json; +import io.hstore.engine.EngineOptions; import java.io.BufferedReader; import java.io.IOException; @@ -21,6 +22,7 @@ import java.util.function.IntToLongFunction; import java.util.function.LongSupplier; import java.util.function.Supplier; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; public final class Comparison { @@ -70,6 +72,9 @@ private interface Slice { 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 COMPACTION_WINDOW_NANOS = 10_000_000_000L; + private static final long COMPACTION_DELAY_NANOS = 1_000_000_000L; + private static final long COMPACTION_INTERVAL_NANOS = 20_000_000L; private static final long SEED = 20_260_904L; private Comparison() { @@ -83,7 +88,7 @@ public static void main(String[] args) throws Exception { case "report" -> IO.println(Report.markdown(args)); default -> { IO.println(""" - usage: Comparison run --store hstore|hypergraphdb [--scale N] [--durability async|sync] [--threads N] [--history N] [--cache-mb N] --out FILE + usage: Comparison run --store hstore|hypergraphdb [--scale N] [--durability async|sync] [--threads N] [--history N] [--cache-mb N] [--compaction-live-ratio R] [--pages-per-segment N] --out FILE Comparison crash --store hstore|hypergraphdb --dir DIRECTORY (used by run; ingests until killed) Comparison report FILE... (runs of two stores; cells show the median and range)"""); System.exit(2); @@ -189,13 +194,18 @@ private static void run(Map options) throws Exception { })))); results.add(read("read.incidence.churned", "enumerate incidence sets after the deletes", nodes.length, (from, to) -> store.incidence(nodes, from, to), true)); + long[] churned = new long[1]; + store.reopen(() -> churned[0] = store.diskBytes()); + results.add(new Result("disk.churned", "bytes on disk after the deletes and a clean shutdown", churned[0], 1000.0, churned[0])); + results.add(compaction(store, dataset, false)); + results.add(compaction(store, dataset, true)); store.flush(); totals.put("bytesWritten", number(store.bytesWritten())); } finally { store.close(); } - long churned = store.diskBytes(); - results.add(new Result("disk.churned", "bytes on disk after the deletes and a clean shutdown", churned, 1000.0, churned)); + long compacted = store.diskBytes(); + results.add(new Result("disk.compacted", "bytes on disk after a full compaction and a clean shutdown", compacted, 1000.0, compacted)); results.add(recover(options, dataset)); write(out, store, scale, sync, threads, dataset, totals, results); results.forEach(result -> IO.println(" %-22s %,14.0f ops/s %,10.1f ms checksum %d" @@ -206,7 +216,9 @@ private static Store open(Map options, Path directory, boolean c boolean sync = options.getOrDefault("durability", "async").equals("sync"); long cacheBytes = Long.parseLong(options.getOrDefault("cache-mb", "0")) << 20; return switch (options.getOrDefault("store", "hstore")) { - case "hstore" -> new HStoreStore(directory, sync, Integer.parseInt(options.getOrDefault("history", "64")), cacheBytes, create); + case "hstore" -> new HStoreStore(directory, sync, Integer.parseInt(options.getOrDefault("history", "64")), cacheBytes, + Double.parseDouble(options.getOrDefault("compaction-live-ratio", String.valueOf(EngineOptions.defaults().compactionLiveRatio()))), + Integer.parseInt(options.getOrDefault("pages-per-segment", "0")), create); case "hypergraphdb" -> new HyperGraphDbStore(directory, sync, cacheBytes); case String other -> throw new IllegalArgumentException("unknown store " + other); }; @@ -228,7 +240,7 @@ private static Result recover(Map options, Dataset dataset) thro command.add(ProcessHandle.current().info().command().orElse("java")); command.addAll(ManagementFactory.getRuntimeMXBean().getInputArguments()); command.addAll(List.of("-cp", System.getProperty("java.class.path"), Comparison.class.getName(), "crash", "--dir", directory.toString())); - for (String option : List.of("store", "scale", "durability", "history", "cache-mb")) { + for (String option : List.of("store", "scale", "durability", "history", "cache-mb", "compaction-live-ratio", "pages-per-segment")) { if (options.containsKey(option)) { command.addAll(List.of("--" + option, options.get(option))); } @@ -375,6 +387,68 @@ private static List mixed(Store store, Dataset dataset, int threads) thr return List.of(read, write); } + private static Result compaction(Store store, Dataset dataset, boolean compact) throws Exception { + int[] updates = dataset.updates(); + Reservoir commits = new Reservoir(SAMPLES_PER_THREAD, SEED); + AtomicReference failure = new AtomicReference<>(); + long[] compactNanos = new long[1]; + long[] diskBefore = new long[1]; + if (compact) { + store.reopen(() -> diskBefore[0] = store.diskBytes()); + } + long before = store.bytesWritten(); + JvmUsage usage = JvmUsage.now(); + long started = System.nanoTime(); + long deadline = started + COMPACTION_WINDOW_NANOS; + Thread compactor = Thread.ofPlatform().name("bench-compaction").start(() -> { + if (!compact) { + return; + } + for (long wait = started + COMPACTION_DELAY_NANOS - System.nanoTime(); wait > 0; wait = started + COMPACTION_DELAY_NANOS - System.nanoTime()) { + LockSupport.parkNanos(wait); + } + long begin = System.nanoTime(); + try { + store.compact(); + } catch (Throwable e) { + failure.set(e); + } + compactNanos[0] = System.nanoTime() - begin; + }); + int committed = 0; + for (int k = 0; ; k++) { + long intended = started + k * COMPACTION_INTERVAL_NANOS; + if (intended >= deadline && !compactor.isAlive()) { + break; + } + for (long wait = intended - System.nanoTime(); wait > 0; wait = intended - System.nanoTime()) { + LockSupport.parkNanos(wait); + } + int index = k % updates.length; + store.update(updates, index, index + 1, (compact ? 4_000_000 : 3_000_000) + k); + commits.add(System.nanoTime() - intended); + committed++; + } + compactor.join(); + if (failure.get() != null) { + throw new IllegalStateException("compaction failed", failure.get()); + } + double millis = (System.nanoTime() - started) / 1e6; + JvmUsage used = JvmUsage.now().since(usage); + store.flush(); + String schedule = "single-update commits at %d/s".formatted(1_000_000_000L / COMPACTION_INTERVAL_NANOS); + Result result = new Result(compact ? "compaction" : "compaction.baseline", + compact ? schedule + " while a full compaction runs" : schedule + " with no compaction, for comparison", + committed, millis, 0).using(used).with("latency", Reservoir.merge(commits).json()).with("checked", new Json.Bool(false)) + .with("bytesWritten", number(store.bytesWritten() - before)); + if (!compact) { + return result; + } + long[] diskAfter = new long[1]; + store.reopen(() -> diskAfter[0] = store.diskBytes()); + return result.with("compactionMillis", number(compactNanos[0] / 1_000_000)).with("reclaimed", number(diskBefore[0] - diskAfter[0])); + } + 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/HStoreStore.java b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java index 66c70f51..7d8fd0df 100644 --- a/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java +++ b/benchmarks/src/main/java/io/hstore/bench/HStoreStore.java @@ -38,14 +38,15 @@ final class HStoreStore implements Store { private final int cachedNodes; private final long cacheBytes; - HStoreStore(Path directory, boolean sync, int history, long cacheBytes, boolean create) { + HStoreStore(Path directory, boolean sync, int history, long cacheBytes, double liveRatio, int pagesPerSegment, boolean create) { this.directory = directory; this.history = history; this.cacheBytes = cacheBytes; EngineOptions engine = EngineOptions.defaults(); this.cachedNodes = cacheBytes > 0 ? Math.toIntExact(cacheBytes / engine.pageSize()) : CACHED_NODES; this.options = DatabaseOptions.defaults().withEngine(engine - .withDurability(sync ? Durability.SYNC : Durability.ASYNC).withCachedNodes(cachedNodes).withHistoryLimit(history)); + .withDurability(sync ? Durability.SYNC : Durability.ASYNC).withCachedNodes(cachedNodes).withHistoryLimit(history) + .withCompactionLiveRatio(liveRatio).withPagesPerSegment(pagesPerSegment > 0 ? pagesPerSegment : engine.pagesPerSegment())); this.database = HypergraphDatabase.open(directory, options); if (create) { database.write(writer -> { @@ -215,6 +216,11 @@ public void flush() { database.engine().checkpoint(); } + @Override + public void compact() { + database.engine().compact(); + } + @Override public long bytesWritten() { EngineStats stats = database.engine().stats(); diff --git a/benchmarks/src/main/java/io/hstore/bench/HyperGraphDbStore.java b/benchmarks/src/main/java/io/hstore/bench/HyperGraphDbStore.java index 5d48958d..b6efe144 100644 --- a/benchmarks/src/main/java/io/hstore/bench/HyperGraphDbStore.java +++ b/benchmarks/src/main/java/io/hstore/bench/HyperGraphDbStore.java @@ -1,5 +1,6 @@ package io.hstore.bench; +import com.sleepycat.je.CheckpointConfig; import com.sleepycat.je.Durability; import com.sleepycat.je.Environment; import com.sleepycat.je.EnvironmentConfig; @@ -242,6 +243,15 @@ public void flush() { environment().sync(); } + @Override + public void compact() { + Environment environment = environment(); + while (environment.cleanLog() > 0) { + Thread.onSpinWait(); + } + environment.checkpoint(new CheckpointConfig().setForce(true)); + } + @Override public long bytesWritten() { EnvironmentStats stats = environment().getStats(new StatsConfig()); diff --git a/benchmarks/src/main/java/io/hstore/bench/Report.java b/benchmarks/src/main/java/io/hstore/bench/Report.java index a81eaab7..9e0c2e2f 100644 --- a/benchmarks/src/main/java/io/hstore/bench/Report.java +++ b/benchmarks/src/main/java/io/hstore/bench/Report.java @@ -49,7 +49,7 @@ static String markdown(String[] args) { out.append("|---|---:|---:|---:|:---:|%n".formatted()); for (Map.Entry> entry : subject.results().entrySet()) { String workload = entry.getKey(); - if (baselines.stream().anyMatch(run -> !run.results().containsKey(workload))) { + if (workload.startsWith("compaction") || baselines.stream().anyMatch(run -> !run.results().containsKey(workload))) { continue; } boolean disk = workload.startsWith("disk"); @@ -71,13 +71,15 @@ static String markdown(String[] args) { return out.toString(); } - private record Metric(String key, String name, boolean perOperation, boolean showZero, DoubleFunction format) { + private record Metric(String key, String name, boolean perOperation, boolean showZero, boolean higherIsBetter, DoubleFunction format) { } - private static final List METRICS = List.of(new Metric("bytesWritten", "bytes written per operation", true, false, Report::bytes), - new Metric("allocated", "heap allocated per operation", true, false, Report::bytes), - new Metric("gcMillis", "GC pause time", false, false, value -> "%,.0f ms".formatted(value)), - new Metric("lost", "acknowledged nodes lost", false, true, value -> "%,.0f".formatted(value))); + private static final List METRICS = List.of(new Metric("bytesWritten", "bytes written per operation", true, false, false, Report::bytes), + new Metric("allocated", "heap allocated per operation", true, false, false, Report::bytes), + new Metric("gcMillis", "GC pause time", false, false, false, value -> "%,.0f ms".formatted(value)), + new Metric("lost", "acknowledged nodes lost", false, true, false, value -> "%,.0f".formatted(value)), + new Metric("compactionMillis", "compaction time", false, true, false, value -> "%,.0f ms".formatted(value)), + new Metric("reclaimed", "disk space reclaimed", false, true, true, Report::bytes)); private static void resources(StringBuilder out, List subjects, List baselines) { List all = Stream.concat(subjects.stream(), baselines.stream()).toList(); @@ -90,7 +92,7 @@ private static void resources(StringBuilder out, List subjects, List b if (mine == 0 && theirs == 0 && !metric.showZero()) { continue; } - double ratio = theirs / mine; + double ratio = metric.higherIsBetter() ? (mine > 0 && theirs > 0 ? mine / theirs : Double.NaN) : theirs / mine; rows.append("| `%s` | %s | %s | %s | %s |%n".formatted(workload, metric.name(), metric.format().apply(mine), metric.format().apply(theirs), Double.isFinite(ratio) ? "**%.2f×**".formatted(ratio) : "n/a")); } @@ -127,10 +129,12 @@ private static double[] header(List runs, String key) { } private static String bytes(double value) { - if (value < 1024) { - return "%.0f B".formatted(value); + double size = Math.abs(value); + String sign = value < 0 ? "-" : ""; + if (size < 1024) { + return sign + "%.0f B".formatted(size); } - return value < 1 << 20 ? "%.1f KiB".formatted(value / 1024) : "%,.1f MiB".formatted(value / (1 << 20)); + return sign + (size < 1 << 20 ? "%.1f KiB".formatted(size / 1024) : "%,.1f MiB".formatted(size / (1 << 20))); } private static void latency(StringBuilder out, List subjects, List baselines) { diff --git a/benchmarks/src/main/java/io/hstore/bench/Store.java b/benchmarks/src/main/java/io/hstore/bench/Store.java index 8287e8db..187a3663 100644 --- a/benchmarks/src/main/java/io/hstore/bench/Store.java +++ b/benchmarks/src/main/java/io/hstore/bench/Store.java @@ -55,6 +55,8 @@ default void reopen() { void flush(); + void compact(); + long bytesWritten(); Path directory(); diff --git a/docs/benchmarks.md b/docs/benchmarks.md index 87f3e80b..c1ef8f5b 100644 --- a/docs/benchmarks.md +++ b/docs/benchmarks.md @@ -62,6 +62,9 @@ taken from a common hyperedge (so co-membership counts are non-zero), and 20,000 | `churn.remove` | remove one member from each of 20,000 other hyperedges, 1,000 per transaction | `Writer.remove` | `graph.replace` with a new `HGPlainLink` (links are immutable) | | `read.incidence.churned` | `read.incidence` after the deletes | | | | `disk.churned` | bytes on disk after the deletes and a clean shutdown | | | +| `compaction.baseline` | single-update commits at 50/s for 10 s, no compaction | `Writer.set` | `graph.replace` | +| `compaction` | the same commits while a full compaction runs, starting 1 s in | `StorageEngine.compact` | `Environment.cleanLog` until it finds nothing, then a forced checkpoint | +| `disk.compacted` | bytes on disk after the compaction and a clean shutdown | | | | `recover` | open the store after the process ingesting into it was killed with SIGKILL | `HypergraphDatabase.open` | `HGEnvironment.get` | Read workloads run a warm-up pass over 10% of the probes before timing, except `read.incidence.cold`. @@ -93,6 +96,18 @@ the loaded store after a clean shutdown. The checksum of `read.incidence.churned both engines applied the same deletes. `disk.churned` shows whether the space came back: through compaction for HStore, and the log cleaner for JE. +**Compaction.** After the churn, `disk.churned` is recorded at a reopen, and then two 10 s windows run. In both, one +writer commits a single update every 20 ms on a fixed schedule, and each commit's latency is measured from when it +was scheduled. In `compaction`, the store also runs a full compaction starting 1 s in, and the window lasts until +the compaction has finished. HStore calls `StorageEngine.compact()`. JE calls `cleanLog()` until it cleans nothing, +then forces a checkpoint so the cleaned files are deleted. The latency table compares the two windows, and the +*Resources* table adds the compaction time and the space reclaimed. Disk is measured at a reopen just before the +window and again at a reopen just after it, so segments that HStore only deletes once older readers are gone are +counted too. That figure can be negative when compaction frees less than the writer adds. The two windows run at a +fixed rate, so they are left out of the main table, where their ratio would always be about 1. HStore's default segments are 256 MiB, so +at scale 1 most data shares one or two segments. `--pages-per-segment` (`PAGES_PER_SEGMENT`) and +`--compaction-live-ratio` (`COMPACTION_LIVE_RATIO`) change HStore's settings for a run. + **Crash recovery.** `reopen` closes the store cleanly first. `recover` doesn't: 1. The run starts a second JVM (`Comparison crash`) with the same flags and classpath. It ingests nodes into a fresh directory and prints the count after every acknowledged commit.