Fragmentation (#481)

* Alternate Framing Abstraction

This change introduces a framing package that contains strongly typed
representations of all of the RSocket frame types.  These representations are
introduced as a separate abstraction because they specifically remove the
frame length and stream id that are an optional, transport-layer concern. This
allows will allow the main body of the code to be gradually moved over to an
implementation that only deals with the logical frame types, and pushes the
decoration of the frames with frame length and/or stream id down to the
transport layer.  Note that no implementation has been converted over; this is
just the prework to have an the frame types ready for migration.

These frame type representations are all implemented using the Flyweight
pattern (i.e. use the Netty Recycler) and preserve the zero-copy nature of the
existing frame representation.  These implementations have consistent and
symmetric management of ByteBuf retain and release lifecycle, in addition to
well documented and proper nullability enforced APIs, specific to each frame
type. Each implementation is also tested against raw binary representations to
ensure the strictest specification compliance.

Finally, improvements were made to testing with regards to configurable test
logging, an AssertJ ByteBuf Representation for visual comparison on test
failure, and other conveniences.

* Improvements to Fragmentation and Reassembly

This change improves the implementation and test coverage of the fragmentation
and reassembly functionality.  The fragmentation functionality now support all
frame types (previously, only the PAYLOAD frame could be fragmented) and
ensures that a fragmentation and reassembly roundtrip results in an identical
frame.

The implementations of the fragmentation and reassembly functionality has been
implemented on top of the new framing abstraction for testability and to start
moving off of the abstraction that leaks frame lengths and stream ids.  Note
that the implementation of DuplexConnection *does not* move to the new framing
abstraction.  As this interface is not yet migrated, the
FragmentationDuplexConnection is responsible for translating back and forth
between the old abstraction and the new one.  This has an implication that
overhead will increase (although minimally because of zero-copy data and the
use of Flyweights) temporarily during the migration to the new abstraction.
Once that abstraction has been adopted everywhere, the translation will be
removed, and the overheard will go back to it's current levels.

Author: nebhale

build.gradle

@@ -40,6 +40,7 @@ subprojects {
         }
 
         dependencies {
+            dependency 'ch.qos.logback:logback-classic:1.2.3'
             dependency 'com.google.code.findbugs:jsr305:3.0.2'
             dependency 'com.netflix.spectator:spectator-api:0.63.0'
             dependency 'io.netty:netty-buffer:4.1.21.Final'
@@ -49,6 +50,7 @@ subprojects {
             dependency 'org.jctools:jctools-core:2.1.2'
             dependency 'org.mockito:mockito-core:2.16.0'
             dependency 'org.openjdk.jmh:jmh-core:1.20'
+            dependency 'org.slf4j:slf4j-api:1.7.25'
 
             dependencySet(group: 'org.junit.jupiter', version: '5.1.0') {
                 entry 'junit-jupiter-api'
@@ -66,11 +68,6 @@ subprojects {
                 entry 'jmh-core'
                 entry 'jmh-generator-annprocess'
             }
-
-            dependencySet(group: 'org.slf4j', version: '1.7.25') {
-                entry 'slf4j-api'
-                entry 'slf4j-nop'
-            }
         }
     }
 
@@ -86,6 +83,14 @@ subprojects {
             options.compilerArgs << '-Xlint:all,-overloads,-rawtypes,-unchecked'
         }
 
+        javadoc {
+            options.with {
+                links 'https://docs.oracle.com/javase/8/docs/api/'
+                links 'https://projectreactor.io/docs/core/release/api/'
+                links 'https://netty.io/4.1/api/'
+            }
+        }
+
         test {
             useJUnitPlatform()
         }

rsocket-core/build.gradle

@@ -27,6 +27,7 @@ plugins {
 dependencies {
     api 'io.netty:netty-buffer'
     api 'io.projectreactor:reactor-core'
+    api 'io.projectreactor.addons:reactor-extra'
 
     implementation 'com.google.code.findbugs:jsr305'
     implementation 'org.jctools:jctools-core'
@@ -37,6 +38,7 @@ dependencies {
     testImplementation 'org.junit.jupiter:junit-jupiter-api'
     testImplementation 'org.mockito:mockito-core'
 
+    testRuntimeOnly 'ch.qos.logback:logback-classic'
     testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
 
     // TODO: Remove after JUnit5 migration

rsocket-core/jmh.gradle

@@ -16,11 +16,9 @@
 
 dependencies {
     jmh configurations.api
+    jmh configurations.implementation
     jmh 'org.openjdk.jmh:jmh-core'
     jmh 'org.openjdk.jmh:jmh-generator-annprocess'
-
-    jmhRuntime 'org.openjdk.jmh:jmh-core'
-    jmhRuntime 'org.slf4j:slf4j-nop'
 }
 
 jmhCompileGeneratedClasses.enabled = false
@@ -36,7 +34,6 @@ jmh {
 
 jmhJar {
     from project.configurations.jmh
-    from project.configurations.jmhRuntime
 }
 
 tasks.jmh.finalizedBy tasks.jmhReport

rsocket-core/src/jmh/java/io/rsocket/FragmentationPerf.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2015-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.fragmentation;
+
+import static io.netty.buffer.UnpooledByteBufAllocator.DEFAULT;
+import static io.rsocket.framing.RequestResponseFrame.createRequestResponseFrame;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.rsocket.framing.Frame;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+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.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import reactor.core.publisher.SynchronousSink;
+import reactor.util.context.Context;
+
+@BenchmarkMode(Mode.Throughput)
+@Fork(
+  value = 1 // , jvmArgsAppend = {"-Dio.netty.leakDetection.level=advanced"}
+)
+@Warmup(iterations = 10)
+@Measurement(iterations = 10_000)
+@State(Scope.Thread)
+public class FragmentationPerformanceTest {
+
+  @Benchmark
+  public void largeFragmentation(Input input) {
+    Frame frame =
+        input.largeFragmenter.fragment(input.largeFrame).doOnNext(Frame::dispose).blockLast();
+
+    input.bh.consume(frame);
+  }
+
+  @Benchmark
+  public void largeReassembly(Input input) {
+    input.largeFrames.forEach(frame -> input.reassembler.reassemble(frame, input.sink));
+
+    input.bh.consume(input.sink.next);
+  }
+
+  @Benchmark
+  public void smallFragmentation(Input input) {
+    Frame frame =
+        input.smallFragmenter.fragment(input.smallFrame).doOnNext(Frame::dispose).blockLast();
+
+    input.bh.consume(frame);
+  }
+
+  @Benchmark
+  public void smallReassembly(Input input) {
+    input.smallFrames.forEach(frame -> input.reassembler.reassemble(frame, input.sink));
+
+    input.bh.consume(input.sink.next);
+  }
+
+  @State(Scope.Benchmark)
+  public static class Input {
+
+    Blackhole bh;
+
+    FrameFragmenter largeFragmenter;
+
+    Frame largeFrame;
+
+    List<Frame> largeFrames;
+
+    FrameReassembler reassembler = FrameReassembler.createFrameReassembler(DEFAULT);
+
+    MockSynchronousSink<Frame> sink;
+
+    FrameFragmenter smallFragmenter;
+
+    Frame smallFrame;
+
+    List<Frame> smallFrames;
+
+    @Setup
+    public void setup(Blackhole bh) {
+      this.bh = bh;
+
+      sink = new MockSynchronousSink<>();
+
+      largeFrame =
+          createRequestResponseFrame(
+              DEFAULT, false, getRandomByteBuf(1 << 18), getRandomByteBuf(1 << 18));
+
+      smallFrame =
+          createRequestResponseFrame(DEFAULT, false, getRandomByteBuf(16), getRandomByteBuf(16));
+
+      largeFragmenter = new FrameFragmenter(DEFAULT, 1024);
+      smallFragmenter = new FrameFragmenter(ByteBufAllocator.DEFAULT, 2);
+
+      largeFrames = largeFragmenter.fragment(largeFrame).collectList().block();
+      smallFrames = smallFragmenter.fragment(smallFrame).collectList().block();
+    }
+
+    private static ByteBuf getRandomByteBuf(int size) {
+      byte[] bytes = new byte[size];
+      ThreadLocalRandom.current().nextBytes(bytes);
+      return Unpooled.wrappedBuffer(bytes);
+    }
+  }
+
+  static final class MockSynchronousSink<T> implements SynchronousSink<T> {
+
+    Throwable error;
+
+    T next;
+
+    @Override
+    public void complete() {}
+
+    @Override
+    public Context currentContext() {
+      return null;
+    }
+
+    @Override
+    public void error(Throwable e) {
+      this.error = e;
+    }
+
+    @Override
+    public void next(T t) {
+      this.next = t;
+    }
+  }
+}

rsocket-core/src/jmh/java/io/rsocket/fragmentation/FragmentationPerformanceTest.java

@@ -23,6 +23,7 @@
 import io.netty.util.AbstractReferenceCounted;
 import io.rsocket.Frame.Setup;
 import io.rsocket.frame.SetupFrameFlyweight;
+import io.rsocket.framing.FrameType;
 
 /**
  * Exposed to server for determination of RequestHandler based on mime types and SETUP metadata/data

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

@@ -33,11 +33,12 @@ public interface DuplexConnection extends Availability, Closeable {
    *
    * The passed {@code Publisher} must
    *
-   * @param frame Stream of {@code Frame}s to send on the connection.
+   * @param frames Stream of {@code Frame}s to send on the connection.
    * @return {@code Publisher} that completes when all the frames are written on the connection
    *     successfully and errors when it fails.
+   * @throws NullPointerException if {@code frames} is {@code null}
    */
-  Mono<Void> send(Publisher<Frame> frame);
+  Mono<Void> send(Publisher<Frame> frames);
 
   /**
    * Sends a single {@code Frame} on this connection and returns the {@code Publisher} representing