add fragmentation support using new frames #585

robertroeser · 2019-02-26T11:04:18Z

Signed-off-by: Robert Roeser rroeserr@gmail.com

Signed-off-by: Robert Roeser <rroeserr@gmail.com>

yschimke · 2019-02-26T19:29:08Z

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

+              } catch (Throwable t) {
+                if (byteBuf != null) {
+                  ReferenceCountUtil.safeRelease(byteBuf);
+                  ReferenceCountUtil.safeRelease(payload);


Why this release as well as line 362? I can see this repeated below also, so assuming deliberate.

I can probably remove the payload one. I might have added it when I was chasing done some memory leaks.

yschimke · 2019-02-26T19:30:52Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationDuplexConnection.java

+    Objects.requireNonNull(delegate, "delegate must not be null");
+    Objects.requireNonNull(allocator, "byteBufAllocator must not be null");
+    if (mtu < MIN_MTU_SIZE) {
+      throw new IllegalArgumentException("smallest allowed mtu size is 64 bytes");


for a non hot path exception, worth referring to MIN_MTU_SIZE in the message instead of 64

yschimke · 2019-02-26T19:32:51Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationDuplexConnection.java

-            })
-        .subscribe();
+  private boolean shouldFragment(FrameType frameType, int readableBytes) {
+    if (frameType.isFragmentable()) {


tioli: more readable as a one liner

return frameType.isFragmentable() && readableBytes > mtu

Signed-off-by: Robert Roeser <rroeserr@gmail.com>

yschimke · 2019-02-26T19:45:08Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameFragmenter.java

+        return PayloadFrameFlyweight.encode(
+            allocator, streamId, true, true, false, metadataFragment, dataFragment);
+      default:
+        throw new IllegalStateException("unsupported fragment type");


nit: useful to include frameType in the message

yschimke · 2019-02-26T19:46:45Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameFragmenter.java

-      }
-    }
+    boolean follows =
+        metadata == Unpooled.EMPTY_BUFFER


why check this? doesn't metadata.isReadable() do this?

yschimke · 2019-02-26T19:47:17Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameFragmenter.java

+          metadata = PayloadFrameFlyweight.metadata(frame);
+          break;
+        default:
+          throw new IllegalStateException("unsupported fragment type");


as above include frameType

yschimke · 2019-02-26T19:48:49Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java

 *     and Reassembly</a>
 */
-final class FrameReassembler implements Disposable {
+final class FrameReassembler extends AtomicBoolean implements Disposable {


why extend AtomicBoolean instead of a field? memory?

something I picked up from rxjava

yschimke · 2019-02-26T19:50:30Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java

-          ? metadata.retain()
-          : Unpooled.wrappedBuffer(this.metadata, metadata.retain());
+    } catch (Throwable t) {
+      logger.error("error reassemble frame", t);


Suggested change

logger.error("error reassemble frame", t);

logger.error("error reassembling frame", t);

yschimke · 2019-02-26T19:52:33Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java

-      return this.metadata == null
-          ? metadata.retain()
-          : Unpooled.wrappedBuffer(this.metadata, metadata.retain());
+    } catch (Throwable t) {


This is a very generic error to handle here when there doesn't seem to be a handoff to client code, i.e. it's either an encoding problem on the wire or a logic bug in this code.

I've gotten in the habit of doing this with publishers to maker sure exceptions don't slip by - its less about handling the error and more about making sure that I call sink.error less the exception disappears in the ether.

yschimke · 2019-02-26T19:55:14Z

rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java

+  public static String toString(ByteBuf frame) {
+    FrameType frameType = FrameHeaderFlyweight.frameType(frame);
+    int streamId = FrameHeaderFlyweight.streamId(frame);
+    StringBuilder payload = new StringBuilder();


worth presizing? 16 default will require a realloc. Can be estimated pretty definitively.

maybe? this was kind of for debugging purposes, but I can add a size

yschimke · 2019-02-26T19:57:45Z

rsocket-core/src/main/java/io/rsocket/frame/RequestChannelFrameFlyweight.java

+      long requestN,
+      Payload payload) {
+
+    int reqN = requestN > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) requestN;


suggest asserting on negative values as a public util method

yschimke · 2019-02-26T19:59:49Z

rsocket-core/src/main/java/io/rsocket/frame/RequestFlyweight.java

    int flags = 0;

-    if (metadata != null) {
+    if (metadata != null && Unpooled.EMPTY_BUFFER != metadata) {


Isn't EMPTY_BUFFER the "" case? and null is the case where flag should be unset?

yeah I that is correct - that would explain the other thing I changed in the unit test - it was kind of early - I will go through and check the flyweights.

yschimke · 2019-02-26T20:01:54Z

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


    @Override
    public Mono<Void> send(Publisher<ByteBuf> frame) {
      if (debugEnabled) {


being pedantic, shouldn't debug levels be able to change in a running app?

I set this way so that eventually it would get JIT'd away and you could only set it startup so you don't accidentally turn on frame logging.

Your call, it's not common.

yschimke · 2019-02-26T20:03:53Z

rsocket-core/src/main/java/io/rsocket/util/ByteBufPayload.java

+
+  @Override
+  public ByteBuf sliceMetadata() {
+    return metadata == null ? Unpooled.EMPTY_BUFFER : metadata.slice();


I wonder whether this implicit conversion from null to EMPTY_BUFFER could be confusing for the null/"" edge case

yschimke · 2019-02-26T20:04:54Z

rsocket-core/src/test/java/io/rsocket/fragmentation/FragmentationIntegrationTest.java

+
+  private ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;
+
+  @DisplayName("fragments and reassmbles data")


Suggested change

@DisplayName("fragments and reassmbles data")

@DisplayName("fragments and reassembles data")

yschimke · 2019-02-26T20:05:05Z

rsocket-core/src/test/java/io/rsocket/fragmentation/FragmentationIntegrationTest.java

+  @DisplayName("fragments and reassmbles data")
+  @Test
+  void fragmentAndReassembleData() {
+    System.out.println("original");


yschimke · 2019-02-26T20:06:13Z

rsocket-core/src/test/java/io/rsocket/frame/RequestFlyweightTest.java

    frame = FrameLengthFlyweight.encode(ByteBufAllocator.DEFAULT, frame.readableBytes(), frame);

-    assertEquals("00000e0000000119000000000100000064", ByteBufUtil.hexDump(frame));
+    assertEquals("00000b0000000118000000000164", ByteBufUtil.hexDump(frame));


Was this a bug?

yschimke · 2019-02-26T20:07:21Z

rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java

+   * Creates a new instance
+   *
+   * @param encodeLength indicates if this connection should encode the length or not.
+   * @param connection the {@link Connection} to for managing the server


yschimke

LGTM

mostroverkhov · 2019-02-26T22:11:39Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java

+        return;
+      }
+
+      boolean hasFollows = FrameHeaderFlyweight.hasFollows(frame);


nit: there is a mention in a spec that follows flag should be ignored if complete flag is set

mostroverkhov · 2019-02-26T22:14:30Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java

+    CompositeByteBuf cd = removeData(streamId);
+    if (cd != null) {
+      ByteBuf d = PayloadFrameFlyweight.data(frame);
+      if (d != null) {


nit: data is not nullable

mostroverkhov · 2019-02-26T22:24:33Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameFragmenter.java

+      case NEXT_COMPLETE:
+        return PayloadFrameFlyweight.encode(
+            allocator, streamId, true, true, true, metadataFragment, dataFragment);
+      case COMPLETE:


Is this possible for first fragment? COMPLETE is synthetic type for frame not having Payload (data/metadata) - unlikely to be fragmented

mostroverkhov · 2019-02-26T22:40:05Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameFragmenter.java

+            allocator, streamId, true, false, true, metadataFragment, dataFragment);
+      case NEXT_COMPLETE:
+        return PayloadFrameFlyweight.encode(
+            allocator, streamId, true, true, true, metadataFragment, dataFragment);


nit: spec mentions follows flag should be ignored if complete flag is set - wondering if complete flag should be deferred to last fragment?

not sure - the spec is a little unclear around fragmentation. If we send it last we can't generate the header until the last frame, and the metadata is supposed to come complete before the data - so how do you read it? Also I don't sent next_complete frames in the fragments i send next frames. Maybe we should change the spec because it seems more complicated to defer this till the end.

mostroverkhov · 2019-02-26T23:10:54Z

rsocket-core/src/main/java/io/rsocket/fragmentation/FrameFragmenter.java

+        // Payload and synthetic types
+      case PAYLOAD:
+        return PayloadFrameFlyweight.encode(
+            allocator, streamId, true, false, false, metadataFragment, dataFragment);


next should be set - cant have both next and complete false

Signed-off-by: Robert Roeser <rroeserr@gmail.com>

* add fragmentation support using new frames Signed-off-by: Robert Roeser <rroeserr@gmail.com> Signed-off-by: Maksym Ostroverkhov <m.ostroverkhov@gmail.com>

The flags were introduced in rsocket#585 such that when fragmentation is enabled, it takes over the encoding and decoding of the frame length instead of the underlying transport. However this doesn't seem to have any obvious purpose. On the sending side, fragmentation breaks down the frames and then the transport can add the length just as well. On the receiving side, the transport is first anyway. It's better to simplify this and leave it to transports to decide this whether a length is needed and expected.

The flags were introduced in rsocket#585 such that when fragmentation is enabled, it takes over the encoding and decoding of the frame length instead of the underlying transport. However this doesn't seem to have any obvious purpose. On the sending side, fragmentation breaks down the frames and then the transport can add the length just as well. On the receiving side, the transport is first anyway. It's better to simplify this and leave it to transports to decide this whether a length is needed and expected. Signed-off-by: Rossen Stoyanchev <rstoyanchev@pivotal.io>

add fragmentation support using new frames

c41e949

Signed-off-by: Robert Roeser <rroeserr@gmail.com>

robertroeser requested review from mostroverkhov, nebhale and rdegnan February 26, 2019 19:06

yschimke reviewed Feb 26, 2019

View reviewed changes

remove comment

f717659

Signed-off-by: Robert Roeser <rroeserr@gmail.com>

yschimke reviewed Feb 26, 2019

View reviewed changes

yschimke approved these changes Feb 26, 2019

View reviewed changes

mostroverkhov reviewed Feb 26, 2019

View reviewed changes

mostroverkhov approved these changes Feb 26, 2019

View reviewed changes

rdegnan approved these changes Feb 27, 2019

View reviewed changes

robertroeser added 2 commits February 27, 2019 10:43

updates

a2c1ff9

Signed-off-by: Robert Roeser <rroeserr@gmail.com>

more updates

b31ae36

Signed-off-by: Robert Roeser <rroeserr@gmail.com>

robertroeser merged commit b15a1fc into develop Feb 27, 2019

robertroeser deleted the feature/fragmentationNewFrames branch February 27, 2019 19:12

OlegDokuka mentioned this pull request Apr 17, 2020

Fix Fragmentation from framing change #577

Closed

	logger.error("error reassemble frame", t);
	logger.error("error reassembling frame", t);


		private ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;

		@DisplayName("fragments and reassmbles data")

	@DisplayName("fragments and reassmbles data")
	@DisplayName("fragments and reassembles data")

add fragmentation support using new frames #585

add fragmentation support using new frames #585

Uh oh!

Conversation

robertroeser commented Feb 26, 2019

Uh oh!

yschimke Feb 26, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yschimke Feb 26, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

robertroeser Feb 26, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yschimke left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

yschimke Feb 26, 2019 •

edited

Loading

yschimke Feb 26, 2019 •

edited

Loading

robertroeser Feb 26, 2019 •

edited

Loading