Fix JMH deadlock: use tryPost with backpressure, add missing zRuntime.start(), fix HandlerRegistration leaks

Author: namanONcode

zthread-benchmark/src/main/java/io/github/namanoncode/zthread/benchmark/inotify/FileWatchBenchmark.java

@@ -102,7 +102,7 @@ public void benchJavaNioWatchService(Blackhole bh) throws InterruptedException,
     @OperationsPerInvocation(BATCH_SIZE)
     public void benchZThreadInotify(Blackhole bh) throws InterruptedException, IOException {
         latch = new CountDownLatch(BATCH_SIZE);
-        zRuntime.on(FileEvent.class, evt -> {
+        io.github.namanoncode.zthread.handler.HandlerRegistration reg = zRuntime.on(FileEvent.class, evt -> {
             bh.consume(evt);
             latch.countDown();
         });
@@ -112,6 +112,7 @@ public void benchZThreadInotify(Blackhole bh) throws InterruptedException, IOExc
         }
 
         latch.await();
+        reg.cancel();
 
         // Cleanup for next invocation
         for (int i = 0; i < BATCH_SIZE; i++) {

zthread-benchmark/src/main/java/io/github/namanoncode/zthread/benchmark/throughput/MpmcEventBenchmark.java

@@ -61,6 +61,7 @@ public void setup() {
         zRuntimes = new ZRuntime[CONSUMERS];
         for (int i = 0; i < CONSUMERS; i++) {
             zRuntimes[i] = ZRuntime.builder().build();
+            zRuntimes[i].start();
         }
 
         nettyEventLoopGroup = new DefaultEventLoopGroup(CONSUMERS);
@@ -206,23 +207,29 @@ public void benchConcurrentLinkedQueue(Blackhole bh) throws InterruptedException
     @OperationsPerInvocation(BATCH_SIZE)
     public void benchZThread(Blackhole bh) throws InterruptedException {
         CountDownLatch latch = new CountDownLatch(BATCH_SIZE);
-        EventHandler<io.github.namanoncode.zthread.event.CustomEvent> handler = evt -> {
+        io.github.namanoncode.zthread.handler.EventHandler<io.github.namanoncode.zthread.event.CustomEvent> handler = evt -> {
             bh.consume(evt);
             latch.countDown();
         };
+        java.util.List<io.github.namanoncode.zthread.handler.HandlerRegistration> regs = new java.util.ArrayList<>();
         for (ZRuntime zr : zRuntimes) {
-            zr.on(io.github.namanoncode.zthread.event.CustomEvent.class, handler);
+            regs.add(zr.on(io.github.namanoncode.zthread.event.CustomEvent.class, handler));
         }
 
         AtomicInteger index = new AtomicInteger();
         runProducers(() -> {
             int pIndex = index.getAndIncrement();
             ZRuntime target = zRuntimes[pIndex % CONSUMERS];
             for (int i = 0; i < EVENTS_PER_PRODUCER; i++) {
-                target.post(EVENT);
+                while (!target.tryPost(EVENT)) {
+                    java.util.concurrent.locks.LockSupport.parkNanos(10);
+                }
             }
         });
         latch.await();
+        for (io.github.namanoncode.zthread.handler.HandlerRegistration reg : regs) {
+            reg.cancel();
+        }
     }
 
     @Benchmark
@@ -270,9 +277,13 @@ public void benchReactor(Blackhole bh) throws InterruptedException {
 
         runProducers(() -> {
             for (int i = 0; i < EVENTS_PER_PRODUCER; i++) {
-                while (reactorSink.tryEmitNext(EVENT) == reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED) {
-                    java.util.concurrent.locks.LockSupport.parkNanos(10);
-                }
+                reactor.core.publisher.Sinks.EmitResult res;
+                do {
+                    res = reactorSink.tryEmitNext(EVENT);
+                    if (res != reactor.core.publisher.Sinks.EmitResult.OK) {
+                        java.util.concurrent.locks.LockSupport.parkNanos(10);
+                    }
+                } while (res != reactor.core.publisher.Sinks.EmitResult.OK);
             }
         });
         latch.await();

zthread-benchmark/src/main/java/io/github/namanoncode/zthread/benchmark/throughput/MpscEventBenchmark.java

@@ -59,6 +59,7 @@ public void setup() {
         synchronousQueue = new SynchronousQueue<>();
 
         zRuntime = ZRuntime.builder().build();
+        zRuntime.start();
         nettyEventLoop = new DefaultEventLoop();
         vertx = io.vertx.core.Vertx.vertx();
         virtualThreadExecutor = Executors.newVirtualThreadPerTaskExecutor();
@@ -212,17 +213,20 @@ public void benchConcurrentLinkedQueue(Blackhole bh) throws InterruptedException
     @OperationsPerInvocation(BATCH_SIZE)
     public void benchZThread(Blackhole bh) throws InterruptedException {
         CountDownLatch latch = new CountDownLatch(BATCH_SIZE);
-        zRuntime.on(io.github.namanoncode.zthread.event.CustomEvent.class, evt -> {
+        io.github.namanoncode.zthread.handler.HandlerRegistration reg = zRuntime.on(io.github.namanoncode.zthread.event.CustomEvent.class, evt -> {
             bh.consume(evt);
             latch.countDown();
         });
 
         runProducers(() -> {
             for (int i = 0; i < EVENTS_PER_PRODUCER; i++) {
-                zRuntime.post(EVENT);
+                while (!zRuntime.tryPost(EVENT)) {
+                    java.util.concurrent.locks.LockSupport.parkNanos(10);
+                }
             }
         });
         latch.await();
+        reg.cancel();
     }
 
     @Benchmark
@@ -270,9 +274,13 @@ public void benchReactor(Blackhole bh) throws InterruptedException {
 
         runProducers(() -> {
             for (int i = 0; i < EVENTS_PER_PRODUCER; i++) {
-                while (reactorSink.tryEmitNext(EVENT) == reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED) {
-                    java.util.concurrent.locks.LockSupport.parkNanos(10);
-                }
+                reactor.core.publisher.Sinks.EmitResult res;
+                do {
+                    res = reactorSink.tryEmitNext(EVENT);
+                    if (res != reactor.core.publisher.Sinks.EmitResult.OK) {
+                        java.util.concurrent.locks.LockSupport.parkNanos(10);
+                    }
+                } while (res != reactor.core.publisher.Sinks.EmitResult.OK);
             }
         });
         latch.await();

zthread-benchmark/src/main/java/io/github/namanoncode/zthread/benchmark/throughput/SpscEventBenchmark.java

@@ -194,9 +194,13 @@ public void benchReactor(Blackhole bh) throws InterruptedException {
             });
 
         for (int i = 0; i < BATCH_SIZE; i++) {
-            while (reactorSink.tryEmitNext(EVENT) == reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED) {
-                java.util.concurrent.locks.LockSupport.parkNanos(10);
-            }
+            reactor.core.publisher.Sinks.EmitResult res;
+            do {
+                res = reactorSink.tryEmitNext(EVENT);
+                if (res != reactor.core.publisher.Sinks.EmitResult.OK) {
+                    java.util.concurrent.locks.LockSupport.parkNanos(10);
+                }
+            } while (res != reactor.core.publisher.Sinks.EmitResult.OK);
         }
         latch.await();
     }