Improved sync/async benchmarks

benchmarks-api/src/main/java/io/scalecube/benchmarks/BenchmarksState.java

@@ -7,6 +7,7 @@
 import com.codahale.metrics.Timer;
 import com.codahale.metrics.jvm.MemoryUsageGaugeSet;
 
+import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
@@ -18,7 +19,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -207,12 +210,18 @@ public final void runForSync(Function<SELF, Function<Long, Object>> func) {
 
       Function<Long, Object> unitOfWork = func.apply(self);
 
-      Flux<Long> fromStream = Flux.fromStream(LongStream.range(0, settings.numOfIterations()).boxed());
+      CountDownLatch latch = new CountDownLatch(1);
 
-      Flux.merge(fromStream
-          .publishOn(scheduler()).map(unitOfWork))
-          .take(settings.executionTaskDuration())
-          .blockLast();
+      Flux.fromStream(LongStream.range(0, settings.numOfIterations()).boxed())
+          .parallel()
+          .runOn(scheduler())
+          .map(unitOfWork)
+          .doOnTerminate(latch::countDown)
+          .subscribe();
+
+      latch.await(settings.executionTaskDuration().toMillis(), TimeUnit.MILLISECONDS);
+    } catch (InterruptedException e) {
+      throw Exceptions.propagate(e);
     } finally {
       self.shutdown();
     }
@@ -280,7 +289,7 @@ public final <T> void runWithRampUp(
 
       BiFunction<Long, T, Publisher<?>> unitOfWork = func.apply(self);
 
-      Flux.interval(settings.rampUpInterval())
+      Flux.interval(Duration.ZERO, settings.rampUpInterval())
           .take(settings.rampUpDuration())
           .flatMap(rampUpIteration -> {
 

benchmarks-examples/src/main/java/io/scalecube/benchmarks/examples/ExampleService.java

@@ -5,7 +5,11 @@
 public class ExampleService {
 
   public Mono<String> invoke(String request) {
-    return Mono.defer(() -> Mono.just(request + hardTask()));
+    return Mono.defer(() -> Mono.just(syncInvoke(request)));
+  }
+
+  public String syncInvoke(String request) {
+    return request + hardTask();
   }
 
   private Double hardTask() {

benchmarks-examples/src/main/java/io/scalecube/benchmarks/examples/SyncExampleBenchmarksRunner.java

@@ -0,0 +1,34 @@
+package io.scalecube.benchmarks.examples;
+
+import io.scalecube.benchmarks.BenchmarksSettings;
+
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.Timer;
+
+import java.util.concurrent.TimeUnit;
+
+public class SyncExampleBenchmarksRunner {
+
+  /**
+   * Runs example benchmark.
+   *
+   * @param args command line args
+   */
+  public static void main(String[] args) {
+    BenchmarksSettings settings = BenchmarksSettings.from(args).durationUnit(TimeUnit.NANOSECONDS).build();
+    new ExampleServiceBenchmarksState(settings).runForSync(state -> {
+
+      ExampleService service = state.exampleService();
+      Timer timer = state.timer("timer");
+      Meter meter = state.meter("meter");
+
+      return i -> {
+        Timer.Context timeContext = timer.time();
+        String result = service.syncInvoke("hello");
+        timeContext.stop();
+        meter.mark();
+        return result;
+      };
+    });
+  }
+}