perf changes (#552)

* reset to snapshots for next release

* Switching FrameType flags back to bits (#548)

* switching frame type flag back to bits so we don't have to traverse an array when looking up flags

* jmh test


* Added an EMPTY flag


* changes from performance testing for Netty transport, and underlying RSocket core protocol


* removing unused imports



* switched to netty's IntObjectMap

Signed-off-by: Robert Roeser <[email protected]>

Author: robertroeser

gradle.properties

@@ -12,4 +12,4 @@
 # limitations under the License.
 #
 
-version=0.11.13.BUILD-SNAPSHOT
+version=0.11.14.BUILD-SNAPSHOT

rsocket-core/src/main/java/io/rsocket/RSocketClient.java

@@ -16,22 +16,27 @@
 
 package io.rsocket;
 
+import io.netty.util.collection.IntObjectHashMap;
 import io.rsocket.exceptions.ConnectionErrorException;
 import io.rsocket.exceptions.Exceptions;
 import io.rsocket.framing.FrameType;
 import io.rsocket.internal.LimitableRequestPublisher;
 import io.rsocket.internal.UnboundedProcessor;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
+import reactor.core.publisher.UnicastProcessor;
 
 import java.nio.channels.ClosedChannelException;
 import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
-import org.jctools.maps.NonBlockingHashMapLong;
-import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
-import reactor.core.publisher.*;
 
 /** Client Side of a RSocket socket. Sends {@link Frame}s to a {@link RSocketServer} */
 class RSocketClient implements RSocket {
@@ -40,8 +45,8 @@ class RSocketClient implements RSocket {
   private final Function<Frame, ? extends Payload> frameDecoder;
   private final Consumer<Throwable> errorConsumer;
   private final StreamIdSupplier streamIdSupplier;
-  private final NonBlockingHashMapLong<LimitableRequestPublisher> senders;
-  private final NonBlockingHashMapLong<UnicastProcessor<Payload>> receivers;
+  private final Map<Integer, LimitableRequestPublisher> senders;
+  private final Map<Integer, UnicastProcessor<Payload>> receivers;
   private final UnboundedProcessor<Frame> sendProcessor;
   private KeepAliveHandler keepAliveHandler;
   private final Lifecycle lifecycle = new Lifecycle();
@@ -69,8 +74,8 @@ class RSocketClient implements RSocket {
     this.frameDecoder = frameDecoder;
     this.errorConsumer = errorConsumer;
     this.streamIdSupplier = streamIdSupplier;
-    this.senders = new NonBlockingHashMapLong<>(256);
-    this.receivers = new NonBlockingHashMapLong<>(256);
+    this.senders   = Collections.synchronizedMap(new IntObjectHashMap<>());
+    this.receivers = Collections.synchronizedMap(new IntObjectHashMap<>());
 
     // DO NOT Change the order here. The Send processor must be subscribed to before receiving
     this.sendProcessor = new UnboundedProcessor<>();

rsocket-core/src/main/java/io/rsocket/RSocketServer.java

@@ -16,23 +16,29 @@
 
 package io.rsocket;
 
-import static io.rsocket.Frame.Request.initialRequestN;
-import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_C;
-import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M;
-
+import io.netty.util.collection.IntObjectHashMap;
 import io.rsocket.exceptions.ApplicationErrorException;
 import io.rsocket.exceptions.ConnectionErrorException;
 import io.rsocket.framing.FrameType;
 import io.rsocket.internal.LimitableRequestPublisher;
 import io.rsocket.internal.UnboundedProcessor;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import org.jctools.maps.NonBlockingHashMapLong;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import reactor.core.Disposable;
-import reactor.core.publisher.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
+import reactor.core.publisher.UnicastProcessor;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static io.rsocket.Frame.Request.initialRequestN;
+import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_C;
+import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M;
 
 /** Server side RSocket. Receives {@link Frame}s from a {@link RSocketClient} */
 class RSocketServer implements RSocket {
@@ -42,8 +48,8 @@ class RSocketServer implements RSocket {
   private final Function<Frame, ? extends Payload> frameDecoder;
   private final Consumer<Throwable> errorConsumer;
 
-  private final NonBlockingHashMapLong<Subscription> sendingSubscriptions;
-  private final NonBlockingHashMapLong<UnicastProcessor<Payload>> channelProcessors;
+  private final Map<Integer, Subscription> sendingSubscriptions;
+  private final Map<Integer, UnicastProcessor<Payload>> channelProcessors;
 
   private final UnboundedProcessor<Frame> sendProcessor;
   private KeepAliveHandler keepAliveHandler;
@@ -69,8 +75,8 @@ class RSocketServer implements RSocket {
     this.requestHandler = requestHandler;
     this.frameDecoder = frameDecoder;
     this.errorConsumer = errorConsumer;
-    this.sendingSubscriptions = new NonBlockingHashMapLong<>();
-    this.channelProcessors = new NonBlockingHashMapLong<>();
+    this.sendingSubscriptions = Collections.synchronizedMap(new IntObjectHashMap<>());
+    this.channelProcessors    = Collections.synchronizedMap(new IntObjectHashMap<>());
 
     // DO NOT Change the order here. The Send processor must be subscribed to before receiving
     // connections

rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java

@@ -17,10 +17,6 @@
 package io.rsocket.internal;
 
 import io.netty.util.ReferenceCountUtil;
-import java.util.Objects;
-import java.util.Queue;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
@@ -32,6 +28,11 @@
 import reactor.util.concurrent.Queues;
 import reactor.util.context.Context;
 
+import java.util.Objects;
+import java.util.Queue;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
 /**
  * A Processor implementation that takes a custom queue and allows only a single subscriber.
  *

rsocket-test/src/main/java/io/rsocket/test/TransportTest.java

@@ -33,6 +33,7 @@
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
 
 public interface TransportTest {
@@ -164,16 +165,23 @@ default void requestChannel3() {
         .expectComplete()
         .verify(getTimeout());
   }
-
+  
   @DisplayName("makes 1 requestChannel request with 512 payloads")
   @Test
   default void requestChannel512() {
     Flux<Payload> payloads = Flux.range(0, 512).map(this::createTestPayload);
-
+    
+    Flux.range(0, 1024)
+        .flatMap(v -> Mono.fromRunnable(()-> check(payloads)).subscribeOn(Schedulers.elastic()), 12)
+        .blockLast();
+  }
+  
+  default void check(Flux<Payload> payloads) {
     getClient()
         .requestChannel(payloads)
         .as(StepVerifier::create)
         .expectNextCount(512)
+        .as("expected 512 items")
         .expectComplete()
         .verify(getTimeout());
   }
@@ -233,7 +241,7 @@ default void requestStream10_000() {
         .expectComplete()
         .verify(getTimeout());
   }
-
+  
   @DisplayName("makes 1 requestStream request and receives 5 responses")
   @Test
   default void requestStream5() {