You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ratis.apache.org by GitBox <gi...@apache.org> on 2020/12/09 03:27:27 UTC
[GitHub] [incubator-ratis] runzhiwang opened a new pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
runzhiwang opened a new pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339
## What changes were proposed in this pull request?
For the CLOSE, we can add a new API, writeAndClose(..) to save STREAM_CLOSE RPC call.
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/RATIS-1196
## How was this patch tested?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] szetszwo commented on a change in pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
szetszwo commented on a change in pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339#discussion_r539137277
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java
##########
@@ -28,7 +28,7 @@
import java.util.concurrent.CompletableFuture;
/** An asynchronous output stream supporting zero buffer copying. */
-public interface DataStreamOutput extends CloseAsync<DataStreamReply> {
Review comment:
> If so, user maybe confused between writeAndCloseAsync and closeAsync. ...
Since writeAndCloseAsync is uncommon, I think more users will be confused with 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] runzhiwang closed pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
runzhiwang closed pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] szetszwo commented on a change in pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
szetszwo commented on a change in pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339#discussion_r544849349
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
##########
@@ -92,7 +94,11 @@ public boolean isOpen() {
@Override
public void close() throws IOException {
- IOUtils.getFromFuture(closeAsync(), () -> "close(" + ClientInvocationId.valueOf(header) + ")");
+ if (isClosed()) {
+ return;
+ }
+ IOUtils.getFromFuture(writeAsync(Unpooled.EMPTY_BUFFER.nioBuffer(), StandardWriteOption.CLOSE),
Review comment:
Use DataStreamPacketByteBuffer.EMPTY_BYTE_BUFFER instead of Unpooled.EMPTY_BUFFER. We should avoid importing Netty in ratis-client.
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
##########
@@ -104,46 +110,52 @@ private DataStreamOutputImpl(RaftClientRequest request) {
this.headerFuture = send(Type.STREAM_HEADER, buffer, buffer.remaining());
}
- private CompletableFuture<DataStreamReply> send(Type type, Object data, long length) {
- final DataStreamRequestHeader h = new DataStreamRequestHeader(type, header.getCallId(), streamOffset, length);
+ private CompletableFuture<DataStreamReply> send(Type type, Object data, long length, WriteOption... options) {
+ final DataStreamRequestHeader h =
+ new DataStreamRequestHeader(type, header.getCallId(), streamOffset, length, options);
return orderedStreamAsync.sendRequest(h, data);
}
- private CompletableFuture<DataStreamReply> send(Type type) {
- return combineHeader(send(type, null, 0));
- }
-
private CompletableFuture<DataStreamReply> combineHeader(CompletableFuture<DataStreamReply> future) {
return future.thenCombine(headerFuture, (reply, headerReply) -> headerReply.isSuccess()? reply : headerReply);
}
- private CompletableFuture<DataStreamReply> writeAsyncImpl(Object data, long length, boolean sync) {
+ private CompletableFuture<DataStreamReply> writeAsyncImpl(Object data, long length, WriteOption... options) {
if (isClosed()) {
return JavaUtils.completeExceptionally(new AlreadyClosedException(
clientId + ": stream already closed, request=" + header));
}
- final CompletableFuture<DataStreamReply> f = send(sync ? Type.STREAM_DATA_SYNC : Type.STREAM_DATA, data, length);
+ final CompletableFuture<DataStreamReply> f = combineHeader(send(Type.STREAM_DATA, data, length, options));
+ if (WriteOption.containsOption(options, StandardWriteOption.CLOSE)) {
+ closeFuture.set(f);
+ closed.set(true);
+ f.thenApply(ClientProtoUtils::getRaftClientReply).whenComplete(JavaUtils.asBiConsumer(raftClientReplyFuture));
+ }
streamOffset += length;
- return combineHeader(f);
+ return f;
}
@Override
- public CompletableFuture<DataStreamReply> writeAsync(ByteBuffer src, boolean sync) {
- return writeAsyncImpl(src, src.remaining(), sync);
+ public CompletableFuture<DataStreamReply> writeAsync(ByteBuffer src, WriteOption... options) {
+ return writeAsyncImpl(src, src.remaining(), options);
}
@Override
- public CompletableFuture<DataStreamReply> writeAsync(FilePositionCount src, boolean sync) {
- return writeAsyncImpl(src, src.getCount(), sync);
+ public CompletableFuture<DataStreamReply> writeAsync(FilePositionCount src, WriteOption... options) {
+ return writeAsyncImpl(src, src.getCount(), options);
+ }
+
+ boolean isClosed() {
+ return closed.get();
Review comment:
We may use closeFuture as below and remove `closed`.
```
return closeFuture.get() != null;
```
##########
File path: ratis-common/src/main/java/org/apache/ratis/io/StandardWriteOption.java
##########
@@ -0,0 +1,27 @@
+/*
+ * 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.io;
+
+public enum StandardWriteOption implements WriteOption {
+ /** Write the data to the underlying storage. */
+ WRITE,
Review comment:
WRITE is never used in server. Let's remove it.
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
##########
@@ -69,11 +70,6 @@
private final RaftClientRequest header;
private final CompletableFuture<DataStreamReply> headerFuture;
private final CompletableFuture<RaftClientReply> raftClientReplyFuture = new CompletableFuture<>();
- private final MemoizedSupplier<CompletableFuture<DataStreamReply>> closeSupplier = JavaUtils.memoize(() -> {
Review comment:
You are right that we don't need AtomicBoolean here. Just a boolean is enough.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] runzhiwang commented on a change in pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on a change in pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339#discussion_r539140517
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java
##########
@@ -28,7 +28,7 @@
import java.util.concurrent.CompletableFuture;
/** An asynchronous output stream supporting zero buffer copying. */
-public interface DataStreamOutput extends CloseAsync<DataStreamReply> {
Review comment:
@szetszwo I see, 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] runzhiwang commented on a change in pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on a change in pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339#discussion_r539131795
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java
##########
@@ -28,7 +28,7 @@
import java.util.concurrent.CompletableFuture;
/** An asynchronous output stream supporting zero buffer copying. */
-public interface DataStreamOutput extends CloseAsync<DataStreamReply> {
Review comment:
@szetszwo Sorry, what should we do in closeAsync() ? Do we need to send rpc call to primary ? If so, user maybe confused between writeAndCloseAsync and closeAsync. I delete `extends CloseAsync` so that user can only close by writeAndCloseAsync.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] szetszwo merged pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
szetszwo merged pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] runzhiwang closed pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
runzhiwang closed pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] runzhiwang commented on pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339#issuecomment-741502577
@szetszwo Could you have a brief review ? If you agree with the idea, I will fix the failed ut and compiled error of ut.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] runzhiwang closed pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
runzhiwang closed pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] runzhiwang commented on a change in pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on a change in pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339#discussion_r544738142
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
##########
@@ -69,11 +70,6 @@
private final RaftClientRequest header;
private final CompletableFuture<DataStreamReply> headerFuture;
private final CompletableFuture<RaftClientReply> raftClientReplyFuture = new CompletableFuture<>();
- private final MemoizedSupplier<CompletableFuture<DataStreamReply>> closeSupplier = JavaUtils.memoize(() -> {
Review comment:
@szetszwo Hi, I have updated the patch, but confused why we need Atomic, I think will not happen multi-thread close the same 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] runzhiwang closed pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
runzhiwang closed pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] szetszwo commented on a change in pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
szetszwo commented on a change in pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339#discussion_r539135309
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java
##########
@@ -28,7 +28,7 @@
import java.util.concurrent.CompletableFuture;
/** An asynchronous output stream supporting zero buffer copying. */
-public interface DataStreamOutput extends CloseAsync<DataStreamReply> {
Review comment:
If it is already closed, do nothing. Otherwise, call writeAsync(DataStreamPacketByteBuffer.EMPTY_BYTE_BUFFER, WriteOption.CLOSE).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-ratis] szetszwo commented on a change in pull request #339: RATIS-1196. Save STREAM_CLOSE RPC call
Posted by GitBox <gi...@apache.org>.
szetszwo commented on a change in pull request #339:
URL: https://github.com/apache/incubator-ratis/pull/339#discussion_r538983256
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java
##########
@@ -71,6 +71,14 @@
*/
CompletableFuture<DataStreamReply> writeAsync(FilePositionCount src, boolean sync);
+ /**
+ * Send out the data in the source file and close stream asynchronously.
+ *
+ * @param src the source file with the starting position and the number of bytes.
+ * @return a future of the reply.
+ */
+ CompletableFuture<DataStreamReply> writeAndCloseAsync(ByteBuffer src);
Review comment:
Similar to https://docs.oracle.com/javase/8/docs/api/java/nio/file/Files.html#newInputStream-java.nio.file.Path-java.nio.file.OpenOption...- , we may define a WriteOption to include both SYNC and CLOSE
```
diff --git a/ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java b/ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java
index c054d356..7bd0db98 100644
--- a/ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java
+++ b/ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java
@@ -41,10 +42,10 @@ public interface DataStreamOutput extends CloseAsync<DataStreamReply> {
* Send out the data in the source buffer asynchronously.
*
* @param src the source buffer to be sent.
- * @param sync Should the data be sync'ed to the underlying storage?
+ * @param options
* @return a future of the reply.
*/
- CompletableFuture<DataStreamReply> writeAsync(ByteBuffer src, boolean sync);
+ CompletableFuture<DataStreamReply> writeAsync(ByteBuffer src, WriteOption... options);
/**
diff --git a/ratis-common/src/main/java/org/apache/ratis/io/StandardWriteOption.java b/ratis-common/src/main/java/org/apache/ratis/io/StandardWriteOption.java
new file mode 100644
index 00000000..5e5bc08b
--- /dev/null
+++ b/ratis-common/src/main/java/org/apache/ratis/io/StandardWriteOption.java
@@ -0,0 +1,8 @@
+package org.apache.ratis.io;
+
+public enum StandardWriteOption implements WriteOption {
+ /** Sync the data to the underlying storage. */
+ SYNC,
+ /** Close the data to the underlying storage? */
+ CLOSE
+}
diff --git a/ratis-common/src/main/java/org/apache/ratis/io/WriteOption.java b/ratis-common/src/main/java/org/apache/ratis/io/WriteOption.java
new file mode 100644
index 00000000..9a28d709
--- /dev/null
+++ b/ratis-common/src/main/java/org/apache/ratis/io/WriteOption.java
@@ -0,0 +1,4 @@
+package org.apache.ratis.io;
+
+public interface WriteOption {
+}
```
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/api/DataStreamOutput.java
##########
@@ -28,7 +28,7 @@
import java.util.concurrent.CompletableFuture;
/** An asynchronous output stream supporting zero buffer copying. */
-public interface DataStreamOutput extends CloseAsync<DataStreamReply> {
Review comment:
I think it is convenient to keep extending CloseAsync<DataStreamReply> so that the API is easier to be used. We may provide a default closeAsync() implementation.
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
##########
@@ -69,11 +70,6 @@
private final RaftClientRequest header;
private final CompletableFuture<DataStreamReply> headerFuture;
private final CompletableFuture<RaftClientReply> raftClientReplyFuture = new CompletableFuture<>();
- private final MemoizedSupplier<CompletableFuture<DataStreamReply>> closeSupplier = JavaUtils.memoize(() -> {
Review comment:
We need some way to make the close idempotent. May be a AtomicBoolean.
##########
File path: ratis-proto/src/main/proto/Raft.proto
##########
@@ -301,7 +301,7 @@ message DataStreamPacketHeaderProto {
STREAM_HEADER = 0;
STREAM_DATA = 1;
STREAM_DATA_SYNC = 2;
- STREAM_CLOSE = 3;
+ STREAM_DATA_SYNC_CLOSE = 3;
Review comment:
How about adding an enum?
```
+++ b/ratis-proto/src/main/proto/Raft.proto
@@ -300,14 +300,18 @@ message DataStreamPacketHeaderProto {
enum Type {
STREAM_HEADER = 0;
STREAM_DATA = 1;
- STREAM_DATA_SYNC = 2;
- STREAM_CLOSE = 3;
+ }
+
+ enum Option {
+ SYNC = 0;
+ CLOSE = 1;
}
uint64 streamId = 1;
uint64 streamOffset = 2;
Type type = 3;
- uint64 dataLength = 4;
+ repeated Option options = 4;
+ uint64 dataLength = 5;
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org