RATIS-2546. Add input stream to DataStreamApi for read operations in Client

[peterxcli]

What changes were proposed in this pull request?

• Added a new DataStreamInput interface and its implementation DataStreamInputImpl for asynchronous, zero-copy read-only data streaming, including the readAsync() method for consuming replies and proper resource release on close.
• Extended the DataStreamApi interface and its implementation to provide streamReadOnly() methods for creating read-only streams.
• Introduced the new DataStreamReplyByteBuf class to support replies backed by Netty ByteBuf.
• Added a new overload of streamAsync() in the DataStreamClientRpc interface that takes a reply consumer, supporting multiple replies per request.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2546

How was this patch tested?

(Please explain how this patch was tested. Ex: unit tests, manual tests)
(If this patch involves UI changes, please attach a screen-shot; otherwise, remove this)

[peterxcli]

cc @szetszwo

[peterxcli]

The no-MD5 sequential reads comparison:

Key size	Buffer	ReadBlock stream	Ratis data stream	Ratis / ReadBlock bandwidth
256 MiB	32 MiB	333.23 MB/s	448.04 MB/s	1.34x bandwidth
256 MiB	8 MiB	651.57 MB/s	753.94 MB/s	1.16x bandwidth
256 MiB	1 MiB	346.72 MB/s	726.91 MB/s	2.10x bandwidth
256 MiB	4 KiB	303.35 MB/s	777.43 MB/s	2.56x bandwidth
500 MiB	32 MiB	731.13 MB/s	445.67 MB/s	0.61x bandwidth
500 MiB	8 MiB	587.45 MB/s	1012.59 MB/s	1.72x bandwidth
500 MiB	1 MiB	520.76 MB/s	598.70 MB/s	1.15x bandwidth
500 MiB	4 KiB	373.36 MB/s	846.50 MB/s	2.27x bandwidth
1 GiB	32 MiB	869.43 MB/s	693.04 MB/s	0.80x bandwidth
1 GiB	8 MiB	508.69 MB/s	474.92 MB/s	0.93x bandwidth
1 GiB	1 MiB	432.13 MB/s	784.25 MB/s	1.81x bandwidth
1 GiB	4 KiB	378.85 MB/s	1087.56 MB/s	2.87x bandwidth

The no-MD5 random reads comparison:

Key size	Random read size	Direct stream	Ratis data stream	Bandwidth ratio	IOPS ratio	Direct elapsed	Ratis elapsed
256 MiB	1 MiB	28.92 MB/s	103.03 MB/s	3.56x bandwidth	3.56x IOPS	1.106 s	0.311 s
256 MiB	4 KiB	0.12 MB/s	0.38 MB/s	3.19x bandwidth	3.19x IOPS	1.059 s	0.332 s
500 MiB	1 MiB	44.18 MB/s	255.86 MB/s	5.79x bandwidth	5.79x IOPS	0.724 s	0.125 s
500 MiB	4 KiB	0.14 MB/s	0.59 MB/s	4.06x bandwidth	4.06x IOPS	0.865 s	0.213 s
1 GiB	1 MiB	33.08 MB/s	193.64 MB/s	5.85x bandwidth	5.85x IOPS	0.967 s	0.165 s
1 GiB	4 KiB	0.10 MB/s	0.87 MB/s	8.34x bandwidth	8.34x IOPS	1.197 s	0.144 s

I'm looking into why larger buffer would cause throughput to degrade.

[szetszwo]

@peterxcli , thanks a lot for working on this!

The PR is quite big. Let's separate the DataStreamReply change first. Please see the comments inlined and also https://issues.apache.org/jira/secure/attachment/13082807/1481_review_DataStreamReply.patch (updated the link in the next comment.)

[szetszwo]

@peterxcli , thanks for updating this!

Let's have one more simple change to add an executor and terminalReply (RATIS-2564). Please see the comments inlined and also https://issues.apache.org/jira/secure/attachment/13082846/1481_review3_refactoring.patch

[szetszwo]

@peterxcli , Let's borrow the idea from gRPC StreamObserver and add a similar interface. It is more flexible for future changes.

//ratis-common
package org.apache.ratis.datastream;

/** An interface similar to gRPC {@link org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver}. */
public interface DataStreamObserver<V> {
  void onNext(V value);

  // see if onError(Throwable) and onCompleted() are useful.  Or we may add them later.
}

Then our interface will be similar to gRPC service https://github.com/apache/ratis/blob/master/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcServerProtocolClient.java#L174-L181

//DataStreamClientRpc
  /** Async call to send a request and receive multiple replies for the request. */
  default CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest request,
      DataStreamObserver<DataStreamReplyByteBuf> replyHandler) {
    throw new UnsupportedOperationException(getClass() + " does not support "
        + JavaUtils.getCurrentStackTraceElement().getMethodName());
  }

[peterxcli]

@szetszwo thanks for the grpc observer suggestion, just added a initial version of it.

I also add onError and onComplete interface method for the observer, because we can decide which hook to call base on the terminal reply.

Please take a look, Thanks!

[amaliujia]

Do we need to call close() or mark closed = true; here?

[peterxcli]

no, the caller might have not counsumed the fetched content yet, if we set close, the readAsync would fail,

[peterxcli]

I think I can refactor this to get better readability.

[amaliujia]

I do not understand why we need to do a failReads upon a ordinary close. Do we not expect close be call in the happy path and we only call close in bad cases?

[peterxcli]

close() could happen while readAsync() calls are pending.

[amaliujia]

What about else? Is that possible other instances are passed here? If no maybe add an assertion, otherwise handle other instances?

[peterxcli]

There are two classes implement the DataStreamReply, which is DataStreamReplyByteBuf and DataStreamReplyByteBuffer.
And only DataStreamReplyByteBuf need to release the Netty byteBuffer reference count

ratis/ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamPacketByteBuf.java

Lines 58 to 63 in 3cb6794

	public final void release() {
	final ByteBuf got = buf.getAndSet(null);
	if (got != null) {
	got.release();
	}
	}

[szetszwo]

@peterxcli , thanks for the update!

In DataStreamInput, readAsync() should return a ReferenceCountedObject so the user can retain/release the buffer.

Please see the comments inlined and also https://issues.apache.org/jira/secure/attachment/13082885/1481_review4.patch

[szetszwo]

If we add them, we should enforce implementing them. Let's remove default.

  void onCompleted();

  void onError(Throwable throwable);

[szetszwo]

We should use reference-counted:

  /**
   * Read the next chunk in the stream asynchronously.
   * The caller owns the returned {@link DataStreamReply} which is a {@link ReferenceCountedObject}.
   * It must call {@link ReferenceCountedObject#release()} after consuming it.
   *
   * @return a future of the reference-counted reply.
   */
  CompletableFuture<ReferenceCountedObject<DataStreamReply>> readAsync();

[szetszwo]

We should use reference-counted:

  /**
   * Async call to send a request to receive a stream of intermediate replies and a final reply.
   *
   * @param request the request
   * @param replyHandler to handle intermediate replies
   * @return a future the final reply
   */
  default CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest request,
      DataStreamObserver<ReferenceCountedObject<DataStreamReply>> replyHandler) {
    throw new UnsupportedOperationException(getClass() + " does not support "
        + JavaUtils.getCurrentStackTraceElement().getMethodName());
  }

[szetszwo]

This will release the reply even before user application reading it. We should use reference-counted:

        final DataStreamReplyByteBuf reply = (DataStreamReplyByteBuf) msg;
        final ReferenceCountedObject<DataStreamReply> ref = ReferenceCountedObject.<DataStreamReply>newBuilder()
            .setValue((DataStreamReplyByteBuf) msg)
            .setReleaseMethod(r -> {
              if (r != null) {
                Preconditions.assertSame(reply, r, "reply");
                reply.release();
              }
            }).build();
        try (UncheckedAutoCloseableSupplier<DataStreamReply> ignored = ref.retainAndReleaseOnClose()) {
          process(ref);
        }

[szetszwo]

We should use reference-counted in the queues and avoid copying the reply.