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/10/14 15:25:30 UTC
[GitHub] [incubator-ratis] szetszwo opened a new pull request #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
szetszwo opened a new pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224
See https://issues.apache.org/jira/browse/RATIS-1097
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#issuecomment-708908767
@szetszwo The failed `testAsyncConfiguration` seems related to this PR.
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505097963
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java
##########
@@ -94,42 +74,28 @@ public void fail(Throwable e) {
public CompletableFuture<DataStreamReply> getReplyFuture(){
return replyFuture;
}
-
- public DataStreamWindowRequest(long streamId, long messageId,
- long seqNum, ByteBuffer data){
- this.streamId = streamId;
- this.messageId = messageId;
- this.data = data.slice();
- this.seqNum = seqNum;
- }
}
- private SlidingWindow.Client<DataStreamWindowRequest, DataStreamReply> slidingWindow;
- private Semaphore requestSemaphore;
- private DataStreamClientRpc dataStreamClientRpc;
+ private final DataStreamClientRpc dataStreamClientRpc;
+ private final SlidingWindow.Client<DataStreamWindowRequest, DataStreamReply> slidingWindow;
+ private final Semaphore requestSemaphore;
- public OrderedStreamAsync(DataStreamClientRpc dataStreamClientRpc,
- RaftProperties properties){
+ OrderedStreamAsync(ClientId clientId, DataStreamClientRpc dataStreamClientRpc, RaftProperties properties){
this.dataStreamClientRpc = dataStreamClientRpc;
+ this.slidingWindow = new SlidingWindow.Client<>(clientId);
this.requestSemaphore = new Semaphore(RaftClientConfigKeys.Async.outstandingRequestsMax(properties)*2);
Review comment:
Sorry, one question unrelated to this patch: why multiply 2 here ?
----------------------------------------------------------------
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 pull request #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
szetszwo commented on pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#issuecomment-708840528
@runzhiwang , thanks for reviewing the change. I just have addressed your comments. Could you review it again?
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505110006
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java
##########
@@ -38,37 +37,18 @@
public class OrderedStreamAsync {
public static final Logger LOG = LoggerFactory.getLogger(OrderedStreamAsync.class);
- public static class DataStreamWindowRequest implements DataStreamRequest,
- SlidingWindow.ClientSideRequest<DataStreamReply> {
- private final long streamId;
- private final long messageId;
+ static class DataStreamWindowRequest extends DataStreamRequestByteBuffer
+ implements SlidingWindow.ClientSideRequest<DataStreamReply> {
private final long seqNum;
- private final ByteBuffer data;
- private boolean isFirst = false;
- private CompletableFuture<DataStreamReply> replyFuture = new CompletableFuture<>();
+ private final CompletableFuture<DataStreamReply> replyFuture = new CompletableFuture<>();
- public DataStreamRequestByteBuffer newDataStreamRequest(){
- return new DataStreamRequestByteBuffer(streamId, messageId, data.slice());
- }
-
- @Override
- public long getStreamId() {
- return streamId;
- }
-
- @Override
- public long getDataOffset() {
- return messageId;
- }
-
- @Override
- public long getDataLength() {
- return data.capacity();
+ DataStreamWindowRequest(long streamId, long offset, ByteBuffer data, long seqNum){
+ super(streamId, offset, data);
+ this.seqNum = seqNum;
}
@Override
public void setFirstRequest() {
- isFirst = true;
Review comment:
@szetszwo I see, can we remove [request.setFirstRequest()](https://github.com/apache/incubator-ratis/blob/76a52efc4c6973d254d189ff81e5665ffea09120/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java#L109) in this file ?
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505097963
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java
##########
@@ -94,42 +74,28 @@ public void fail(Throwable e) {
public CompletableFuture<DataStreamReply> getReplyFuture(){
return replyFuture;
}
-
- public DataStreamWindowRequest(long streamId, long messageId,
- long seqNum, ByteBuffer data){
- this.streamId = streamId;
- this.messageId = messageId;
- this.data = data.slice();
- this.seqNum = seqNum;
- }
}
- private SlidingWindow.Client<DataStreamWindowRequest, DataStreamReply> slidingWindow;
- private Semaphore requestSemaphore;
- private DataStreamClientRpc dataStreamClientRpc;
+ private final DataStreamClientRpc dataStreamClientRpc;
+ private final SlidingWindow.Client<DataStreamWindowRequest, DataStreamReply> slidingWindow;
+ private final Semaphore requestSemaphore;
- public OrderedStreamAsync(DataStreamClientRpc dataStreamClientRpc,
- RaftProperties properties){
+ OrderedStreamAsync(ClientId clientId, DataStreamClientRpc dataStreamClientRpc, RaftProperties properties){
this.dataStreamClientRpc = dataStreamClientRpc;
+ this.slidingWindow = new SlidingWindow.Client<>(clientId);
this.requestSemaphore = new Semaphore(RaftClientConfigKeys.Async.outstandingRequestsMax(properties)*2);
Review comment:
Sorry, one question unrelated to this patch: why multiple 2 here ?
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
szetszwo commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505109763
##########
File path: ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyServerStreamRpc.java
##########
@@ -79,31 +78,34 @@ public NettyServerStreamRpc(RaftPeer server, StateMachine stateMachine) {
}
}
- private void writeTo(ByteBuf buf, DataStream stream, boolean released) {
+ private long writeTo(ByteBuf buf, DataStream stream, boolean released) {
if (released) {
- return;
+ return 0;
}
try {
if (stream == null) {
- return;
+ return 0;
}
final WritableByteChannel channel = stream.getWritableByteChannel();
+ long written = 0;
for (ByteBuffer buffer : buf.nioBuffers()) {
try {
- channel.write(buffer);
+ written += channel.write(buffer);
} catch (Throwable t) {
throw new CompletionException(t);
}
}
+ return written;
} finally {
buf.release();
}
}
- private void sendReply(DataStreamRequestByteBuf request, ChannelHandlerContext ctx) {
- final DataStreamReply reply = new DataStreamReplyByteBuffer(
- request.getStreamId(), request.getDataOffset(), ByteBuffer.wrap("OK".getBytes()));
+ private void sendReply(DataStreamRequestByteBuf request, long written, ChannelHandlerContext ctx) {
+ final boolean success = request.getDataLength() == written;
Review comment:
Yes, it is not useful for now. I don't want to change to many code in this issue. Just have filed https://issues.apache.org/jira/browse/RATIS-1098 as a followup.
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
szetszwo commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505108992
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java
##########
@@ -94,42 +74,28 @@ public void fail(Throwable e) {
public CompletableFuture<DataStreamReply> getReplyFuture(){
return replyFuture;
}
-
- public DataStreamWindowRequest(long streamId, long messageId,
- long seqNum, ByteBuffer data){
- this.streamId = streamId;
- this.messageId = messageId;
- this.data = data.slice();
- this.seqNum = seqNum;
- }
}
- private SlidingWindow.Client<DataStreamWindowRequest, DataStreamReply> slidingWindow;
- private Semaphore requestSemaphore;
- private DataStreamClientRpc dataStreamClientRpc;
+ private final DataStreamClientRpc dataStreamClientRpc;
+ private final SlidingWindow.Client<DataStreamWindowRequest, DataStreamReply> slidingWindow;
+ private final Semaphore requestSemaphore;
- public OrderedStreamAsync(DataStreamClientRpc dataStreamClientRpc,
- RaftProperties properties){
+ OrderedStreamAsync(ClientId clientId, DataStreamClientRpc dataStreamClientRpc, RaftProperties properties){
this.dataStreamClientRpc = dataStreamClientRpc;
+ this.slidingWindow = new SlidingWindow.Client<>(clientId);
this.requestSemaphore = new Semaphore(RaftClientConfigKeys.Async.outstandingRequestsMax(properties)*2);
Review comment:
It was an experimental result. Indeed, we should use a new conf instead of reusing RaftClientConfigKeys.Async.OUTSTANDING_REQUESTS_MAX_KEY. Will change it to use the new RaftClientConfigKeys.DataStream.OUTSTANDING_REQUESTS_MAX_KEY.
----------------------------------------------------------------
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 merged pull request #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang merged pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505103978
##########
File path: ratis-netty/src/main/java/org/apache/ratis/netty/server/NettyServerStreamRpc.java
##########
@@ -79,31 +78,34 @@ public NettyServerStreamRpc(RaftPeer server, StateMachine stateMachine) {
}
}
- private void writeTo(ByteBuf buf, DataStream stream, boolean released) {
+ private long writeTo(ByteBuf buf, DataStream stream, boolean released) {
if (released) {
- return;
+ return 0;
}
try {
if (stream == null) {
- return;
+ return 0;
}
final WritableByteChannel channel = stream.getWritableByteChannel();
+ long written = 0;
for (ByteBuffer buffer : buf.nioBuffers()) {
try {
- channel.write(buffer);
+ written += channel.write(buffer);
} catch (Throwable t) {
throw new CompletionException(t);
}
}
+ return written;
} finally {
buf.release();
}
}
- private void sendReply(DataStreamRequestByteBuf request, ChannelHandlerContext ctx) {
- final DataStreamReply reply = new DataStreamReplyByteBuffer(
- request.getStreamId(), request.getDataOffset(), ByteBuffer.wrap("OK".getBytes()));
+ private void sendReply(DataStreamRequestByteBuf request, long written, ChannelHandlerContext ctx) {
+ final boolean success = request.getDataLength() == written;
Review comment:
`success ` seems useless.
----------------------------------------------------------------
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] amaliujia commented on a change in pull request #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
amaliujia commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505132543
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
##########
@@ -36,59 +36,56 @@
import java.nio.ByteBuffer;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* Streaming client implementation
* allows client to create streams and send asynchronously.
*/
-
public class DataStreamClientImpl implements DataStreamClient {
public static final Logger LOG = LoggerFactory.getLogger(DataStreamClientImpl.class);
- private DataStreamClientRpc dataStreamClientRpc;
- private OrderedStreamAsync orderedStreamAsync;
// TODO Similar to RaftClientImpl, pass ClientId and RaftGroupId/RaftGroup in constructor.
private final ClientId clientId = ClientId.randomId();
private final RaftGroupId groupId = RaftGroupId.randomId();
- private RaftPeer raftServer;
- private RaftProperties properties;
- private Parameters parameters;
- private long streamId = 0;
-
- public DataStreamClientImpl(RaftPeer raftServer,
- RaftProperties properties,
- Parameters parameters) {
- this.raftServer = Objects.requireNonNull(raftServer,
- "peer == null");
- this.properties = properties;
- this.parameters = parameters;
+
+ private final RaftPeer raftServer;
+ private final DataStreamClientRpc dataStreamClientRpc;
+ private final OrderedStreamAsync orderedStreamAsync;
+
+ private final AtomicInteger streamId = new AtomicInteger();
+
+ public DataStreamClientImpl(RaftPeer server, RaftProperties properties, Parameters parameters) {
+ this.raftServer = Objects.requireNonNull(server, "server == null");
final SupportedDataStreamType type = RaftConfigKeys.DataStream.type(properties, LOG::info);
this.dataStreamClientRpc = DataStreamClientFactory.cast(type.newFactory(parameters))
.newDataStreamClientRpc(raftServer, properties);
- this.orderedStreamAsync = new OrderedStreamAsync(dataStreamClientRpc, properties);
+ this.orderedStreamAsync = new OrderedStreamAsync(clientId, dataStreamClientRpc, properties);
}
public class DataStreamOutputImpl implements DataStreamOutput {
- private long streamId = 0;
- private long messageId = 0;
+ private final long streamId;
private final RaftClientRequest header;
private final CompletableFuture<DataStreamReply> headerFuture;
+ private long streamOffset = 0;
+
public DataStreamOutputImpl(long id){
this.streamId = id;
this.header = new RaftClientRequest(clientId, raftServer.getId(), groupId, RaftClientImpl.nextCallId(),
RaftClientRequest.writeRequestType());
- this.headerFuture = orderedStreamAsync.sendRequest(streamId, messageId,
+ this.headerFuture = orderedStreamAsync.sendRequest(streamId, -1,
Review comment:
Not sure on -1 here: so because this is just a header request so set streamOffset as `-1` does not matter? (e.g. no need to use 0).
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
szetszwo commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505109499
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java
##########
@@ -38,37 +37,18 @@
public class OrderedStreamAsync {
public static final Logger LOG = LoggerFactory.getLogger(OrderedStreamAsync.class);
- public static class DataStreamWindowRequest implements DataStreamRequest,
- SlidingWindow.ClientSideRequest<DataStreamReply> {
- private final long streamId;
- private final long messageId;
+ static class DataStreamWindowRequest extends DataStreamRequestByteBuffer
+ implements SlidingWindow.ClientSideRequest<DataStreamReply> {
private final long seqNum;
- private final ByteBuffer data;
- private boolean isFirst = false;
- private CompletableFuture<DataStreamReply> replyFuture = new CompletableFuture<>();
+ private final CompletableFuture<DataStreamReply> replyFuture = new CompletableFuture<>();
- public DataStreamRequestByteBuffer newDataStreamRequest(){
- return new DataStreamRequestByteBuffer(streamId, messageId, data.slice());
- }
-
- @Override
- public long getStreamId() {
- return streamId;
- }
-
- @Override
- public long getDataOffset() {
- return messageId;
- }
-
- @Override
- public long getDataLength() {
- return data.capacity();
+ DataStreamWindowRequest(long streamId, long offset, ByteBuffer data, long seqNum){
+ super(streamId, offset, data);
+ this.seqNum = seqNum;
}
@Override
public void setFirstRequest() {
- isFirst = true;
Review comment:
It is not used anywhere. Steaming is different from Async that Async may change leader in the middle of a series of calls but Stream won't. Therefore, isFirst is not useful in Streaming.
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang closed pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#issuecomment-708952014
@szetszwo Thanks the patch, @amaliujia Thanks for review. I have merged 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 commented on pull request #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#issuecomment-708841759
@szetszwo LGTM, +1. Just wait the fix of checkstyle.
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
szetszwo commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505138148
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/DataStreamClientImpl.java
##########
@@ -36,59 +36,56 @@
import java.nio.ByteBuffer;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* Streaming client implementation
* allows client to create streams and send asynchronously.
*/
-
public class DataStreamClientImpl implements DataStreamClient {
public static final Logger LOG = LoggerFactory.getLogger(DataStreamClientImpl.class);
- private DataStreamClientRpc dataStreamClientRpc;
- private OrderedStreamAsync orderedStreamAsync;
// TODO Similar to RaftClientImpl, pass ClientId and RaftGroupId/RaftGroup in constructor.
private final ClientId clientId = ClientId.randomId();
private final RaftGroupId groupId = RaftGroupId.randomId();
- private RaftPeer raftServer;
- private RaftProperties properties;
- private Parameters parameters;
- private long streamId = 0;
-
- public DataStreamClientImpl(RaftPeer raftServer,
- RaftProperties properties,
- Parameters parameters) {
- this.raftServer = Objects.requireNonNull(raftServer,
- "peer == null");
- this.properties = properties;
- this.parameters = parameters;
+
+ private final RaftPeer raftServer;
+ private final DataStreamClientRpc dataStreamClientRpc;
+ private final OrderedStreamAsync orderedStreamAsync;
+
+ private final AtomicInteger streamId = new AtomicInteger();
+
+ public DataStreamClientImpl(RaftPeer server, RaftProperties properties, Parameters parameters) {
+ this.raftServer = Objects.requireNonNull(server, "server == null");
final SupportedDataStreamType type = RaftConfigKeys.DataStream.type(properties, LOG::info);
this.dataStreamClientRpc = DataStreamClientFactory.cast(type.newFactory(parameters))
.newDataStreamClientRpc(raftServer, properties);
- this.orderedStreamAsync = new OrderedStreamAsync(dataStreamClientRpc, properties);
+ this.orderedStreamAsync = new OrderedStreamAsync(clientId, dataStreamClientRpc, properties);
}
public class DataStreamOutputImpl implements DataStreamOutput {
- private long streamId = 0;
- private long messageId = 0;
+ private final long streamId;
private final RaftClientRequest header;
private final CompletableFuture<DataStreamReply> headerFuture;
+ private long streamOffset = 0;
+
public DataStreamOutputImpl(long id){
this.streamId = id;
this.header = new RaftClientRequest(clientId, raftServer.getId(), groupId, RaftClientImpl.nextCallId(),
RaftClientRequest.writeRequestType());
- this.headerFuture = orderedStreamAsync.sendRequest(streamId, messageId,
+ this.headerFuture = orderedStreamAsync.sendRequest(streamId, -1,
Review comment:
Yes, we use -1 to indicate that it is the header. If we use 0, it may not be clear whether it is the header or the first data packet.
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505110006
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java
##########
@@ -38,37 +37,18 @@
public class OrderedStreamAsync {
public static final Logger LOG = LoggerFactory.getLogger(OrderedStreamAsync.class);
- public static class DataStreamWindowRequest implements DataStreamRequest,
- SlidingWindow.ClientSideRequest<DataStreamReply> {
- private final long streamId;
- private final long messageId;
+ static class DataStreamWindowRequest extends DataStreamRequestByteBuffer
+ implements SlidingWindow.ClientSideRequest<DataStreamReply> {
private final long seqNum;
- private final ByteBuffer data;
- private boolean isFirst = false;
- private CompletableFuture<DataStreamReply> replyFuture = new CompletableFuture<>();
+ private final CompletableFuture<DataStreamReply> replyFuture = new CompletableFuture<>();
- public DataStreamRequestByteBuffer newDataStreamRequest(){
- return new DataStreamRequestByteBuffer(streamId, messageId, data.slice());
- }
-
- @Override
- public long getStreamId() {
- return streamId;
- }
-
- @Override
- public long getDataOffset() {
- return messageId;
- }
-
- @Override
- public long getDataLength() {
- return data.capacity();
+ DataStreamWindowRequest(long streamId, long offset, ByteBuffer data, long seqNum){
+ super(streamId, offset, data);
+ this.seqNum = seqNum;
}
@Override
public void setFirstRequest() {
- isFirst = true;
Review comment:
I see, we can remove [request.setFirstRequest()](https://github.com/apache/incubator-ratis/blob/76a52efc4c6973d254d189ff81e5665ffea09120/ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java#L109) in this file.
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#issuecomment-708879472
Reopen PR to trigger CI.
----------------------------------------------------------------
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 #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
runzhiwang commented on a change in pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#discussion_r505101809
##########
File path: ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedStreamAsync.java
##########
@@ -38,37 +37,18 @@
public class OrderedStreamAsync {
public static final Logger LOG = LoggerFactory.getLogger(OrderedStreamAsync.class);
- public static class DataStreamWindowRequest implements DataStreamRequest,
- SlidingWindow.ClientSideRequest<DataStreamReply> {
- private final long streamId;
- private final long messageId;
+ static class DataStreamWindowRequest extends DataStreamRequestByteBuffer
+ implements SlidingWindow.ClientSideRequest<DataStreamReply> {
private final long seqNum;
- private final ByteBuffer data;
- private boolean isFirst = false;
- private CompletableFuture<DataStreamReply> replyFuture = new CompletableFuture<>();
+ private final CompletableFuture<DataStreamReply> replyFuture = new CompletableFuture<>();
- public DataStreamRequestByteBuffer newDataStreamRequest(){
- return new DataStreamRequestByteBuffer(streamId, messageId, data.slice());
- }
-
- @Override
- public long getStreamId() {
- return streamId;
- }
-
- @Override
- public long getDataOffset() {
- return messageId;
- }
-
- @Override
- public long getDataLength() {
- return data.capacity();
+ DataStreamWindowRequest(long streamId, long offset, ByteBuffer data, long seqNum){
+ super(streamId, offset, data);
+ this.seqNum = seqNum;
}
@Override
public void setFirstRequest() {
- isFirst = true;
Review comment:
Why we do not need this any more ?
----------------------------------------------------------------
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 pull request #224: RATIS-1097. DataStreamOutputImpl should use streamOffset instead of messageId
Posted by GitBox <gi...@apache.org>.
szetszwo commented on pull request #224:
URL: https://github.com/apache/incubator-ratis/pull/224#issuecomment-708923229
Oops, I have accidentally changed OrderedAsync.java. It should be OrderedStreamAsync.java
----------------------------------------------------------------
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