Add JMH benchmark harness

benchmarks.sh

@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+#
+# Runs the JMH benchmark suite.
+#
+#   ./benchmarks.sh                          # everything, ~10 min
+#   ./benchmarks.sh RoundTripBenchmark       # one class
+#   ./benchmarks.sh RoundTrip -f 1 -wi 1 -i 2 -r 1s     # quick check
+#
+# Any JMH command-line flag can be appended. Results land in
+# target/jmh-result.json unless -rff says otherwise.
+#
+# JMH takes a single -t, so a scaling curve means one run per thread count:
+#
+#   for t in 1 2 4 8; do
+#     ./benchmarks.sh ConcurrencyBenchmark -t "$t" -rff "target/concurrency-t$t.json"
+#   done
+#
+# JMH forks a fresh JVM per trial for isolation, which is why this builds an
+# explicit classpath rather than using exec:java -- the forked process needs a
+# real java.class.path, and Maven's classloader does not provide one.
+
+set -euo pipefail
+
+cd "$(dirname "$0")"
+
+CLASSPATH_FILE=target/benchmark-classpath.txt
+
+echo "==> Compiling"
+mvn -q test-compile
+
+echo "==> Resolving classpath"
+mvn -q dependency:build-classpath \
+    -Dmdep.outputFile="$CLASSPATH_FILE" \
+    -Dmdep.includeScope=test
+
+echo "==> Running benchmarks"
+exec java \
+    -cp "target/classes:target/test-classes:$(cat "$CLASSPATH_FILE")" \
+    com.javaqueue.bench.BenchmarkRunner \
+    "$@"

pom.xml

@@ -14,6 +14,7 @@
     <maven.compiler.release>21</maven.compiler.release>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <exec.mainClass>com.javaqueue.Main</exec.mainClass>
+    <jmh.version>1.37</jmh.version>
   </properties>
 
   <dependencies>
@@ -25,6 +26,20 @@
       <scope>test</scope>
     </dependency>
 
+    <!-- JMH — benchmarks live in src/test/java/com/javaqueue/bench -->
+    <dependency>
+      <groupId>org.openjdk.jmh</groupId>
+      <artifactId>jmh-core</artifactId>
+      <version>${jmh.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.openjdk.jmh</groupId>
+      <artifactId>jmh-generator-annprocess</artifactId>
+      <version>${jmh.version}</version>
+      <scope>test</scope>
+    </dependency>
+
     <!-- Jetty 12 — HTTP server for Phase 4 networking -->
     <dependency>
       <groupId>org.eclipse.jetty</groupId>
@@ -60,6 +75,23 @@
         <artifactId>maven-surefire-plugin</artifactId>
         <version>3.2.5</version>
       </plugin>
+      <!-- JMH generates its benchmark harness from annotations at test-compile -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <version>3.13.0</version>
+        <configuration>
+          <!-- JDK 23+ defaults to -proc:none, which silently skips generation -->
+          <proc>full</proc>
+          <testAnnotationProcessorPaths>
+            <path>
+              <groupId>org.openjdk.jmh</groupId>
+              <artifactId>jmh-generator-annprocess</artifactId>
+              <version>${jmh.version}</version>
+            </path>
+          </testAnnotationProcessorPaths>
+        </configuration>
+      </plugin>
       <!-- Bundles Jetty into a single runnable jar: java -jar javaqueue.jar -->
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>

src/test/java/com/javaqueue/bench/BenchQueue.java

@@ -0,0 +1,76 @@
+package com.javaqueue.bench;
+
+import com.javaqueue.core.MessageQueue;
+import com.javaqueue.core.QueueConfig;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+
+/**
+ * A queue plus the temp directory backing it, so benchmarks can open and
+ * discard configurations without leaking files between iterations.
+ */
+final class BenchQueue implements AutoCloseable {
+
+    /**
+     * Well beyond any benchmark iteration. The visibility scanner must never
+     * requeue a message mid-run, or throughput would include redelivery work
+     * that a healthy consumer would never trigger.
+     */
+    private static final long VISIBILITY_TIMEOUT_MS = 600_000;
+
+    private static final int MAX_RETRIES = 3;
+
+    private final MessageQueue queue;
+    private final Path logDirectory;
+
+    private BenchQueue(MessageQueue queue, Path logDirectory) {
+        this.queue = queue;
+        this.logDirectory = logDirectory;
+    }
+
+    /**
+     * @param durability {@code memory} for heap-only, {@code wal} to write
+     *                   through the write-ahead log
+     */
+    static BenchQueue open(String name, String durability) throws IOException {
+        Path directory = switch (durability) {
+            case "memory" -> null;
+            case "wal" -> Files.createTempDirectory("javaqueue-bench-");
+            default -> throw new IllegalArgumentException("unknown durability: " + durability);
+        };
+
+        QueueConfig config = new QueueConfig(
+                VISIBILITY_TIMEOUT_MS,
+                MAX_RETRIES,
+                null,
+                directory == null ? null : directory.toString());
+
+        return new BenchQueue(new MessageQueue(name, config), directory);
+    }
+
+    MessageQueue queue() {
+        return queue;
+    }
+
+    @Override
+    public void close() throws IOException {
+        queue.close();
+        if (logDirectory != null) {
+            deleteRecursively(logDirectory);
+        }
+    }
+
+    private static void deleteRecursively(Path root) throws IOException {
+        if (!Files.exists(root)) {
+            return;
+        }
+        try (var paths = Files.walk(root)) {
+            for (Path path : paths.sorted(Comparator.reverseOrder()).toList()) {
+                Files.deleteIfExists(path);
+            }
+        }
+    }
+}

src/test/java/com/javaqueue/bench/BenchmarkRunner.java

@@ -0,0 +1,42 @@
+package com.javaqueue.bench;
+
+import org.openjdk.jmh.results.format.ResultFormatType;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.options.ChainedOptionsBuilder;
+import org.openjdk.jmh.runner.options.CommandLineOptions;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Entry point for the benchmark suite.
+ *
+ * <pre>
+ *   mvn -q test-compile exec:java \
+ *       -Dexec.classpathScope=test \
+ *       -Dexec.mainClass=com.javaqueue.bench.BenchmarkRunner
+ * </pre>
+ *
+ * Accepts the full JMH command line via {@code -Dexec.args}, so filters,
+ * thread counts and iteration counts work as documented upstream. With no
+ * arguments it runs every benchmark and writes JSON to
+ * {@code target/jmh-result.json}.
+ */
+public final class BenchmarkRunner {
+
+    private BenchmarkRunner() {
+    }
+
+    public static void main(String[] args) throws Exception {
+        CommandLineOptions commandLine = new CommandLineOptions(args);
+        ChainedOptionsBuilder options = new OptionsBuilder().parent(commandLine);
+
+        if (commandLine.getIncludes().isEmpty()) {
+            options.include("com\\.javaqueue\\.bench\\..*");
+        }
+
+        if (!commandLine.getResult().hasValue()) {
+            options.resultFormat(ResultFormatType.JSON).result("target/jmh-result.json");
+        }
+
+        new Runner(options.build()).run();
+    }
+}

src/test/java/com/javaqueue/bench/ConcurrencyBenchmark.java

@@ -0,0 +1,75 @@
+package com.javaqueue.bench;
+
+import com.javaqueue.core.Message;
+import com.javaqueue.core.MessageQueue;
+import com.javaqueue.core.Receipt;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * How aggregate throughput scales as producers and consumers contend for one
+ * queue. Every thread runs the same publish -> consume -> acknowledge round
+ * trip against a single shared {@link MessageQueue}.
+ *
+ * Thread count comes from the JMH command line rather than an annotation, so a
+ * scaling curve is one invocation:
+ *
+ * <pre>
+ *   -t 1 -t 2 -t 4 -t 8
+ * </pre>
+ *
+ * A thread may consume a message published by a different thread. That is the
+ * point -- it is the contended multi-producer, multi-consumer path. It also
+ * stays deadlock-free: a thread only consumes after its own publish, so at
+ * least one message is always outstanding when any thread calls consume.
+ */
+@State(Scope.Benchmark)
+@Fork(value = 1, jvmArgs = {"-Xms1g", "-Xmx1g"})
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+public class ConcurrencyBenchmark {
+
+    private static final long CONSUME_TIMEOUT_MS = 30_000;
+    private static final int PAYLOAD_SIZE = 1024;
+
+    @Param({"memory", "wal"})
+    public String durability;
+
+    private BenchQueue fixture;
+    private String payload;
+
+    @Setup(Level.Trial)
+    public void createPayload() {
+        payload = "x".repeat(PAYLOAD_SIZE);
+    }
+
+    @Setup(Level.Iteration)
+    public void openQueue() throws IOException {
+        fixture = BenchQueue.open("concurrency", durability);
+    }
+
+    @TearDown(Level.Iteration)
+    public void closeQueue() throws IOException {
+        fixture.close();
+    }
+
+    @Benchmark
+    public void contendedRoundTrip(Blackhole bh) throws InterruptedException {
+        MessageQueue queue = fixture.queue();
+        queue.publish(new Message(payload));
+
+        Receipt receipt = queue.consume(CONSUME_TIMEOUT_MS);
+        if (receipt == null) {
+            throw new IllegalStateException(
+                    "consume timed out after " + CONSUME_TIMEOUT_MS + "ms under contention");
+        }
+
+        queue.acknowledge(receipt.getReceiptHandle());
+        bh.consume(receipt);
+    }
+}

src/test/java/com/javaqueue/bench/RoundTripBenchmark.java

@@ -0,0 +1,94 @@
+package com.javaqueue.bench;
+
+import com.javaqueue.core.Message;
+import com.javaqueue.core.MessageQueue;
+import com.javaqueue.core.QueueConfig;
+import com.javaqueue.core.Receipt;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * End-to-end publish -> consume -> acknowledge, measured as throughput and as a
+ * latency distribution.
+ *
+ * The round trip is deliberately the unit of work rather than a bare publish:
+ * publishing without consuming grows the queue without bound, so a long run
+ * would measure allocation pressure rather than queue behaviour. Acknowledging
+ * every message keeps depth flat and the numbers steady-state.
+ *
+ * The {@code durability} axis is the interesting one. {@code memory} keeps
+ * everything in the heap. {@code wal} writes each operation through the
+ * write-ahead log, which currently means a BufferedWriter flush into the OS
+ * page cache -- not an fsync -- so it costs a syscall per publish but not a
+ * disk seek.
+ */
+@State(Scope.Benchmark)
+@Fork(value = 1, jvmArgs = {"-Xms1g", "-Xmx1g"})
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+public class RoundTripBenchmark {
+
+    /** Long enough that a healthy queue never hits it; short enough to fail fast if it stalls. */
+    private static final long CONSUME_TIMEOUT_MS = 10_000;
+
+    @Param({"64", "1024", "8192"})
+    public int payloadSize;
+
+    @Param({"memory", "wal"})
+    public String durability;
+
+    private BenchQueue fixture;
+    private String payload;
+
+    @Setup(Level.Trial)
+    public void createPayload() {
+        payload = "x".repeat(payloadSize);
+    }
+
+    /**
+     * Rebuilt every iteration so a WAL run starts from an empty log. Without
+     * this the log file grows across the whole trial and later iterations pay
+     * for earlier ones.
+     */
+    @Setup(Level.Iteration)
+    public void openQueue() throws IOException {
+        fixture = BenchQueue.open("roundtrip", durability);
+    }
+
+    @TearDown(Level.Iteration)
+    public void closeQueue() throws IOException {
+        fixture.close();
+    }
+
+    @Benchmark
+    @BenchmarkMode(Mode.Throughput)
+    @OutputTimeUnit(TimeUnit.SECONDS)
+    public void throughput(Blackhole bh) throws InterruptedException {
+        roundTrip(bh);
+    }
+
+    @Benchmark
+    @BenchmarkMode(Mode.SampleTime)
+    @OutputTimeUnit(TimeUnit.MICROSECONDS)
+    public void latency(Blackhole bh) throws InterruptedException {
+        roundTrip(bh);
+    }
+
+    private void roundTrip(Blackhole bh) throws InterruptedException {
+        MessageQueue queue = fixture.queue();
+        queue.publish(new Message(payload));
+
+        Receipt receipt = queue.consume(CONSUME_TIMEOUT_MS);
+        if (receipt == null) {
+            throw new IllegalStateException(
+                    "consume timed out after " + CONSUME_TIMEOUT_MS + "ms -- "
+                            + "the queue should always hold this thread's own message");
+        }
+
+        queue.acknowledge(receipt.getReceiptHandle());
+        bh.consume(receipt);
+    }
+}