ci fix 5 benchmark

Author: namanONcode

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

@@ -24,7 +24,9 @@ public void start(EventHandler handler, int consumers) {
 
     @Override
     public void submit(BenchmarkEvent event) {
-        sink.tryEmitNext(event);
+        while (sink.tryEmitNext(event) == Sinks.EmitResult.FAIL_NON_SERIALIZED) {
+            java.util.concurrent.locks.LockSupport.parkNanos(10);
+        }
     }
 
     @Override

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

@@ -270,7 +270,9 @@ public void benchReactor(Blackhole bh) throws InterruptedException {
 
         runProducers(() -> {
             for (int i = 0; i < EVENTS_PER_PRODUCER; i++) {
-                reactorSink.tryEmitNext(EVENT);
+                while (reactorSink.tryEmitNext(EVENT) == reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED) {
+                    java.util.concurrent.locks.LockSupport.parkNanos(10);
+                }
             }
         });
         latch.await();

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

@@ -270,7 +270,9 @@ public void benchReactor(Blackhole bh) throws InterruptedException {
 
         runProducers(() -> {
             for (int i = 0; i < EVENTS_PER_PRODUCER; i++) {
-                reactorSink.tryEmitNext(EVENT);
+                while (reactorSink.tryEmitNext(EVENT) == reactor.core.publisher.Sinks.EmitResult.FAIL_NON_SERIALIZED) {
+                    java.util.concurrent.locks.LockSupport.parkNanos(10);
+                }
             }
         });
         latch.await();

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

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