szetszwo commented on code in PR #1481: URL: https://github.com/apache/ratis/pull/1481#discussion_r3455864706
########## ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamInputImpl.java: ########## @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.ratis.client.impl; + +import org.apache.ratis.client.DataStreamClientRpc; +import org.apache.ratis.client.api.DataStreamInput; +import org.apache.ratis.datastream.DataStreamObserver; +import org.apache.ratis.datastream.impl.DataStreamRequestByteBuffer; +import org.apache.ratis.io.StandardWriteOption; +import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type; +import org.apache.ratis.protocol.ClientId; +import org.apache.ratis.protocol.DataStreamReply; +import org.apache.ratis.protocol.DataStreamRequestHeader; +import org.apache.ratis.protocol.RaftClientRequest; +import org.apache.ratis.protocol.exceptions.AlreadyClosedException; +import org.apache.ratis.util.JavaUtils; +import org.apache.ratis.util.ReferenceCountedObject; + +import java.io.EOFException; +import java.nio.ByteBuffer; +import java.util.LinkedList; +import java.util.Objects; +import java.util.Queue; +import java.util.concurrent.CompletableFuture; + +final class DataStreamInputImpl implements DataStreamInput, + DataStreamObserver<ReferenceCountedObject<DataStreamReply>> { + private final RaftClientRequest header; + private final ClientId clientId; + private final Queue<ReferenceCountedObject<DataStreamReply>> replies = new LinkedList<>(); + private final Queue<CompletableFuture<ReferenceCountedObject<DataStreamReply>>> pendingReads = new LinkedList<>(); + + /* + * null : the stream is open. + * AlreadyClosedException: the stream is closed. + * Other exception : the stream is failed. + */ + private Throwable readException; + + DataStreamInputImpl(DataStreamClientRpc dataStreamClientRpc, RaftClientRequest request) { + this.header = request; + this.clientId = request.getClientId(); + final ByteBuffer buffer = ClientProtoUtils.toRaftClientRequestProtoByteBuffer(header); + final DataStreamRequestHeader h = new DataStreamRequestHeader(clientId, Type.STREAM_HEADER, + header.getCallId(), 0, buffer.remaining(), StandardWriteOption.FLUSH, StandardWriteOption.CLOSE); + dataStreamClientRpc.streamAsync(new DataStreamRequestByteBuffer(h, buffer), this) + .whenComplete(asWhenCompleteBiConsumer()); + } + + @Override + public synchronized void onNext(ReferenceCountedObject<DataStreamReply> reply) { + if (readException != null) { + return; + } + + reply.retain(); + for (CompletableFuture<ReferenceCountedObject<DataStreamReply>> pending; + (pending = pendingReads.poll()) != null; ) { + if (pending.complete(reply)) { + return; + } + } + replies.add(reply); Review Comment: Since all entries in pendingReads must not be completed, remove the loop. ```java reply.retain(); final CompletableFuture<ReferenceCountedObject<DataStreamReply>> pending = pendingReads.poll(); if (pending != null) { final boolean completed = pending.complete(reply); Preconditions.assertTrue(completed); return; } replies.add(reply); ``` ########## ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamInputImpl.java: ########## @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.ratis.client.impl; + +import org.apache.ratis.client.DataStreamClientRpc; +import org.apache.ratis.client.api.DataStreamInput; +import org.apache.ratis.datastream.DataStreamObserver; +import org.apache.ratis.datastream.impl.DataStreamRequestByteBuffer; +import org.apache.ratis.io.StandardWriteOption; +import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type; +import org.apache.ratis.protocol.ClientId; +import org.apache.ratis.protocol.DataStreamReply; +import org.apache.ratis.protocol.DataStreamRequestHeader; +import org.apache.ratis.protocol.RaftClientRequest; +import org.apache.ratis.protocol.exceptions.AlreadyClosedException; +import org.apache.ratis.util.JavaUtils; +import org.apache.ratis.util.ReferenceCountedObject; + +import java.io.EOFException; +import java.nio.ByteBuffer; +import java.util.LinkedList; +import java.util.Objects; +import java.util.Queue; +import java.util.concurrent.CompletableFuture; + +final class DataStreamInputImpl implements DataStreamInput, + DataStreamObserver<ReferenceCountedObject<DataStreamReply>> { + private final RaftClientRequest header; + private final ClientId clientId; + private final Queue<ReferenceCountedObject<DataStreamReply>> replies = new LinkedList<>(); + private final Queue<CompletableFuture<ReferenceCountedObject<DataStreamReply>>> pendingReads = new LinkedList<>(); + + /* + * null : the stream is open. + * AlreadyClosedException: the stream is closed. + * Other exception : the stream is failed. + */ + private Throwable readException; + + DataStreamInputImpl(DataStreamClientRpc dataStreamClientRpc, RaftClientRequest request) { + this.header = request; + this.clientId = request.getClientId(); + final ByteBuffer buffer = ClientProtoUtils.toRaftClientRequestProtoByteBuffer(header); + final DataStreamRequestHeader h = new DataStreamRequestHeader(clientId, Type.STREAM_HEADER, + header.getCallId(), 0, buffer.remaining(), StandardWriteOption.FLUSH, StandardWriteOption.CLOSE); + dataStreamClientRpc.streamAsync(new DataStreamRequestByteBuffer(h, buffer), this) + .whenComplete(asWhenCompleteBiConsumer()); + } + + @Override + public synchronized void onNext(ReferenceCountedObject<DataStreamReply> reply) { + if (readException != null) { + return; + } + + reply.retain(); + for (CompletableFuture<ReferenceCountedObject<DataStreamReply>> pending; + (pending = pendingReads.poll()) != null; ) { + if (pending.complete(reply)) { + return; + } + } + replies.add(reply); + } + + @Override + public synchronized void onError(Throwable throwable) { + // An error case, release the replies Review Comment: Add requireNonNull: ``` Objects.requireNonNull(throwable, "throwable == null"); ``` ########## ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java: ########## @@ -358,15 +364,40 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) { LOG.error("{}: unexpected message {}", name, msg.getClass()); return; } - try (DataStreamReplyByteBuf reply = (DataStreamReplyByteBuf) msg) { - process(reply); + 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(); Review Comment: Let's change Preconditions.assertSame to return DataStreamReplyByteBuf: ```java final ReferenceCountedObject<DataStreamReply> ref = ReferenceCountedObject.<DataStreamReply>newBuilder() .setValue((DataStreamReplyByteBuf) msg) .setReleaseMethod(r -> Preconditions.assertSame(reply, r, "reply").release()) .build(); ``` ```diff +++ b/ratis-common/src/main/java/org/apache/ratis/util/Preconditions.java @@ -88,9 +88,10 @@ public interface Preconditions { () -> name + ": expected == " + expected + " but computed == " + computed); } - static void assertSame(Object expected, Object computed, String name) { + static <T> T assertSame(T expected, Object computed, String name) { assertTrue(expected == computed, () -> name + ": expected == " + expected + " but computed == " + computed); + return expected; } ``` ########## ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamInputImpl.java: ########## @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.ratis.client.impl; + +import org.apache.ratis.client.DataStreamClientRpc; +import org.apache.ratis.client.api.DataStreamInput; +import org.apache.ratis.datastream.DataStreamObserver; +import org.apache.ratis.datastream.impl.DataStreamRequestByteBuffer; +import org.apache.ratis.io.StandardWriteOption; +import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type; +import org.apache.ratis.protocol.ClientId; +import org.apache.ratis.protocol.DataStreamReply; +import org.apache.ratis.protocol.DataStreamRequestHeader; +import org.apache.ratis.protocol.RaftClientRequest; +import org.apache.ratis.protocol.exceptions.AlreadyClosedException; +import org.apache.ratis.util.JavaUtils; +import org.apache.ratis.util.ReferenceCountedObject; + +import java.io.EOFException; +import java.nio.ByteBuffer; +import java.util.LinkedList; +import java.util.Objects; +import java.util.Queue; +import java.util.concurrent.CompletableFuture; + +final class DataStreamInputImpl implements DataStreamInput, + DataStreamObserver<ReferenceCountedObject<DataStreamReply>> { + private final RaftClientRequest header; + private final ClientId clientId; + private final Queue<ReferenceCountedObject<DataStreamReply>> replies = new LinkedList<>(); + private final Queue<CompletableFuture<ReferenceCountedObject<DataStreamReply>>> pendingReads = new LinkedList<>(); + + /* + * null : the stream is open. + * AlreadyClosedException: the stream is closed. + * Other exception : the stream is failed. Review Comment: Add EOFException: ```java * null : the stream is open. * EOFException : the stream is ended, i.e. no more data. * AlreadyClosedException: the stream is closed by the caller. * Other exception : the stream is failed. ``` ########## ratis-common/src/main/java/org/apache/ratis/protocol/DataStreamReply.java: ########## @@ -30,4 +30,4 @@ public interface DataStreamReply extends DataStreamPacket { /** @return the commit information when the reply is created. */ Collection<CommitInfoProto> getCommitInfos(); -} \ No newline at end of file +} Review Comment: Revert white space change. ########## ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java: ########## @@ -64,20 +64,19 @@ static class ReadStream implements WritableByteChannel { this.clientId = request.getClientId(); this.streamId = streamId; this.ctx = ctx; - final RaftClientReply reply = RaftClientReply.newBuilder() - .setRequest(request) - .setSuccess() - .build(); + .setRequest(request) + .setSuccess() + .build(); this.terminalReply = DataStreamReplyByteBuffer.newBuilder() - .setClientId(clientId) - .setType(Type.STREAM_HEADER) - .setStreamId(streamId) - .setStreamOffset(0) - .setBuffer(toRaftClientReplyProto(reply).toByteString().asReadOnlyByteBuffer()) - .setSuccess(true) - .setBytesWritten(0) - .build(); Review Comment: All the changes in ReadStreamManagement are whitespace change. Let's revert them since this change is already quite big. ########## ratis-test/src/test/java/org/apache/ratis/client/impl/TestDataStreamClientImpl.java: ########## @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.ratis.client.impl; + +import org.apache.ratis.client.DataStreamClient; +import org.apache.ratis.client.DataStreamClientRpc; +import org.apache.ratis.client.api.DataStreamInput; +import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.datastream.DataStreamObserver; +import org.apache.ratis.datastream.impl.DataStreamReplyByteBuf; +import org.apache.ratis.datastream.impl.DataStreamRequestByteBuffer; +import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type; +import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto; +import org.apache.ratis.protocol.ClientId; +import org.apache.ratis.protocol.DataStreamReply; +import org.apache.ratis.protocol.DataStreamRequest; +import org.apache.ratis.protocol.RaftClientRequest; +import org.apache.ratis.protocol.RaftGroupId; +import org.apache.ratis.protocol.RaftPeer; +import org.apache.ratis.thirdparty.io.netty.buffer.Unpooled; +import org.apache.ratis.util.ReferenceCountedObject; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.io.EOFException; +import java.nio.ByteBuffer; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicReference; + +public class TestDataStreamClientImpl { + private static RaftPeer newPeer(String id) { + return RaftPeer.newBuilder().setId(id).build(); + } + + private static class RecordingDataStreamClientRpc implements DataStreamClientRpc { + private final AtomicReference<RaftClientRequest> request = new AtomicReference<>(); + private final AtomicReference<DataStreamObserver<ReferenceCountedObject<DataStreamReply>>> replyHandler = new AtomicReference<>(); + private final AtomicReference<CompletableFuture<DataStreamReply>> replyFuture = new AtomicReference<>(); + + @Override + public CompletableFuture<DataStreamReply> streamAsync( + DataStreamRequest dataStreamRequest, + DataStreamObserver<ReferenceCountedObject<DataStreamReply>> replyHandler) { + try { + final ByteBuffer buffer = ((DataStreamRequestByteBuffer) dataStreamRequest).slice(); + request.set(ClientProtoUtils.toRaftClientRequest(RaftClientRequestProto.parseFrom(buffer))); + } catch (Exception e) { + throw new IllegalStateException(e); + } + this.replyHandler.set(replyHandler); + final CompletableFuture<DataStreamReply> future = new CompletableFuture<>(); + replyFuture.set(future); + return future; + } + + RaftClientRequest getRequest() { + return request.get(); + } + + void receive(DataStreamReplyByteBuf reply) { + final ReferenceCountedObject<DataStreamReply> ref = ReferenceCountedObject.<DataStreamReply>newBuilder() + .setValue(reply) + .setReleaseMethod(DataStreamReplyByteBuf::release) Review Comment: Use assertSame: ```java .setReleaseMethod(r -> Preconditions.assertSame(reply, r, "reply").release()) ``` ########## ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java: ########## @@ -513,6 +544,54 @@ public CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest request) return f; } + @Override + public CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest request, + DataStreamObserver<ReferenceCountedObject<DataStreamReply>> replyHandler) { + final CompletableFuture<DataStreamReply> f = new CompletableFuture<>(); + final ClientInvocationId clientInvocationId = ClientInvocationId.valueOf(request); + final ClientReadStream replyEntry = new ClientReadStream(request, f, replyHandler); + if (readStreams.putIfAbsent(clientInvocationId, replyEntry) != null) { + f.completeExceptionally(new AlreadyClosedException(this + ": A read-only stream already exists for " + + clientInvocationId)); + return f; + } + + final ChannelFuture channelFuture; + final Channel channel; + LOG.debug("{}: write read-only stream begin {}", this, request); + synchronized (replyEntry) { + channel = connection.getChannelUninterruptibly(); + if (channel == null) { + readStreams.remove(clientInvocationId, replyEntry); + f.completeExceptionally(new AlreadyClosedException(this + ": Failed to send " + request)); Review Comment: Let's change the message to "Failed to getChannel": ```java f.completeExceptionally(new AlreadyClosedException(this + ": Failed to getChannel for " + request)); ``` ########## ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java: ########## @@ -513,6 +544,54 @@ public CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest request) return f; } + @Override + public CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest request, + DataStreamObserver<ReferenceCountedObject<DataStreamReply>> replyHandler) { + final CompletableFuture<DataStreamReply> f = new CompletableFuture<>(); + final ClientInvocationId clientInvocationId = ClientInvocationId.valueOf(request); + final ClientReadStream replyEntry = new ClientReadStream(request, f, replyHandler); Review Comment: Let's rename it to `readStream`. ########## ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientStreamRpc.java: ########## @@ -513,6 +544,54 @@ public CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest request) return f; } + @Override + public CompletableFuture<DataStreamReply> streamAsync(DataStreamRequest request, + DataStreamObserver<ReferenceCountedObject<DataStreamReply>> replyHandler) { + final CompletableFuture<DataStreamReply> f = new CompletableFuture<>(); + final ClientInvocationId clientInvocationId = ClientInvocationId.valueOf(request); + final ClientReadStream replyEntry = new ClientReadStream(request, f, replyHandler); + if (readStreams.putIfAbsent(clientInvocationId, replyEntry) != null) { + f.completeExceptionally(new AlreadyClosedException(this + ": A read-only stream already exists for " + + clientInvocationId)); Review Comment: Let's include the request: ```java f.completeExceptionally(new AlreadyClosedException(this + ": Read stream already exists for " + clientInvocationId + ", request=" + request)); ``` ########## ratis-common/src/main/java/org/apache/ratis/datastream/impl/DataStreamReplyByteBuf.java: ########## @@ -118,4 +118,10 @@ static ByteBuffer copy(ByteBuf buf) { buf.readBytes(bytes); return ByteBuffer.wrap(bytes); } + + public static void release(DataStreamReply reply) { + if (reply instanceof DataStreamReplyByteBuf) { + ((DataStreamReplyByteBuf) reply).release(); + } + } Review Comment: Just found that this is used only in a test. Let remove it. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
