You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ratis.apache.org by "peterxcli (via GitHub)" <gi...@apache.org> on 2026/05/25 07:17:33 UTC
[PR] RATIS-1240. Add input stream to DataStreamApi for read operations [ratis]
peterxcli opened a new pull request, #1469:
URL: https://github.com/apache/ratis/pull/1469
## What changes were proposed in this pull request?
adding a new streamReadOnly(..) method to DataStreamApi and then work on DataStreamInput.
```java
//DataStreamApi.java
DataStreamInput streamReadOnly(ByteBuffer message);
```
This corresponds to the sendReadOnly(..) method in AsyncApi
```
//AsyncApi.java
CompletableFuture<RaftClientReply> sendReadOnly(Message message);
```
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/RATIS-1086
## 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations [ratis]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #1469:
URL: https://github.com/apache/ratis/pull/1469#issuecomment-4539983691
> Could you split this into two or three subtasks? One usual way is to change the server first and then the client.
Thanks for the review! Sure, I will split this into server and client patches!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #1469:
URL: https://github.com/apache/ratis/pull/1469#issuecomment-4574155640
@szetszwo Thanks for the review!
> * Phase 2: Reuse RaftClientAsynchronousProtocol.submitClientRequestAsync(..) to submit a dummy read request for linearizable checks and then call DataApi.streamReadOnly(..).
I'm ok with this direction, but curious why use `submitClientRequestAsync(...)` with a dummy read instead of reusing the existing readAsync(...) flow for the stream read body?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3314158984
##########
ratis-server-api/src/main/java/org/apache/ratis/statemachine/StateMachine.java:
##########
@@ -116,6 +116,18 @@ default CompletableFuture<DataStream> stream(RaftClientRequest request) {
return CompletableFuture.completedFuture(null);
}
+ /**
+ * Stream a read-only state machine request. Implementations may write zero or more data
+ * chunks before completing the returned future with the terminal reply message.
+ *
+ * @param request the read-only client request
+ * @param stream the output stream for response data chunks
+ * @return a future for the terminal reply message
+ */
+ default CompletableFuture<Message> streamReadOnly(RaftClientRequest request, DataChannel stream) {
Review Comment:
This method should be similar to query(Message):
```java
/**
* Similar to {@link #query(Message)} except that.
* - In {@link #query(Message)}, the state machine returns the result in a future.
* - In this method, the state machine sends the result using the given stream.
*
* @param request the client request
* @param stream the output stream to send the results
*/
default void query(Message request, DataChannel stream) {
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3314158984
##########
ratis-server-api/src/main/java/org/apache/ratis/statemachine/StateMachine.java:
##########
@@ -116,6 +116,18 @@ default CompletableFuture<DataStream> stream(RaftClientRequest request) {
return CompletableFuture.completedFuture(null);
}
+ /**
+ * Stream a read-only state machine request. Implementations may write zero or more data
+ * chunks before completing the returned future with the terminal reply message.
+ *
+ * @param request the read-only client request
+ * @param stream the output stream for response data chunks
+ * @return a future for the terminal reply message
+ */
+ default CompletableFuture<Message> streamReadOnly(RaftClientRequest request, DataChannel stream) {
Review Comment:
This method should be similar to query(Message):
```java
/**
* Similar to {@link #query(Message)} except below:
* - In {@link #query(Message)}, the state machine returns the result in a future.
* - In this method, the state machine sends the result using the given stream.
*
* @param request the client request
* @param stream the output stream to send the result. The state machine should close the stream at the end.
*/
default void query(Message request, DataChannel stream) {
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #1469:
URL: https://github.com/apache/ratis/pull/1469#issuecomment-4546321760
@szetszwo I integrated the Ratis stream read API into Ozone as a proof of concept, and the results are promising.
The no-MD5 reads comparison:
| Key size | Buffer | ReadBlock stream | Ratis data stream | Ratis / ReadBlock |
| --- | ---: | ---: | ---: | ---: |
| 256 MiB | 32 MiB | 558 MB/s | 107a5 MB/s | 1.93x |
| 256 MiB | 8 MiB | 1449 MB/s | 1492 MB/s | 1.03x |
| 256 MiB | 1 MiB | 387 MB/s | 1737 MB/s | 4.48x |
| 256 MiB | 4 KiB | 723 MB/s | 1653 MB/s | 2.29x |
| 500 MiB | 32 MiB | 1725 MB/s | 1690 MB/s | 0.98x |
| 500 MiB | 8 MiB | 1392 MB/s | 2001 MB/s | 1.44x |
| 500 MiB | 1 MiB | 1080 MB/s | 2029 MB/s | 1.88x |
| 500 MiB | 4 KiB | 838 MB/s | 2052 MB/s | 2.45x |
| 1 GiB | 32 MiB | 1120 MB/s | 1981 MB/s | 1.77x |
| 1 GiB | 8 MiB | 1861 MB/s | 2352 MB/s | 1.26x |
| 1 GiB | 1 MiB | 1280 MB/s | 2357 MB/s | 1.84x |
| 1 GiB | 4 KiB | 933 MB/s | 2227 MB/s | 2.39x |
ozone poc: https://github.com/peterxcli/ozone/pull/16
ratis stream read client poc: https://github.com/peterxcli/ratis/pull/1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo merged PR #1469:
URL: https://github.com/apache/ratis/pull/1469
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #1469:
URL: https://github.com/apache/ratis/pull/1469#issuecomment-4581329506
@szetszwo Thanks for the thorough review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #1469:
URL: https://github.com/apache/ratis/pull/1469#issuecomment-4543081451
@szetszwo thanks for the review, I address the data channel reuse suggestion and only keep server side code in this PR. Please take another look, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3314158984
##########
ratis-server-api/src/main/java/org/apache/ratis/statemachine/StateMachine.java:
##########
@@ -116,6 +116,18 @@ default CompletableFuture<DataStream> stream(RaftClientRequest request) {
return CompletableFuture.completedFuture(null);
}
+ /**
+ * Stream a read-only state machine request. Implementations may write zero or more data
+ * chunks before completing the returned future with the terminal reply message.
+ *
+ * @param request the read-only client request
+ * @param stream the output stream for response data chunks
+ * @return a future for the terminal reply message
+ */
+ default CompletableFuture<Message> streamReadOnly(RaftClientRequest request, DataChannel stream) {
Review Comment:
This method should be similar to query(Message):
```java
/**
* Similar to {@link #query(Message)} except below:
* - In {@link #query(Message)}, the state machine returns the result in a future.
* - In this method, the state machine sends the result using the given stream.
*
* @param request the client request
* @param stream the output stream to send the results
*/
default void query(Message request, DataChannel stream) {
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3299691761
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java:
##########
@@ -510,6 +577,42 @@ private void readImpl(DataStreamRequestByteBuf request, ChannelHandlerContext ct
});
}
+ private static RaftClientRequest toRaftClientRequest(DataStreamRequestByteBuf request) {
+ try {
+ return ClientProtoUtils.toRaftClientRequest(RaftClientRequestProto.parseFrom(request.slice().nioBuffer()));
+ } catch (Throwable e) {
+ throw new CompletionException(e);
+ }
+ }
+
+ private CompletableFuture<Void> submitReadOnlyRequest(DataStreamRequestByteBuf request,
+ RaftClientRequest raftClientRequest, ChannelHandlerContext ctx) {
+ try {
+ final StateMachine.ReadOnlyDataStream readOnlyDataStream = new StateMachine.ReadOnlyDataStream() {
Review Comment:
We should reuse StateMachine.DataChannel which is a Java WritableByteChannel. Then, we can use other Java API such as FileChannel.transferTo (which is a highly efficient, zero-copy data transfer operation).
```java
final StateMachine.DataChannel readOnlyDataStream = new StateMachine.DataChannel() {
private long streamOffset;
private boolean closed = false;
@Override
public synchronized boolean isOpen() {
return !closed;
}
@Override
public synchronized void close() {
closed = true;
}
@Override
public synchronized void force(boolean metadata) throws AlreadyClosedException{
if (!isOpen()) {
throw new AlreadyClosedException("Channel closed at offset " + streamOffset);
}
ctx.flush();
}
@Override
public synchronized int write(ByteBuffer buffer) throws IOException {
if (!isOpen()) {
throw new AlreadyClosedException("Channel closed at offset " + streamOffset);
}
final DataStreamReplyByteBuffer reply = newDataStreamReadOnlyReplyByteBuffer(request, streamOffset, buffer);
final int length = buffer.remaining();
streamOffset += length;
final ChannelFuture future = ctx.write(reply);
try {
future.await();
} catch (InterruptedException e) {
throw new InterruptedIOException();
}
return length;
}
};
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3325852400
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java:
##########
@@ -352,9 +356,36 @@ static DataStreamReplyByteBuffer newDataStreamReplyByteBuffer(DataStreamRequestB
.setDataStreamPacket(request)
.setBuffer(buffer)
.setSuccess(reply.isSuccess())
+ .setCommitInfos(reply.getCommitInfos())
.build();
}
+ static DataStreamReplyByteBuffer newDataStreamReadOnlyReplyByteBuffer(DataStreamRequestByteBuf request,
+ long streamOffset, ByteBuffer buffer) {
+ final ByteBuffer readOnlyBuffer = buffer.asReadOnlyBuffer();
+ return DataStreamReplyByteBuffer.newBuilder()
+ .setClientId(request.getClientId())
+ .setType(Type.STREAM_DATA)
+ .setStreamId(request.getStreamId())
+ .setStreamOffset(streamOffset)
+ .setBuffer(readOnlyBuffer)
+ .setSuccess(true)
+ .setBytesWritten(readOnlyBuffer.remaining())
+ .build();
+ }
+
+ private static CompletableFuture<Void> writeAndFlush(ChannelHandlerContext ctx, DataStreamReply reply) {
+ final CompletableFuture<Void> future = new CompletableFuture<>();
+ ctx.writeAndFlush(reply).addListener(channelFuture -> {
+ if (channelFuture.isSuccess()) {
+ future.complete(null);
+ } else {
+ future.completeExceptionally(channelFuture.cause());
+ }
+ });
+ return future;
+ }
Review Comment:
Let's move all the new code to a new class, say ReadStreamManagement.
##########
dev-support/checkstyle.xml:
##########
@@ -60,6 +60,10 @@
</module>
<module name="SuppressWarningsFilter"/>
+ <module name="SuppressionSingleFilter">
+ <property name="checks" value="FileLength"/>
+ <property name="files" value="RaftServerImpl.java"/>
+ </module>
Review Comment:
Since this PR no longer changes RaftServerImpl. Please revert the checkstyle file.
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java:
##########
@@ -450,6 +481,23 @@ private void readImpl(DataStreamRequestByteBuf request, ChannelHandlerContext ct
// add to ChannelMap
channels.add(channelId, key);
+ if (request.getType() == Type.STREAM_HEADER) {
+ final RaftClientRequest raftClientRequest = toRaftClientRequest(request);
+ if (raftClientRequest.is(TypeCase.READ)) {
+ submitReadOnlyRequest(request, raftClientRequest, ctx).whenComplete((v, exception) -> {
+ try {
+ if (exception != null) {
+ replyDataStreamException(server, exception, raftClientRequest, request, ctx);
+ }
+ } finally {
+ request.release();
+ channels.remove(channelId, key);
Review Comment:
- request can be release earlier -- we only need to clientId and streamId in the read stream.
- channelId is not used. So, we don't need a ChannelMap in ReadStreamManagement.
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java:
##########
@@ -450,6 +481,23 @@ private void readImpl(DataStreamRequestByteBuf request, ChannelHandlerContext ct
// add to ChannelMap
channels.add(channelId, key);
+ if (request.getType() == Type.STREAM_HEADER) {
+ final RaftClientRequest raftClientRequest = toRaftClientRequest(request);
+ if (raftClientRequest.is(TypeCase.READ)) {
+ submitReadOnlyRequest(request, raftClientRequest, ctx).whenComplete((v, exception) -> {
+ try {
+ if (exception != null) {
+ replyDataStreamException(server, exception, raftClientRequest, request, ctx);
+ }
+ } finally {
+ request.release();
+ channels.remove(channelId, key);
+ }
+ });
+ return;
+ }
+ }
Review Comment:
The new read streams have nothing to do with the existing write streams. Let's do the check in NettyServerStreamRpc.
```java
@@ -235,6 +237,9 @@ public class NettyServerStreamRpc implements DataStreamServerRpc {
final DataStreamRequestByteBuf request = (DataStreamRequestByteBuf)msg;
try(UncheckedAutoCloseable autoReset = requestRef.set(request)) {
+ if (reads.process(request, ctx)) {
+ return;
+ }
requests.read(request, ctx, proxies.get(request)::getDataStreamOutput);
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3408064254
##########
ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java:
##########
@@ -150,6 +153,18 @@ default RaftGroup getGroup() {
/** @return the data stream rpc service. */
DataStreamServerRpc getDataStreamServerRpc();
+ /**
+ * Submit a read-only request whose response may be streamed through the data stream RPC.
+ *
+ * @param request the read-only request
+ * @param stream the stream for response data chunks
+ * @return a future for the terminal reply
+ */
+ default CompletableFuture<RaftClientReply> streamReadOnlyAsync(
+ RaftClientRequest request, StateMachine.DataChannel stream) throws IOException {
+ throw new UnsupportedOperationException("This method is NOT supported.");
+ }
Review Comment:
opened a ticket for this: https://issues.apache.org/jira/browse/RATIS-2559
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3314158984
##########
ratis-server-api/src/main/java/org/apache/ratis/statemachine/StateMachine.java:
##########
@@ -116,6 +116,18 @@ default CompletableFuture<DataStream> stream(RaftClientRequest request) {
return CompletableFuture.completedFuture(null);
}
+ /**
+ * Stream a read-only state machine request. Implementations may write zero or more data
+ * chunks before completing the returned future with the terminal reply message.
+ *
+ * @param request the read-only client request
+ * @param stream the output stream for response data chunks
+ * @return a future for the terminal reply message
+ */
+ default CompletableFuture<Message> streamReadOnly(RaftClientRequest request, DataChannel stream) {
Review Comment:
This method should be similar to query(Message):
```java
/**
* Similar to {@link #query(Message)} except below:
* - In {@link #query(Message)}, the state machine returns the result in a future.
* - In this method, the state machine sends the result using the given stream.
*
* @param request the client request
* @param stream the output stream to send the result. The state machine must close the stream at the end.
*/
default void query(Message request, DataChannel stream) {
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations in Server [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3306458278
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java:
##########
@@ -510,6 +558,72 @@ private void readImpl(DataStreamRequestByteBuf request, ChannelHandlerContext ct
});
}
+ private static RaftClientRequest toRaftClientRequest(DataStreamRequestByteBuf request) {
+ try {
+ return ClientProtoUtils.toRaftClientRequest(RaftClientRequestProto.parseFrom(request.slice().nioBuffer()));
+ } catch (Throwable e) {
+ throw new CompletionException(e);
+ }
+ }
+
+ private CompletableFuture<Void> submitReadOnlyRequest(DataStreamRequestByteBuf request,
+ RaftClientRequest raftClientRequest, ChannelHandlerContext ctx) {
+ try {
+ final StateMachine.DataChannel readOnlyDataStream = new StateMachine.DataChannel() {
+ private long streamOffset;
+ private boolean closed;
+
+ @Override
+ public synchronized boolean isOpen() {
+ return !closed;
+ }
+
+ @Override
+ public synchronized void close() {
+ closed = true;
+ }
+
+ @Override
+ public synchronized void force(boolean metadata) throws IOException {
+ if (!isOpen()) {
+ throw new AlreadyClosedException("Channel closed at offset " + streamOffset);
+ }
+ ctx.flush();
+ }
+
+ @Override
+ public synchronized int write(ByteBuffer buffer) throws IOException {
+ if (!isOpen()) {
+ throw new AlreadyClosedException("Channel closed at offset " + streamOffset);
+ }
+ final int length = buffer.remaining();
+ final DataStreamReplyByteBuffer reply = newDataStreamReadOnlyReplyByteBuffer(request, streamOffset, buffer);
+ final ChannelFuture future = ctx.writeAndFlush(reply);
+ try {
+ future.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException(
+ "Interrupted while writing " + length + " bytes at offset " + streamOffset);
+ }
+ if (!future.isSuccess()) {
+ final Throwable cause = future.cause();
+ if (cause instanceof IOException) {
+ throw (IOException) cause;
+ }
Review Comment:
Let's create a new IOException in both cases since it can add the offset to the message.
##########
ratis-server-api/src/main/java/org/apache/ratis/server/RaftServer.java:
##########
@@ -150,6 +153,18 @@ default RaftGroup getGroup() {
/** @return the data stream rpc service. */
DataStreamServerRpc getDataStreamServerRpc();
+ /**
+ * Submit a read-only request whose response may be streamed through the data stream RPC.
+ *
+ * @param request the read-only request
+ * @param stream the stream for response data chunks
+ * @return a future for the terminal reply
+ */
+ default CompletableFuture<RaftClientReply> streamReadOnlyAsync(
+ RaftClientRequest request, StateMachine.DataChannel stream) throws IOException {
+ throw new UnsupportedOperationException("This method is NOT supported.");
+ }
Review Comment:
This new method seems not needed since we may:
- Phase 1: Directly call DataApi.streamReadOnly(..) and ignore all linearizable checks.
- Phase 2: Reuse RaftClientAsynchronousProtocol.submitClientRequestAsync(..) to submit a dummy read request for linearizable checks and then call DataApi.streamReadOnly(..).
Of course, we should start with Phase 1 for simpilcity.
##########
ratis-server-api/src/main/java/org/apache/ratis/statemachine/StateMachine.java:
##########
@@ -116,6 +116,18 @@ default CompletableFuture<DataStream> stream(RaftClientRequest request) {
return CompletableFuture.completedFuture(null);
}
+ /**
+ * Stream a read-only state machine request. Implementations may write zero or more data
+ * chunks before completing the returned future with the terminal reply message.
+ *
+ * @param request the read-only client request
+ * @param stream the output stream for response data chunks
+ * @return a future for the terminal reply message
+ */
+ default CompletableFuture<Message> streamReadOnly(RaftClientRequest request, DataChannel stream) {
Review Comment:
This method should be similar to query(Message):
```java
/**
* Similar to {@link #query(Message)} except that
* {@link #query(Message)} returns the result in a future
* while this method sends the result using the given stream.
*
* @param request the client request
* @param stream the output stream to send the results
*/
default void query(Message request, DataChannel stream) {
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] RATIS-1240. Add input stream to DataStreamApi for read operations [ratis]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #1469:
URL: https://github.com/apache/ratis/pull/1469#discussion_r3299691761
##########
ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java:
##########
@@ -510,6 +577,42 @@ private void readImpl(DataStreamRequestByteBuf request, ChannelHandlerContext ct
});
}
+ private static RaftClientRequest toRaftClientRequest(DataStreamRequestByteBuf request) {
+ try {
+ return ClientProtoUtils.toRaftClientRequest(RaftClientRequestProto.parseFrom(request.slice().nioBuffer()));
+ } catch (Throwable e) {
+ throw new CompletionException(e);
+ }
+ }
+
+ private CompletableFuture<Void> submitReadOnlyRequest(DataStreamRequestByteBuf request,
+ RaftClientRequest raftClientRequest, ChannelHandlerContext ctx) {
+ try {
+ final StateMachine.ReadOnlyDataStream readOnlyDataStream = new StateMachine.ReadOnlyDataStream() {
Review Comment:
We should reuse StateMachine.DataChannel which is a Java WritableByteChannel. Then, we can use other Java API such as FileChannel.transferTo (which is a highly efficient, zero-copy data transfer operation).
```java
final StateMachine.DataChannel readOnlyDataStream = new StateMachine.DataChannel() {
private long streamOffset;
private boolean closed = false;
@Override
public synchronized boolean isOpen() {
return !closed;
}
@Override
public synchronized void close() {
closed = true;
}
@Override
public synchronized void force(boolean metadata) throws AlreadyClosedException{
if (!isOpen()) {
throw new AlreadyClosedException("Channel closed at offset " + streamOffset);
}
ctx.flush();
}
@Override
public synchronized int write(ByteBuffer buffer) throws IOException {
if (!isOpen()) {
throw new AlreadyClosedException("Channel closed at offset " + streamOffset);
}
final DataStreamReplyByteBuffer reply = newDataStreamReadOnlyReplyByteBuffer(request, streamOffset, buffer);
final int length = buffer.remaining();
final ChannelFuture future = ctx.write(reply);
try {
future.await();
} catch (InterruptedException e) {
throw new InterruptedIOException("Interrupted while writing " + length
+ " bytes at offset " + streamOffset);
}
streamOffset += length;
return length;
}
};
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ratis.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org