Removed benchVertx from Multi-Producer Suites (Mpmc, Mpsc)

Author: namanONcode

zthread-benchmark/src/main/java/io/github/namanoncode/zthread/benchmark/adapters/VertxAdapter.java

@@ -19,7 +19,9 @@ public void start(EventHandler handler, int consumers) {
         this.vertx = Vertx.vertx(options);
         this.eventBus = vertx.eventBus();
 
-        eventBus.localConsumer(ADDRESS, message -> {
+        io.vertx.core.eventbus.MessageConsumer<Object> consumer = eventBus.localConsumer(ADDRESS);
+        consumer.setMaxBufferedMessages(10_000_000);
+        consumer.handler(message -> {
             handler.onEvent((BenchmarkEvent) message.body());
         });
     }

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

@@ -110,18 +110,13 @@ public void teardownInvocation() {
         activeConcurrentQueue = null;
     }
 
-    private void runProducers(Runnable producerTask) throws InterruptedException {
-        CountDownLatch producersDone = new CountDownLatch(PRODUCERS);
-        for (int p = 0; p < PRODUCERS; p++) {
-            producerPool.submit(() -> {
-                try {
-                    producerTask.run();
-                } finally {
-                    producersDone.countDown();
-                }
-            });
+    protected void runProducers(Runnable producerTask) throws InterruptedException {
+        java.util.concurrent.ExecutorService executor = java.util.concurrent.Executors.newFixedThreadPool(PRODUCERS);
+        for (int i = 0; i < PRODUCERS; i++) {
+            executor.submit(producerTask);
         }
-        producersDone.await();
+        executor.shutdown();
+        executor.awaitTermination(1, java.util.concurrent.TimeUnit.MINUTES);
     }
 
     private void startBlockingQueueConsumers(BlockingQueue<Object> queue, Blackhole bh) {
@@ -278,25 +273,5 @@ public void benchReactor(Blackhole bh) throws InterruptedException {
         latch.await();
     }
 
-    @Benchmark
-    @OperationsPerInvocation(BATCH_SIZE)
-    public void benchVertx(Blackhole bh) throws InterruptedException {
-        CountDownLatch latch = new CountDownLatch(BATCH_SIZE);
-        io.vertx.core.eventbus.MessageConsumer<Object> consumer = vertx.eventBus().localConsumer("benchmark.address", msg -> {
-            bh.consume(msg.body());
-            latch.countDown();
-        });
 
-        CountDownLatch regLatch = new CountDownLatch(1);
-        consumer.completionHandler(res -> regLatch.countDown());
-        regLatch.await();
-
-        runProducers(() -> {
-            for (int i = 0; i < EVENTS_PER_PRODUCER; i++) {
-                vertx.eventBus().send("benchmark.address", EVENT);
-            }
-        });
-        latch.await();
-        consumer.unregister();
-    }
 }

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

@@ -278,26 +278,4 @@ public void benchReactor(Blackhole bh) throws InterruptedException {
         latch.await();
     }
 
-    @Benchmark
-    @OperationsPerInvocation(BATCH_SIZE)
-    public void benchVertx(Blackhole bh) throws InterruptedException {
-        CountDownLatch latch = new CountDownLatch(BATCH_SIZE);
-        io.vertx.core.eventbus.MessageConsumer<Object> consumer = vertx.eventBus().localConsumer("benchmark.address", msg -> {
-            bh.consume(msg.body());
-            latch.countDown();
-        });
-
-        // Wait for consumer to be registered to avoid dropping messages
-        CountDownLatch regLatch = new CountDownLatch(1);
-        consumer.completionHandler(res -> regLatch.countDown());
-        regLatch.await();
-
-        runProducers(() -> {
-            for (int i = 0; i < EVENTS_PER_PRODUCER; i++) {
-                vertx.eventBus().send("benchmark.address", EVENT);
-            }
-        });
-        latch.await();
-        consumer.unregister();
-    }
 }

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

@@ -205,7 +205,9 @@ public void benchReactor(Blackhole bh) throws InterruptedException {
     @OperationsPerInvocation(BATCH_SIZE)
     public void benchVertx(Blackhole bh) throws InterruptedException {
         CountDownLatch latch = new CountDownLatch(BATCH_SIZE);
-        io.vertx.core.eventbus.MessageConsumer<Object> consumer = vertx.eventBus().localConsumer("benchmark.address", msg -> {
+        io.vertx.core.eventbus.MessageConsumer<Object> consumer = vertx.eventBus().localConsumer("benchmark.address");
+        consumer.setMaxBufferedMessages(10_000_000);
+        consumer.handler(msg -> {
             bh.consume(msg.body());
             latch.countDown();
         });
@@ -219,6 +221,6 @@ public void benchVertx(Blackhole bh) throws InterruptedException {
             vertx.eventBus().send("benchmark.address", "bench");
         }
         latch.await();
-        consumer.unregister();
+        consumer.unregister().toCompletionStage().toCompletableFuture().join();
     }
 }