Huge performance degradation when fragmentation is enabled

[koldat]

When one define custom mtu to be used for fragment size it significantly degrades performance. Attached example code sends 1M records with size 5 bytes. Before fix it takes 39 seconds. After fix it takes 5 seconds (same time as with no custom fragmentation). We need to enable it, because Websocket max data size is 64kB and we support both transports.

Expected Behavior

Fragmentation setup should not affect performance. Especially when data buffers are smaller than mtu.

Actual Behavior

There is huge performance degradation.

Steps to Reproduce

public class RsocketTest {

    private static final Logger logger = LoggerFactory.getLogger(RsocketTest.class);

    public static void main(String[] args) {
        RSocket rsocket = new RSocket() {
            @Override
            public Flux<Payload> requestStream(Payload payload) {
                ByteBuffer data = ByteBuffer.wrap(new byte[5]);
                return Flux.generate(() -> new AtomicInteger(1000000), (state, s) -> {
                    if (state.decrementAndGet() == 0) {
                        s.complete();
                    } else {
                        s.next(DefaultPayload.create(data));
                    }
                    return state;
                });
            }
        };

        RSocketServer.create(SocketAcceptor.with(rsocket))
            .fragment(60000)
            .bind(TcpServerTransport.create(
                TcpServer.create().host("localhost").port(7000)))
            .subscribe();

        RSocket socket = RSocketConnector.create()
            .fragment(60000)
            .connect(TcpClientTransport.create(
                TcpClient.create().host("localhost").port(7000)))
            .block();

        Flux<Payload> res = socket.requestStream(DefaultPayload.create(ByteBuffer.wrap("Hello".getBytes())));
        long start = System.currentTimeMillis();
        res
            .limitRate(1000)
            .blockLast();
        
        logger.info("Call took: {}", System.currentTimeMillis() - start);
        
        socket.dispose();
    }
}

Possible Solution

FragmentationDuplexConnection change (I am creating PR now):

  @Override
  public Mono<Void> send(Publisher<ByteBuf> frames) {
    return delegate.send(Flux.from(frames)
        .concatMap(frame -> {
            FrameType frameType = FrameHeaderCodec.frameType(frame);
            int readableBytes = frame.readableBytes();
            if (!shouldFragment(frameType, readableBytes)) {
              return Flux.just(frame);
            }
            
            return logFragments(Flux.from(fragmentFrame(alloc(), mtu, frame, frameType)));
        }));
  }

  @Override
  public Mono<Void> sendOne(ByteBuf frame) {
    FrameType frameType = FrameHeaderCodec.frameType(frame);
    int readableBytes = frame.readableBytes();
    if (!shouldFragment(frameType, readableBytes)) {
      return delegate.sendOne(frame);
    }
    Flux<ByteBuf> fragments = Flux.from(fragmentFrame(alloc(), mtu, frame, frameType));
    fragments = logFragments(fragments);
    return delegate.send(fragments);
  }

  protected Flux<ByteBuf> logFragments(Flux<ByteBuf> fragments) {
    if (logger.isDebugEnabled()) {
      fragments =
          fragments.doOnNext(
              byteBuf -> {
                logger.debug(
                    "{} - stream id {} - frame type {} - \n {}",
                    type,
                    FrameHeaderCodec.streamId(byteBuf),
                    FrameHeaderCodec.frameType(byteBuf),
                    ByteBufUtil.prettyHexDump(byteBuf));
              });
    }
    return fragments;
  }

Your Environment

• RSocket version(s) used: 1.0.2

@OlegDokuka can you take a look? Can we please include this into 1.0.4?

[OlegDokuka]

@koldat I can suggest migrating to 1.1.x. We are not planning to do ground rework in 1.0.x anymore and planning to do maintenance releases.

However, if you have some performance profiles that show hot places in our code, and if it is not going to be drastic to improve, we can try to do something about it.

[OlegDokuka]

We need to enable it, because Websocket max data size is 64kB and we support both transports.

Also, this can be tuned. Have you seen this config -> https://github.com/reactor/reactor-netty/blob/master/reactor-netty-http/src/main/java/reactor/netty/http/websocket/WebsocketSpec.java#L92

[OlegDokuka]

@koldat can you explain in more detail what kind of client-server over rsocket + WebSocket transport you have? Is it RSocket-Java <-> RSocket-Java case?

[koldat]

1. Impact is 39 vs. 5 seconds. I would call it drastic improvement.
2. Current Spring Boot 2.3.x release is still running on old 1.0.x release. That is why I would prefer that compared to custom build.
3. we have Websocket or socket client going through gateway service routing (websocket, socket) going to final service (socket). All is Java and 1.0.2 version.
4. Thank you pointing to websocket configuration. I have overlooked that. It can be nice workaround. I will try.

Thanks for help.

[koldat]

Still original issue stays. Even if I tune Websocket frame to 16M (it is actually your default that cannot be changed in WebsocketClientTransport) I am in original problem you describe in PR. One frame blocks my connection and others go to queue.

If I enable frame even 1M, I am in original slowdown. As I have described, even if you do not hit the limit sendOne is used which degrades performance for small items. This basically says we cannot have APIs with small entities, but only bigger chunks of data.

[rstoyanchev]

@koldat for future reference, it is not required to create both an issue and a PR. If you submit both, the issue is closed as superseded, so if you start with a PR you can submit only that.