Feature/mono processor rework (#723)

* reworking unicast mono processor impl

This reverts commit c5684eb.

Signed-off-by: Oleh Dokuka <[email protected]>

* reworking unicast mono processor impl

This reverts commit c5684eb.

Signed-off-by: Oleh Dokuka <[email protected]>

* provides perf comparison

Signed-off-by: Oleh Dokuka <[email protected]>

* provides more strict tests

Signed-off-by: Oleh Dokuka <[email protected]>

* provides formatting

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

benchmarks/src/main/java/io/rsocket/MaxPerfSubscriber.java

@@ -5,12 +5,12 @@
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
 
-public class MaxPerfSubscriber implements CoreSubscriber<Payload> {
+public class MaxPerfSubscriber<T> extends CountDownLatch implements CoreSubscriber<T> {
 
-  final CountDownLatch latch = new CountDownLatch(1);
   final Blackhole blackhole;
 
   public MaxPerfSubscriber(Blackhole blackhole) {
+    super(1);
     this.blackhole = blackhole;
   }
 
@@ -20,19 +20,18 @@ public void onSubscribe(Subscription s) {
   }
 
   @Override
-  public void onNext(Payload payload) {
-    payload.release();
+  public void onNext(T payload) {
     blackhole.consume(payload);
   }
 
   @Override
   public void onError(Throwable t) {
     blackhole.consume(t);
-    latch.countDown();
+    countDown();
   }
 
   @Override
   public void onComplete() {
-    latch.countDown();
+    countDown();
   }
 }

benchmarks/src/main/java/io/rsocket/PayloadsMaxPerfSubscriber.java

@@ -0,0 +1,16 @@
+package io.rsocket;
+
+import org.openjdk.jmh.infra.Blackhole;
+
+public class PayloadsMaxPerfSubscriber extends MaxPerfSubscriber<Payload> {
+
+  public PayloadsMaxPerfSubscriber(Blackhole blackhole) {
+    super(blackhole);
+  }
+
+  @Override
+  public void onNext(Payload payload) {
+    payload.release();
+    super.onNext(payload);
+  }
+}

benchmarks/src/main/java/io/rsocket/PerfSubscriber.java

@@ -5,14 +5,14 @@
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
 
-public class PerfSubscriber<T> implements CoreSubscriber<T> {
+public class PerfSubscriber<T> extends CountDownLatch implements CoreSubscriber<T> {
 
-  public final CountDownLatch latch = new CountDownLatch(1);
   final Blackhole blackhole;
 
   Subscription s;
 
   public PerfSubscriber(Blackhole blackhole) {
+    super(1);
     this.blackhole = blackhole;
   }
 
@@ -31,11 +31,11 @@ public void onNext(T payload) {
   @Override
   public void onError(Throwable t) {
     blackhole.consume(t);
-    latch.countDown();
+    countDown();
   }
 
   @Override
   public void onComplete() {
-    latch.countDown();
+    countDown();
   }
 }

benchmarks/src/main/java/io/rsocket/RSocketPerf.java

@@ -112,7 +112,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
   public PayloadsPerfSubscriber fireAndForget(Blackhole blackhole) throws InterruptedException {
     PayloadsPerfSubscriber subscriber = new PayloadsPerfSubscriber(blackhole);
     client.fireAndForget(PAYLOAD).subscribe((CoreSubscriber) subscriber);
-    subscriber.latch.await();
+    subscriber.await();
 
     return subscriber;
   }
@@ -121,7 +121,7 @@ public PayloadsPerfSubscriber fireAndForget(Blackhole blackhole) throws Interrup
   public PayloadsPerfSubscriber requestResponse(Blackhole blackhole) throws InterruptedException {
     PayloadsPerfSubscriber subscriber = new PayloadsPerfSubscriber(blackhole);
     client.requestResponse(PAYLOAD).subscribe(subscriber);
-    subscriber.latch.await();
+    subscriber.await();
 
     return subscriber;
   }
@@ -131,17 +131,17 @@ public PayloadsPerfSubscriber requestStreamWithRequestByOneStrategy(Blackhole bl
       throws InterruptedException {
     PayloadsPerfSubscriber subscriber = new PayloadsPerfSubscriber(blackhole);
     client.requestStream(PAYLOAD).subscribe(subscriber);
-    subscriber.latch.await();
+    subscriber.await();
 
     return subscriber;
   }
 
   @Benchmark
-  public MaxPerfSubscriber requestStreamWithRequestAllStrategy(Blackhole blackhole)
+  public PayloadsMaxPerfSubscriber requestStreamWithRequestAllStrategy(Blackhole blackhole)
       throws InterruptedException {
-    MaxPerfSubscriber subscriber = new MaxPerfSubscriber(blackhole);
+    PayloadsMaxPerfSubscriber subscriber = new PayloadsMaxPerfSubscriber(blackhole);
     client.requestStream(PAYLOAD).subscribe(subscriber);
-    subscriber.latch.await();
+    subscriber.await();
 
     return subscriber;
   }
@@ -151,17 +151,17 @@ public PayloadsPerfSubscriber requestChannelWithRequestByOneStrategy(Blackhole b
       throws InterruptedException {
     PayloadsPerfSubscriber subscriber = new PayloadsPerfSubscriber(blackhole);
     client.requestChannel(PAYLOAD_FLUX).subscribe(subscriber);
-    subscriber.latch.await();
+    subscriber.await();
 
     return subscriber;
   }
 
   @Benchmark
-  public MaxPerfSubscriber requestChannelWithRequestAllStrategy(Blackhole blackhole)
+  public PayloadsMaxPerfSubscriber requestChannelWithRequestAllStrategy(Blackhole blackhole)
       throws InterruptedException {
-    MaxPerfSubscriber subscriber = new MaxPerfSubscriber(blackhole);
+    PayloadsMaxPerfSubscriber subscriber = new PayloadsMaxPerfSubscriber(blackhole);
     client.requestChannel(PAYLOAD_FLUX).subscribe(subscriber);
-    subscriber.latch.await();
+    subscriber.await();
 
     return subscriber;
   }

benchmarks/src/main/java/io/rsocket/internal/UnicastVsDefaultMonoProcessorPerf.java

@@ -0,0 +1,46 @@
+package io.rsocket.internal;
+
+import io.rsocket.MaxPerfSubscriber;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import reactor.core.publisher.MonoProcessor;
+
+@BenchmarkMode({Mode.Throughput, Mode.SampleTime})
+@Fork(1)
+@Warmup(iterations = 10)
+@Measurement(iterations = 10, time = 20)
+@State(Scope.Benchmark)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+public class UnicastVsDefaultMonoProcessorPerf {
+
+  @Benchmark
+  public void monoProcessorPerf(Blackhole bh) {
+    MaxPerfSubscriber<Integer> subscriber = new MaxPerfSubscriber<>(bh);
+    MonoProcessor<Integer> monoProcessor = MonoProcessor.create();
+    monoProcessor.onNext(1);
+    monoProcessor.subscribe(subscriber);
+
+    bh.consume(monoProcessor);
+    bh.consume(subscriber);
+  }
+
+  @Benchmark
+  public void unicastMonoProcessorPerf(Blackhole bh) {
+    MaxPerfSubscriber<Integer> subscriber = new MaxPerfSubscriber<>(bh);
+    UnicastMonoProcessor<Integer> monoProcessor = UnicastMonoProcessor.create();
+    monoProcessor.onNext(1);
+    monoProcessor.subscribe(subscriber);
+
+    bh.consume(monoProcessor);
+    bh.consume(subscriber);
+  }
+}