org.apache.maven.plugins
diff --git a/src/test/java/com/javaqueue/bench/BenchQueue.java b/src/test/java/com/javaqueue/bench/BenchQueue.java
new file mode 100644
index 0000000..0fbd3cd
--- /dev/null
+++ b/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);
+ }
+ }
+ }
+}
diff --git a/src/test/java/com/javaqueue/bench/BenchmarkRunner.java b/src/test/java/com/javaqueue/bench/BenchmarkRunner.java
new file mode 100644
index 0000000..6d52b3b
--- /dev/null
+++ b/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.
+ *
+ *
+ * mvn -q test-compile exec:java \
+ * -Dexec.classpathScope=test \
+ * -Dexec.mainClass=com.javaqueue.bench.BenchmarkRunner
+ *
+ *
+ * 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();
+ }
+}
diff --git a/src/test/java/com/javaqueue/bench/ConcurrencyBenchmark.java b/src/test/java/com/javaqueue/bench/ConcurrencyBenchmark.java
new file mode 100644
index 0000000..5054440
--- /dev/null
+++ b/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:
+ *
+ *
+ * -t 1 -t 2 -t 4 -t 8
+ *
+ *
+ * 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);
+ }
+}
diff --git a/src/test/java/com/javaqueue/bench/RoundTripBenchmark.java b/src/test/java/com/javaqueue/bench/RoundTripBenchmark.java
new file mode 100644
index 0000000..14a5565
--- /dev/null
+++ b/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);
+ }
+}