HDDS-6137. [Ozone-Streaming] Refactor KeyDataStreamOutput.

[guohao-rosicky]

What changes were proposed in this pull request?

Optimized the putBlock method of BlockDataStreamOutput

1. PutBlock does not use RaftRpc, but through the stream close() method

// putBlock metadata
ByteBuffer buf =
ContainerCommandRequestMessage.toMessage(putBlockRequestProto, null)
.getContent().asReadOnlyByteBuffer();
out.writeAsync(buf, StandardWriteOption.CLOSE); 

2.Mark the stream data boundary by sending an empty stream packet

// Mark the stream data boundary
out.writeAsync(ByteBuffer.allocateDirect(0).asReadOnlyBuffer()); 

3. Flush is also in the form of async RPC, because putBlock needs to fetch bcsId (Raft log index)

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-6137

How was this patch tested?

Use the existing UT

[captainzmc]

Thanks @guohao-rosicky Improve this. LGTM over all. TestBlockDataStreamOutput has some error, can you confirm?

[szetszwo]

@guohao-rosicky, Writing an empty buffer may not work as expected. In the streaming model, data is continuous but not block-by-block. Both sender and receiver must consider that the data is a stream of bytes.

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

Let me think about how to make it work.

[guohao-rosicky]

Thanks for your comment @szetszwo . I wrote a piece of pseudo-code, Ideas about the modification of org.apache.ratis.netty.server.DataStreamManagement#writeTo.

What do you think of this idea? There is no need to send this empty buffer rpc.

[szetszwo]

@guohao-rosicky , this is a standard problem in streaming. The standard solution is to write the length with the data. Let's don't change the API for it.

[guohao-rosicky]

Hi, @szetszwo
I have a question, about

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

I noticed that in HDFS, marking the end of stream is also represented by an empty package.
In Ratis Stream, sticky packets have already been dealt with.

org.apache.ratis.netty.NettyDataStreamUtils#encodeDataStreamRequestHeader
https://github.com/apache/ratis/blob/master/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java#L90

@captainzmc tested the code from some time ago and did not find this problem

[szetszwo]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

[guohao-rosicky]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

Thanks @szetszwo, Patch has been applied

[szetszwo]

@guohao-rosicky , please fix the checkstyle warnings and test the change manually. I have not tested it. Thanks.

[captainzmc]

Thanks @guohao-rosicky@szetszwo update this. I used this PR test and found that the file failed to write. In DataNode IndexOutOfBoundsException will appear.

[szetszwo]

@captainzmc , thanks. Let me test the serialization/deserialization code.

[szetszwo]

Found the bug: ByteBu.nioBuffers(index, length) somehow is not working. Let me fix it.

[szetszwo]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

[captainzmc]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

Thanks @szetszwo for the update. Using new patch I found IndexOutOfBoundsException still appear.
I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

[szetszwo]

@captainzmc , thanks a lot for testing it!

I added one more test for the serialization and buffers. It worked fine. See https://issues.apache.org/jira/secure/attachment/13041275/3195_testBuffers.patch

I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

It seems there are some synchronization problems. Let me check.

[szetszwo]

@captainzmc , could you post the test code?

[captainzmc]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Actually there a problem with my previous description. I noticed that the when I use 30 threads writing, we’ll get same error.

[captainzmc]

@szetszwo Here is my datanode log. https://issues.apache.org/jira/secure/attachment/13041276/ozone-root-datanode-9-29-173-57.log
In the test, I wrote 30 files concurrently. Each file is 128MB and the chunk size is 1MB

[szetszwo]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

[szetszwo]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Thanks for sharing the test code. It looks good.

[captainzmc]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

Thanks @szetszwo for update this. I had just test this patch, we still get same error. Just as the same as this.

[szetszwo]

Let's add an abstract class instead static methods. Then, KeyDataStreamOutput can extends it. If we need to add SmallFileDataStreamOutput, it can extends it too.

package org.apache.hadoop.ozone.client.io;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.protocol.exceptions.RaftRetryFailureException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
public abstract class AbstractDataStreamOutput implements ByteBufferStreamOutput {
private final Map<Class<? extends Throwable>, RetryPolicy> retryPolicyMap;
private int retryCount;
private boolean isException;
protected AbstractDataStreamOutput() {
this.retryPolicyMap = HddsClientUtils.getExceptionList()
.stream()
.collect(Collectors.toMap(Function.identity(),
e -> RetryPolicies.TRY_ONCE_THEN_FAIL));
}
@VisibleForTesting
public int getRetryCount() {
return retryCount;
}
void resetRetryCount() {
retryCount = 0;
}
boolean isException() {
return isException;
}
/**
* Checks if the provided exception signifies retry failure in ratis client.
* In case of retry failure, ratis client throws RaftRetryFailureException
* and all succeeding operations are failed with AlreadyClosedException.
*/
boolean checkForRetryFailure(Throwable t) {
return t instanceof RaftRetryFailureException
|| t instanceof AlreadyClosedException;
}
// Every container specific exception from datatnode will be seen as
// StorageContainerException
boolean checkIfContainerToExclude(Throwable t) {
return t instanceof StorageContainerException;
}
private void setExceptionAndThrow(IOException ioe) throws IOException {
isException = true;
throw ioe;
}
void handleRetry(IOException exception) throws IOException {
RetryPolicy retryPolicy = retryPolicyMap
.get(HddsClientUtils.checkForException(exception).getClass());
if (retryPolicy == null) {
retryPolicy = retryPolicyMap.get(Exception.class);
}
handleRetry(exception, retryPolicy);
}
private void handleRetry(IOException exception, RetryPolicy retryPolicy)
throws IOException {
RetryPolicy.RetryAction action = null;
try {
action = retryPolicy.shouldRetry(exception, retryCount, 0, true);
} catch (Exception e) {
setExceptionAndThrow(new IOException(e));
}
if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) {
String msg = "";
if (action.reason != null) {
msg = "Retry request failed. " + action.reason;
//LOG.error(msg, exception);
}
setExceptionAndThrow(new IOException(msg, exception));
}
// Throw the exception if the thread is interrupted
if (Thread.currentThread().isInterrupted()) {
//LOG.warn("Interrupted while trying for retry");
setExceptionAndThrow(exception);
}
Preconditions.checkArgument(
action.action == RetryPolicy.RetryAction.RetryDecision.RETRY);
if (action.delayMillis > 0) {
try {
Thread.sleep(action.delayMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
IOException ioe = (IOException) new InterruptedIOException(
"Interrupted: action=" + action + ", retry policy=" + retryPolicy)
.initCause(e);
setExceptionAndThrow(ioe);
}
}
retryCount++;
}
}

[guohao-rosicky]

Ok, I'll change it

[guohao-rosicky]

Has been modified. @szetszwo Please take a look.Thanks.

[szetszwo]

@guohao-rosicky , please review it yourself -- go through the change and remove the unused code (e.g. ConsumerWithIOException). Thanks.

[szetszwo]

@guohao-rosicky , thanks for the update. Some comments inlined.

[szetszwo]

runCommandAsync is not used. Let's add it later.

[szetszwo]

Let's remove LOG.error(..) since it already throws an exception.

[szetszwo]

Add final.

[szetszwo]

This method is not used anywhere. Let's remove it.

[szetszwo]

We should pass the retryPolicyMap to the super constructor.

[szetszwo]

The map is passed incorrectly. Below is the original map.

- this.retryPolicyMap = HddsClientUtils.getRetryPolicyByException(
- config.getMaxRetryCount(), config.getRetryInterval());

[captainzmc]

+1 Thanks @guohao-rosicky. The change looks good.

[szetszwo]

+1 the change looks good.

[guohao-rosicky]

What changes were proposed in this pull request?

Optimized the putBlock method of BlockDataStreamOutput

1. PutBlock does not use RaftRpc, but through the stream close() method

// putBlock metadata
ByteBuffer buf =
ContainerCommandRequestMessage.toMessage(putBlockRequestProto, null)
.getContent().asReadOnlyByteBuffer();
out.writeAsync(buf, StandardWriteOption.CLOSE); 

2.Mark the stream data boundary by sending an empty stream packet

// Mark the stream data boundary
out.writeAsync(ByteBuffer.allocateDirect(0).asReadOnlyBuffer()); 

3. Flush is also in the form of async RPC, because putBlock needs to fetch bcsId (Raft log index)

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-6137

How was this patch tested?

Use the existing UT

[captainzmc]

Thanks @guohao-rosicky Improve this. LGTM over all. TestBlockDataStreamOutput has some error, can you confirm?

[szetszwo]

@guohao-rosicky, Writing an empty buffer may not work as expected. In the streaming model, data is continuous but not block-by-block. Both sender and receiver must consider that the data is a stream of bytes.

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

Let me think about how to make it work.

[guohao-rosicky]

Thanks for your comment @szetszwo . I wrote a piece of pseudo-code, Ideas about the modification of org.apache.ratis.netty.server.DataStreamManagement#writeTo.

What do you think of this idea? There is no need to send this empty buffer rpc.

[szetszwo]

@guohao-rosicky , this is a standard problem in streaming. The standard solution is to write the length with the data. Let's don't change the API for it.

[guohao-rosicky]

Hi, @szetszwo
I have a question, about

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

I noticed that in HDFS, marking the end of stream is also represented by an empty package.
In Ratis Stream, sticky packets have already been dealt with.

org.apache.ratis.netty.NettyDataStreamUtils#encodeDataStreamRequestHeader
https://github.com/apache/ratis/blob/master/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java#L90

@captainzmc tested the code from some time ago and did not find this problem

[szetszwo]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

[guohao-rosicky]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

Thanks @szetszwo, Patch has been applied

[szetszwo]

@guohao-rosicky , please fix the checkstyle warnings and test the change manually. I have not tested it. Thanks.

[captainzmc]

Thanks @guohao-rosicky@szetszwo update this. I used this PR test and found that the file failed to write. In DataNode IndexOutOfBoundsException will appear.

[szetszwo]

@captainzmc , thanks. Let me test the serialization/deserialization code.

[szetszwo]

Found the bug: ByteBu.nioBuffers(index, length) somehow is not working. Let me fix it.

[szetszwo]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

[captainzmc]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

Thanks @szetszwo for the update. Using new patch I found IndexOutOfBoundsException still appear.
I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

[szetszwo]

@captainzmc , thanks a lot for testing it!

I added one more test for the serialization and buffers. It worked fine. See https://issues.apache.org/jira/secure/attachment/13041275/3195_testBuffers.patch

I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

It seems there are some synchronization problems. Let me check.

[szetszwo]

@captainzmc , could you post the test code?

[captainzmc]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Actually there a problem with my previous description. I noticed that the when I use 30 threads writing, we’ll get same error.

[captainzmc]

@szetszwo Here is my datanode log. https://issues.apache.org/jira/secure/attachment/13041276/ozone-root-datanode-9-29-173-57.log
In the test, I wrote 30 files concurrently. Each file is 128MB and the chunk size is 1MB

[szetszwo]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

[szetszwo]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Thanks for sharing the test code. It looks good.

[captainzmc]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

Thanks @szetszwo for update this. I had just test this patch, we still get same error. Just as the same as this.

[szetszwo]

Let's add an abstract class instead static methods. Then, KeyDataStreamOutput can extends it. If we need to add SmallFileDataStreamOutput, it can extends it too.

package org.apache.hadoop.ozone.client.io;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.protocol.exceptions.RaftRetryFailureException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
public abstract class AbstractDataStreamOutput implements ByteBufferStreamOutput {
private final Map<Class<? extends Throwable>, RetryPolicy> retryPolicyMap;
private int retryCount;
private boolean isException;
protected AbstractDataStreamOutput() {
this.retryPolicyMap = HddsClientUtils.getExceptionList()
.stream()
.collect(Collectors.toMap(Function.identity(),
e -> RetryPolicies.TRY_ONCE_THEN_FAIL));
}
@VisibleForTesting
public int getRetryCount() {
return retryCount;
}
void resetRetryCount() {
retryCount = 0;
}
boolean isException() {
return isException;
}
/**
* Checks if the provided exception signifies retry failure in ratis client.
* In case of retry failure, ratis client throws RaftRetryFailureException
* and all succeeding operations are failed with AlreadyClosedException.
*/
boolean checkForRetryFailure(Throwable t) {
return t instanceof RaftRetryFailureException
|| t instanceof AlreadyClosedException;
}
// Every container specific exception from datatnode will be seen as
// StorageContainerException
boolean checkIfContainerToExclude(Throwable t) {
return t instanceof StorageContainerException;
}
private void setExceptionAndThrow(IOException ioe) throws IOException {
isException = true;
throw ioe;
}
void handleRetry(IOException exception) throws IOException {
RetryPolicy retryPolicy = retryPolicyMap
.get(HddsClientUtils.checkForException(exception).getClass());
if (retryPolicy == null) {
retryPolicy = retryPolicyMap.get(Exception.class);
}
handleRetry(exception, retryPolicy);
}
private void handleRetry(IOException exception, RetryPolicy retryPolicy)
throws IOException {
RetryPolicy.RetryAction action = null;
try {
action = retryPolicy.shouldRetry(exception, retryCount, 0, true);
} catch (Exception e) {
setExceptionAndThrow(new IOException(e));
}
if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) {
String msg = "";
if (action.reason != null) {
msg = "Retry request failed. " + action.reason;
//LOG.error(msg, exception);
}
setExceptionAndThrow(new IOException(msg, exception));
}
// Throw the exception if the thread is interrupted
if (Thread.currentThread().isInterrupted()) {
//LOG.warn("Interrupted while trying for retry");
setExceptionAndThrow(exception);
}
Preconditions.checkArgument(
action.action == RetryPolicy.RetryAction.RetryDecision.RETRY);
if (action.delayMillis > 0) {
try {
Thread.sleep(action.delayMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
IOException ioe = (IOException) new InterruptedIOException(
"Interrupted: action=" + action + ", retry policy=" + retryPolicy)
.initCause(e);
setExceptionAndThrow(ioe);
}
}
retryCount++;
}
}

[guohao-rosicky]

Ok, I'll change it

[guohao-rosicky]

Has been modified. @szetszwo Please take a look.Thanks.

[szetszwo]

@guohao-rosicky , please review it yourself -- go through the change and remove the unused code (e.g. ConsumerWithIOException). Thanks.

[szetszwo]

@guohao-rosicky , thanks for the update. Some comments inlined.

[szetszwo]

runCommandAsync is not used. Let's add it later.

[szetszwo]

Let's remove LOG.error(..) since it already throws an exception.

[szetszwo]

Add final.

[szetszwo]

This method is not used anywhere. Let's remove it.

[szetszwo]

We should pass the retryPolicyMap to the super constructor.

[szetszwo]

The map is passed incorrectly. Below is the original map.

- this.retryPolicyMap = HddsClientUtils.getRetryPolicyByException(
- config.getMaxRetryCount(), config.getRetryInterval());

[captainzmc]

+1 Thanks @guohao-rosicky. The change looks good.

[szetszwo]

+1 the change looks good.

[guohao-rosicky]

What changes were proposed in this pull request?

Optimized the putBlock method of BlockDataStreamOutput

1. PutBlock does not use RaftRpc, but through the stream close() method

// putBlock metadata
ByteBuffer buf =
ContainerCommandRequestMessage.toMessage(putBlockRequestProto, null)
.getContent().asReadOnlyByteBuffer();
out.writeAsync(buf, StandardWriteOption.CLOSE); 

2.Mark the stream data boundary by sending an empty stream packet

// Mark the stream data boundary
out.writeAsync(ByteBuffer.allocateDirect(0).asReadOnlyBuffer()); 

3. Flush is also in the form of async RPC, because putBlock needs to fetch bcsId (Raft log index)

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-6137

How was this patch tested?

Use the existing UT

[captainzmc]

Thanks @guohao-rosicky Improve this. LGTM over all. TestBlockDataStreamOutput has some error, can you confirm?

[szetszwo]

@guohao-rosicky, Writing an empty buffer may not work as expected. In the streaming model, data is continuous but not block-by-block. Both sender and receiver must consider that the data is a stream of bytes.

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

Let me think about how to make it work.

[guohao-rosicky]

Thanks for your comment @szetszwo . I wrote a piece of pseudo-code, Ideas about the modification of org.apache.ratis.netty.server.DataStreamManagement#writeTo.

What do you think of this idea? There is no need to send this empty buffer rpc.

[szetszwo]

@guohao-rosicky , this is a standard problem in streaming. The standard solution is to write the length with the data. Let's don't change the API for it.

[guohao-rosicky]

Hi, @szetszwo
I have a question, about

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

I noticed that in HDFS, marking the end of stream is also represented by an empty package.
In Ratis Stream, sticky packets have already been dealt with.

org.apache.ratis.netty.NettyDataStreamUtils#encodeDataStreamRequestHeader
https://github.com/apache/ratis/blob/master/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java#L90

@captainzmc tested the code from some time ago and did not find this problem

[szetszwo]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

[guohao-rosicky]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

Thanks @szetszwo, Patch has been applied

[szetszwo]

@guohao-rosicky , please fix the checkstyle warnings and test the change manually. I have not tested it. Thanks.

[captainzmc]

Thanks @guohao-rosicky@szetszwo update this. I used this PR test and found that the file failed to write. In DataNode IndexOutOfBoundsException will appear.

[szetszwo]

@captainzmc , thanks. Let me test the serialization/deserialization code.

[szetszwo]

Found the bug: ByteBu.nioBuffers(index, length) somehow is not working. Let me fix it.

[szetszwo]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

[captainzmc]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

Thanks @szetszwo for the update. Using new patch I found IndexOutOfBoundsException still appear.
I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

[szetszwo]

@captainzmc , thanks a lot for testing it!

I added one more test for the serialization and buffers. It worked fine. See https://issues.apache.org/jira/secure/attachment/13041275/3195_testBuffers.patch

I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

It seems there are some synchronization problems. Let me check.

[szetszwo]

@captainzmc , could you post the test code?

[captainzmc]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Actually there a problem with my previous description. I noticed that the when I use 30 threads writing, we’ll get same error.

[captainzmc]

@szetszwo Here is my datanode log. https://issues.apache.org/jira/secure/attachment/13041276/ozone-root-datanode-9-29-173-57.log
In the test, I wrote 30 files concurrently. Each file is 128MB and the chunk size is 1MB

[szetszwo]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

[szetszwo]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Thanks for sharing the test code. It looks good.

[captainzmc]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

Thanks @szetszwo for update this. I had just test this patch, we still get same error. Just as the same as this.

[szetszwo]

Let's add an abstract class instead static methods. Then, KeyDataStreamOutput can extends it. If we need to add SmallFileDataStreamOutput, it can extends it too.

package org.apache.hadoop.ozone.client.io;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.protocol.exceptions.RaftRetryFailureException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
public abstract class AbstractDataStreamOutput implements ByteBufferStreamOutput {
private final Map<Class<? extends Throwable>, RetryPolicy> retryPolicyMap;
private int retryCount;
private boolean isException;
protected AbstractDataStreamOutput() {
this.retryPolicyMap = HddsClientUtils.getExceptionList()
.stream()
.collect(Collectors.toMap(Function.identity(),
e -> RetryPolicies.TRY_ONCE_THEN_FAIL));
}
@VisibleForTesting
public int getRetryCount() {
return retryCount;
}
void resetRetryCount() {
retryCount = 0;
}
boolean isException() {
return isException;
}
/**
* Checks if the provided exception signifies retry failure in ratis client.
* In case of retry failure, ratis client throws RaftRetryFailureException
* and all succeeding operations are failed with AlreadyClosedException.
*/
boolean checkForRetryFailure(Throwable t) {
return t instanceof RaftRetryFailureException
|| t instanceof AlreadyClosedException;
}
// Every container specific exception from datatnode will be seen as
// StorageContainerException
boolean checkIfContainerToExclude(Throwable t) {
return t instanceof StorageContainerException;
}
private void setExceptionAndThrow(IOException ioe) throws IOException {
isException = true;
throw ioe;
}
void handleRetry(IOException exception) throws IOException {
RetryPolicy retryPolicy = retryPolicyMap
.get(HddsClientUtils.checkForException(exception).getClass());
if (retryPolicy == null) {
retryPolicy = retryPolicyMap.get(Exception.class);
}
handleRetry(exception, retryPolicy);
}
private void handleRetry(IOException exception, RetryPolicy retryPolicy)
throws IOException {
RetryPolicy.RetryAction action = null;
try {
action = retryPolicy.shouldRetry(exception, retryCount, 0, true);
} catch (Exception e) {
setExceptionAndThrow(new IOException(e));
}
if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) {
String msg = "";
if (action.reason != null) {
msg = "Retry request failed. " + action.reason;
//LOG.error(msg, exception);
}
setExceptionAndThrow(new IOException(msg, exception));
}
// Throw the exception if the thread is interrupted
if (Thread.currentThread().isInterrupted()) {
//LOG.warn("Interrupted while trying for retry");
setExceptionAndThrow(exception);
}
Preconditions.checkArgument(
action.action == RetryPolicy.RetryAction.RetryDecision.RETRY);
if (action.delayMillis > 0) {
try {
Thread.sleep(action.delayMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
IOException ioe = (IOException) new InterruptedIOException(
"Interrupted: action=" + action + ", retry policy=" + retryPolicy)
.initCause(e);
setExceptionAndThrow(ioe);
}
}
retryCount++;
}
}

[guohao-rosicky]

Ok, I'll change it

[guohao-rosicky]

Has been modified. @szetszwo Please take a look.Thanks.

[szetszwo]

@guohao-rosicky , please review it yourself -- go through the change and remove the unused code (e.g. ConsumerWithIOException). Thanks.

[szetszwo]

@guohao-rosicky , thanks for the update. Some comments inlined.

[szetszwo]

runCommandAsync is not used. Let's add it later.

[szetszwo]

Let's remove LOG.error(..) since it already throws an exception.

[szetszwo]

Add final.

[szetszwo]

This method is not used anywhere. Let's remove it.

[szetszwo]

We should pass the retryPolicyMap to the super constructor.

[szetszwo]

The map is passed incorrectly. Below is the original map.

- this.retryPolicyMap = HddsClientUtils.getRetryPolicyByException(
- config.getMaxRetryCount(), config.getRetryInterval());

[captainzmc]

+1 Thanks @guohao-rosicky. The change looks good.

[szetszwo]

+1 the change looks good.

[guohao-rosicky]

What changes were proposed in this pull request?

Optimized the putBlock method of BlockDataStreamOutput

1. PutBlock does not use RaftRpc, but through the stream close() method

// putBlock metadata
ByteBuffer buf =
ContainerCommandRequestMessage.toMessage(putBlockRequestProto, null)
.getContent().asReadOnlyByteBuffer();
out.writeAsync(buf, StandardWriteOption.CLOSE); 

2.Mark the stream data boundary by sending an empty stream packet

// Mark the stream data boundary
out.writeAsync(ByteBuffer.allocateDirect(0).asReadOnlyBuffer()); 

3. Flush is also in the form of async RPC, because putBlock needs to fetch bcsId (Raft log index)

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-6137

How was this patch tested?

Use the existing UT

[captainzmc]

Thanks @guohao-rosicky Improve this. LGTM over all. TestBlockDataStreamOutput has some error, can you confirm?

[szetszwo]

@guohao-rosicky, Writing an empty buffer may not work as expected. In the streaming model, data is continuous but not block-by-block. Both sender and receiver must consider that the data is a stream of bytes.

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

Let me think about how to make it work.

[guohao-rosicky]

Thanks for your comment @szetszwo . I wrote a piece of pseudo-code, Ideas about the modification of org.apache.ratis.netty.server.DataStreamManagement#writeTo.

What do you think of this idea? There is no need to send this empty buffer rpc.

[szetszwo]

@guohao-rosicky , this is a standard problem in streaming. The standard solution is to write the length with the data. Let's don't change the API for it.

[guohao-rosicky]

Hi, @szetszwo
I have a question, about

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

I noticed that in HDFS, marking the end of stream is also represented by an empty package.
In Ratis Stream, sticky packets have already been dealt with.

org.apache.ratis.netty.NettyDataStreamUtils#encodeDataStreamRequestHeader
https://github.com/apache/ratis/blob/master/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java#L90

@captainzmc tested the code from some time ago and did not find this problem

[szetszwo]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

[guohao-rosicky]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

Thanks @szetszwo, Patch has been applied

[szetszwo]

@guohao-rosicky , please fix the checkstyle warnings and test the change manually. I have not tested it. Thanks.

[captainzmc]

Thanks @guohao-rosicky@szetszwo update this. I used this PR test and found that the file failed to write. In DataNode IndexOutOfBoundsException will appear.

[szetszwo]

@captainzmc , thanks. Let me test the serialization/deserialization code.

[szetszwo]

Found the bug: ByteBu.nioBuffers(index, length) somehow is not working. Let me fix it.

[szetszwo]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

[captainzmc]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

Thanks @szetszwo for the update. Using new patch I found IndexOutOfBoundsException still appear.
I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

[szetszwo]

@captainzmc , thanks a lot for testing it!

I added one more test for the serialization and buffers. It worked fine. See https://issues.apache.org/jira/secure/attachment/13041275/3195_testBuffers.patch

I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

It seems there are some synchronization problems. Let me check.

[szetszwo]

@captainzmc , could you post the test code?

[captainzmc]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Actually there a problem with my previous description. I noticed that the when I use 30 threads writing, we’ll get same error.

[captainzmc]

@szetszwo Here is my datanode log. https://issues.apache.org/jira/secure/attachment/13041276/ozone-root-datanode-9-29-173-57.log
In the test, I wrote 30 files concurrently. Each file is 128MB and the chunk size is 1MB

[szetszwo]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

[szetszwo]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Thanks for sharing the test code. It looks good.

[captainzmc]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

Thanks @szetszwo for update this. I had just test this patch, we still get same error. Just as the same as this.

[szetszwo]

Let's add an abstract class instead static methods. Then, KeyDataStreamOutput can extends it. If we need to add SmallFileDataStreamOutput, it can extends it too.

package org.apache.hadoop.ozone.client.io;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.protocol.exceptions.RaftRetryFailureException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
public abstract class AbstractDataStreamOutput implements ByteBufferStreamOutput {
private final Map<Class<? extends Throwable>, RetryPolicy> retryPolicyMap;
private int retryCount;
private boolean isException;
protected AbstractDataStreamOutput() {
this.retryPolicyMap = HddsClientUtils.getExceptionList()
.stream()
.collect(Collectors.toMap(Function.identity(),
e -> RetryPolicies.TRY_ONCE_THEN_FAIL));
}
@VisibleForTesting
public int getRetryCount() {
return retryCount;
}
void resetRetryCount() {
retryCount = 0;
}
boolean isException() {
return isException;
}
/**
* Checks if the provided exception signifies retry failure in ratis client.
* In case of retry failure, ratis client throws RaftRetryFailureException
* and all succeeding operations are failed with AlreadyClosedException.
*/
boolean checkForRetryFailure(Throwable t) {
return t instanceof RaftRetryFailureException
|| t instanceof AlreadyClosedException;
}
// Every container specific exception from datatnode will be seen as
// StorageContainerException
boolean checkIfContainerToExclude(Throwable t) {
return t instanceof StorageContainerException;
}
private void setExceptionAndThrow(IOException ioe) throws IOException {
isException = true;
throw ioe;
}
void handleRetry(IOException exception) throws IOException {
RetryPolicy retryPolicy = retryPolicyMap
.get(HddsClientUtils.checkForException(exception).getClass());
if (retryPolicy == null) {
retryPolicy = retryPolicyMap.get(Exception.class);
}
handleRetry(exception, retryPolicy);
}
private void handleRetry(IOException exception, RetryPolicy retryPolicy)
throws IOException {
RetryPolicy.RetryAction action = null;
try {
action = retryPolicy.shouldRetry(exception, retryCount, 0, true);
} catch (Exception e) {
setExceptionAndThrow(new IOException(e));
}
if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) {
String msg = "";
if (action.reason != null) {
msg = "Retry request failed. " + action.reason;
//LOG.error(msg, exception);
}
setExceptionAndThrow(new IOException(msg, exception));
}
// Throw the exception if the thread is interrupted
if (Thread.currentThread().isInterrupted()) {
//LOG.warn("Interrupted while trying for retry");
setExceptionAndThrow(exception);
}
Preconditions.checkArgument(
action.action == RetryPolicy.RetryAction.RetryDecision.RETRY);
if (action.delayMillis > 0) {
try {
Thread.sleep(action.delayMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
IOException ioe = (IOException) new InterruptedIOException(
"Interrupted: action=" + action + ", retry policy=" + retryPolicy)
.initCause(e);
setExceptionAndThrow(ioe);
}
}
retryCount++;
}
}

[guohao-rosicky]

Ok, I'll change it

[guohao-rosicky]

Has been modified. @szetszwo Please take a look.Thanks.

[szetszwo]

@guohao-rosicky , please review it yourself -- go through the change and remove the unused code (e.g. ConsumerWithIOException). Thanks.

[szetszwo]

@guohao-rosicky , thanks for the update. Some comments inlined.

[szetszwo]

runCommandAsync is not used. Let's add it later.

[szetszwo]

Let's remove LOG.error(..) since it already throws an exception.

[szetszwo]

Add final.

[szetszwo]

This method is not used anywhere. Let's remove it.

[szetszwo]

We should pass the retryPolicyMap to the super constructor.

[szetszwo]

The map is passed incorrectly. Below is the original map.

- this.retryPolicyMap = HddsClientUtils.getRetryPolicyByException(
- config.getMaxRetryCount(), config.getRetryInterval());

[captainzmc]

+1 Thanks @guohao-rosicky. The change looks good.

[szetszwo]

+1 the change looks good.

[guohao-rosicky]

What changes were proposed in this pull request?

Optimized the putBlock method of BlockDataStreamOutput

1. PutBlock does not use RaftRpc, but through the stream close() method

// putBlock metadata
ByteBuffer buf =
ContainerCommandRequestMessage.toMessage(putBlockRequestProto, null)
.getContent().asReadOnlyByteBuffer();
out.writeAsync(buf, StandardWriteOption.CLOSE); 

2.Mark the stream data boundary by sending an empty stream packet

// Mark the stream data boundary
out.writeAsync(ByteBuffer.allocateDirect(0).asReadOnlyBuffer()); 

3. Flush is also in the form of async RPC, because putBlock needs to fetch bcsId (Raft log index)

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-6137

How was this patch tested?

Use the existing UT

[captainzmc]

Thanks @guohao-rosicky Improve this. LGTM over all. TestBlockDataStreamOutput has some error, can you confirm?

[szetszwo]

@guohao-rosicky, Writing an empty buffer may not work as expected. In the streaming model, data is continuous but not block-by-block. Both sender and receiver must consider that the data is a stream of bytes.

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

Let me think about how to make it work.

[guohao-rosicky]

Thanks for your comment @szetszwo . I wrote a piece of pseudo-code, Ideas about the modification of org.apache.ratis.netty.server.DataStreamManagement#writeTo.

What do you think of this idea? There is no need to send this empty buffer rpc.

[szetszwo]

@guohao-rosicky , this is a standard problem in streaming. The standard solution is to write the length with the data. Let's don't change the API for it.

[guohao-rosicky]

Hi, @szetszwo
I have a question, about

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

I noticed that in HDFS, marking the end of stream is also represented by an empty package.
In Ratis Stream, sticky packets have already been dealt with.

org.apache.ratis.netty.NettyDataStreamUtils#encodeDataStreamRequestHeader
https://github.com/apache/ratis/blob/master/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java#L90

@captainzmc tested the code from some time ago and did not find this problem

[szetszwo]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

[guohao-rosicky]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

Thanks @szetszwo, Patch has been applied

[szetszwo]

@guohao-rosicky , please fix the checkstyle warnings and test the change manually. I have not tested it. Thanks.

[captainzmc]

Thanks @guohao-rosicky@szetszwo update this. I used this PR test and found that the file failed to write. In DataNode IndexOutOfBoundsException will appear.

[szetszwo]

@captainzmc , thanks. Let me test the serialization/deserialization code.

[szetszwo]

Found the bug: ByteBu.nioBuffers(index, length) somehow is not working. Let me fix it.

[szetszwo]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

[captainzmc]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

Thanks @szetszwo for the update. Using new patch I found IndexOutOfBoundsException still appear.
I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

[szetszwo]

@captainzmc , thanks a lot for testing it!

I added one more test for the serialization and buffers. It worked fine. See https://issues.apache.org/jira/secure/attachment/13041275/3195_testBuffers.patch

I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

It seems there are some synchronization problems. Let me check.

[szetszwo]

@captainzmc , could you post the test code?

[captainzmc]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Actually there a problem with my previous description. I noticed that the when I use 30 threads writing, we’ll get same error.

[captainzmc]

@szetszwo Here is my datanode log. https://issues.apache.org/jira/secure/attachment/13041276/ozone-root-datanode-9-29-173-57.log
In the test, I wrote 30 files concurrently. Each file is 128MB and the chunk size is 1MB

[szetszwo]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

[szetszwo]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Thanks for sharing the test code. It looks good.

[captainzmc]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

Thanks @szetszwo for update this. I had just test this patch, we still get same error. Just as the same as this.

[szetszwo]

Let's add an abstract class instead static methods. Then, KeyDataStreamOutput can extends it. If we need to add SmallFileDataStreamOutput, it can extends it too.

package org.apache.hadoop.ozone.client.io;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.protocol.exceptions.RaftRetryFailureException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
public abstract class AbstractDataStreamOutput implements ByteBufferStreamOutput {
private final Map<Class<? extends Throwable>, RetryPolicy> retryPolicyMap;
private int retryCount;
private boolean isException;
protected AbstractDataStreamOutput() {
this.retryPolicyMap = HddsClientUtils.getExceptionList()
.stream()
.collect(Collectors.toMap(Function.identity(),
e -> RetryPolicies.TRY_ONCE_THEN_FAIL));
}
@VisibleForTesting
public int getRetryCount() {
return retryCount;
}
void resetRetryCount() {
retryCount = 0;
}
boolean isException() {
return isException;
}
/**
* Checks if the provided exception signifies retry failure in ratis client.
* In case of retry failure, ratis client throws RaftRetryFailureException
* and all succeeding operations are failed with AlreadyClosedException.
*/
boolean checkForRetryFailure(Throwable t) {
return t instanceof RaftRetryFailureException
|| t instanceof AlreadyClosedException;
}
// Every container specific exception from datatnode will be seen as
// StorageContainerException
boolean checkIfContainerToExclude(Throwable t) {
return t instanceof StorageContainerException;
}
private void setExceptionAndThrow(IOException ioe) throws IOException {
isException = true;
throw ioe;
}
void handleRetry(IOException exception) throws IOException {
RetryPolicy retryPolicy = retryPolicyMap
.get(HddsClientUtils.checkForException(exception).getClass());
if (retryPolicy == null) {
retryPolicy = retryPolicyMap.get(Exception.class);
}
handleRetry(exception, retryPolicy);
}
private void handleRetry(IOException exception, RetryPolicy retryPolicy)
throws IOException {
RetryPolicy.RetryAction action = null;
try {
action = retryPolicy.shouldRetry(exception, retryCount, 0, true);
} catch (Exception e) {
setExceptionAndThrow(new IOException(e));
}
if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) {
String msg = "";
if (action.reason != null) {
msg = "Retry request failed. " + action.reason;
//LOG.error(msg, exception);
}
setExceptionAndThrow(new IOException(msg, exception));
}
// Throw the exception if the thread is interrupted
if (Thread.currentThread().isInterrupted()) {
//LOG.warn("Interrupted while trying for retry");
setExceptionAndThrow(exception);
}
Preconditions.checkArgument(
action.action == RetryPolicy.RetryAction.RetryDecision.RETRY);
if (action.delayMillis > 0) {
try {
Thread.sleep(action.delayMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
IOException ioe = (IOException) new InterruptedIOException(
"Interrupted: action=" + action + ", retry policy=" + retryPolicy)
.initCause(e);
setExceptionAndThrow(ioe);
}
}
retryCount++;
}
}

[guohao-rosicky]

Ok, I'll change it

[guohao-rosicky]

Has been modified. @szetszwo Please take a look.Thanks.

[szetszwo]

@guohao-rosicky , please review it yourself -- go through the change and remove the unused code (e.g. ConsumerWithIOException). Thanks.

[szetszwo]

@guohao-rosicky , thanks for the update. Some comments inlined.

[szetszwo]

runCommandAsync is not used. Let's add it later.

[szetszwo]

Let's remove LOG.error(..) since it already throws an exception.

[szetszwo]

Add final.

[szetszwo]

This method is not used anywhere. Let's remove it.

[szetszwo]

We should pass the retryPolicyMap to the super constructor.

[szetszwo]

The map is passed incorrectly. Below is the original map.

- this.retryPolicyMap = HddsClientUtils.getRetryPolicyByException(
- config.getMaxRetryCount(), config.getRetryInterval());

[captainzmc]

+1 Thanks @guohao-rosicky. The change looks good.

[szetszwo]

+1 the change looks good.

[guohao-rosicky]

What changes were proposed in this pull request?

Optimized the putBlock method of BlockDataStreamOutput

1. PutBlock does not use RaftRpc, but through the stream close() method

// putBlock metadata
ByteBuffer buf =
ContainerCommandRequestMessage.toMessage(putBlockRequestProto, null)
.getContent().asReadOnlyByteBuffer();
out.writeAsync(buf, StandardWriteOption.CLOSE); 

2.Mark the stream data boundary by sending an empty stream packet

// Mark the stream data boundary
out.writeAsync(ByteBuffer.allocateDirect(0).asReadOnlyBuffer()); 

3. Flush is also in the form of async RPC, because putBlock needs to fetch bcsId (Raft log index)

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-6137

How was this patch tested?

Use the existing UT

[captainzmc]

Thanks @guohao-rosicky Improve this. LGTM over all. TestBlockDataStreamOutput has some error, can you confirm?

[szetszwo]

@guohao-rosicky, Writing an empty buffer may not work as expected. In the streaming model, data is continuous but not block-by-block. Both sender and receiver must consider that the data is a stream of bytes.

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

Let me think about how to make it work.

[guohao-rosicky]

Thanks for your comment @szetszwo . I wrote a piece of pseudo-code, Ideas about the modification of org.apache.ratis.netty.server.DataStreamManagement#writeTo.

What do you think of this idea? There is no need to send this empty buffer rpc.

[szetszwo]

@guohao-rosicky , this is a standard problem in streaming. The standard solution is to write the length with the data. Let's don't change the API for it.

[guohao-rosicky]

Hi, @szetszwo
I have a question, about

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

I noticed that in HDFS, marking the end of stream is also represented by an empty package.
In Ratis Stream, sticky packets have already been dealt with.

org.apache.ratis.netty.NettyDataStreamUtils#encodeDataStreamRequestHeader
https://github.com/apache/ratis/blob/master/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java#L90

@captainzmc tested the code from some time ago and did not find this problem

[szetszwo]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

[guohao-rosicky]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

Thanks @szetszwo, Patch has been applied

[szetszwo]

@guohao-rosicky , please fix the checkstyle warnings and test the change manually. I have not tested it. Thanks.

[captainzmc]

Thanks @guohao-rosicky@szetszwo update this. I used this PR test and found that the file failed to write. In DataNode IndexOutOfBoundsException will appear.

[szetszwo]

@captainzmc , thanks. Let me test the serialization/deserialization code.

[szetszwo]

Found the bug: ByteBu.nioBuffers(index, length) somehow is not working. Let me fix it.

[szetszwo]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

[captainzmc]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

Thanks @szetszwo for the update. Using new patch I found IndexOutOfBoundsException still appear.
I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

[szetszwo]

@captainzmc , thanks a lot for testing it!

I added one more test for the serialization and buffers. It worked fine. See https://issues.apache.org/jira/secure/attachment/13041275/3195_testBuffers.patch

I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

It seems there are some synchronization problems. Let me check.

[szetszwo]

@captainzmc , could you post the test code?

[captainzmc]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Actually there a problem with my previous description. I noticed that the when I use 30 threads writing, we’ll get same error.

[captainzmc]

@szetszwo Here is my datanode log. https://issues.apache.org/jira/secure/attachment/13041276/ozone-root-datanode-9-29-173-57.log
In the test, I wrote 30 files concurrently. Each file is 128MB and the chunk size is 1MB

[szetszwo]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

[szetszwo]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Thanks for sharing the test code. It looks good.

[captainzmc]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

Thanks @szetszwo for update this. I had just test this patch, we still get same error. Just as the same as this.

[szetszwo]

Let's add an abstract class instead static methods. Then, KeyDataStreamOutput can extends it. If we need to add SmallFileDataStreamOutput, it can extends it too.

package org.apache.hadoop.ozone.client.io;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.protocol.exceptions.RaftRetryFailureException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
public abstract class AbstractDataStreamOutput implements ByteBufferStreamOutput {
private final Map<Class<? extends Throwable>, RetryPolicy> retryPolicyMap;
private int retryCount;
private boolean isException;
protected AbstractDataStreamOutput() {
this.retryPolicyMap = HddsClientUtils.getExceptionList()
.stream()
.collect(Collectors.toMap(Function.identity(),
e -> RetryPolicies.TRY_ONCE_THEN_FAIL));
}
@VisibleForTesting
public int getRetryCount() {
return retryCount;
}
void resetRetryCount() {
retryCount = 0;
}
boolean isException() {
return isException;
}
/**
* Checks if the provided exception signifies retry failure in ratis client.
* In case of retry failure, ratis client throws RaftRetryFailureException
* and all succeeding operations are failed with AlreadyClosedException.
*/
boolean checkForRetryFailure(Throwable t) {
return t instanceof RaftRetryFailureException
|| t instanceof AlreadyClosedException;
}
// Every container specific exception from datatnode will be seen as
// StorageContainerException
boolean checkIfContainerToExclude(Throwable t) {
return t instanceof StorageContainerException;
}
private void setExceptionAndThrow(IOException ioe) throws IOException {
isException = true;
throw ioe;
}
void handleRetry(IOException exception) throws IOException {
RetryPolicy retryPolicy = retryPolicyMap
.get(HddsClientUtils.checkForException(exception).getClass());
if (retryPolicy == null) {
retryPolicy = retryPolicyMap.get(Exception.class);
}
handleRetry(exception, retryPolicy);
}
private void handleRetry(IOException exception, RetryPolicy retryPolicy)
throws IOException {
RetryPolicy.RetryAction action = null;
try {
action = retryPolicy.shouldRetry(exception, retryCount, 0, true);
} catch (Exception e) {
setExceptionAndThrow(new IOException(e));
}
if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) {
String msg = "";
if (action.reason != null) {
msg = "Retry request failed. " + action.reason;
//LOG.error(msg, exception);
}
setExceptionAndThrow(new IOException(msg, exception));
}
// Throw the exception if the thread is interrupted
if (Thread.currentThread().isInterrupted()) {
//LOG.warn("Interrupted while trying for retry");
setExceptionAndThrow(exception);
}
Preconditions.checkArgument(
action.action == RetryPolicy.RetryAction.RetryDecision.RETRY);
if (action.delayMillis > 0) {
try {
Thread.sleep(action.delayMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
IOException ioe = (IOException) new InterruptedIOException(
"Interrupted: action=" + action + ", retry policy=" + retryPolicy)
.initCause(e);
setExceptionAndThrow(ioe);
}
}
retryCount++;
}
}

[guohao-rosicky]

Ok, I'll change it

[guohao-rosicky]

Has been modified. @szetszwo Please take a look.Thanks.

[szetszwo]

@guohao-rosicky , please review it yourself -- go through the change and remove the unused code (e.g. ConsumerWithIOException). Thanks.

[szetszwo]

@guohao-rosicky , thanks for the update. Some comments inlined.

[szetszwo]

runCommandAsync is not used. Let's add it later.

[szetszwo]

Let's remove LOG.error(..) since it already throws an exception.

[szetszwo]

Add final.

[szetszwo]

This method is not used anywhere. Let's remove it.

[szetszwo]

We should pass the retryPolicyMap to the super constructor.

[szetszwo]

The map is passed incorrectly. Below is the original map.

- this.retryPolicyMap = HddsClientUtils.getRetryPolicyByException(
- config.getMaxRetryCount(), config.getRetryInterval());

[captainzmc]

+1 Thanks @guohao-rosicky. The change looks good.

[szetszwo]

+1 the change looks good.

[guohao-rosicky]

What changes were proposed in this pull request?

Optimized the putBlock method of BlockDataStreamOutput

1. PutBlock does not use RaftRpc, but through the stream close() method

// putBlock metadata
ByteBuffer buf =
ContainerCommandRequestMessage.toMessage(putBlockRequestProto, null)
.getContent().asReadOnlyByteBuffer();
out.writeAsync(buf, StandardWriteOption.CLOSE); 

2.Mark the stream data boundary by sending an empty stream packet

// Mark the stream data boundary
out.writeAsync(ByteBuffer.allocateDirect(0).asReadOnlyBuffer()); 

3. Flush is also in the form of async RPC, because putBlock needs to fetch bcsId (Raft log index)

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-6137

How was this patch tested?

Use the existing UT

[captainzmc]

Thanks @guohao-rosicky Improve this. LGTM over all. TestBlockDataStreamOutput has some error, can you confirm?

[szetszwo]

@guohao-rosicky, Writing an empty buffer may not work as expected. In the streaming model, data is continuous but not block-by-block. Both sender and receiver must consider that the data is a stream of bytes.

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

Let me think about how to make it work.

[guohao-rosicky]

Thanks for your comment @szetszwo . I wrote a piece of pseudo-code, Ideas about the modification of org.apache.ratis.netty.server.DataStreamManagement#writeTo.

What do you think of this idea? There is no need to send this empty buffer rpc.

[szetszwo]

@guohao-rosicky , this is a standard problem in streaming. The standard solution is to write the length with the data. Let's don't change the API for it.

[guohao-rosicky]

Hi, @szetszwo
I have a question, about

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

I noticed that in HDFS, marking the end of stream is also represented by an empty package.
In Ratis Stream, sticky packets have already been dealt with.

org.apache.ratis.netty.NettyDataStreamUtils#encodeDataStreamRequestHeader
https://github.com/apache/ratis/blob/master/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java#L90

@captainzmc tested the code from some time ago and did not find this problem

[szetszwo]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

[guohao-rosicky]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

Thanks @szetszwo, Patch has been applied

[szetszwo]

@guohao-rosicky , please fix the checkstyle warnings and test the change manually. I have not tested it. Thanks.

[captainzmc]

Thanks @guohao-rosicky@szetszwo update this. I used this PR test and found that the file failed to write. In DataNode IndexOutOfBoundsException will appear.

[szetszwo]

@captainzmc , thanks. Let me test the serialization/deserialization code.

[szetszwo]

Found the bug: ByteBu.nioBuffers(index, length) somehow is not working. Let me fix it.

[szetszwo]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

[captainzmc]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

Thanks @szetszwo for the update. Using new patch I found IndexOutOfBoundsException still appear.
I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

[szetszwo]

@captainzmc , thanks a lot for testing it!

I added one more test for the serialization and buffers. It worked fine. See https://issues.apache.org/jira/secure/attachment/13041275/3195_testBuffers.patch

I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

It seems there are some synchronization problems. Let me check.

[szetszwo]

@captainzmc , could you post the test code?

[captainzmc]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Actually there a problem with my previous description. I noticed that the when I use 30 threads writing, we’ll get same error.

[captainzmc]

@szetszwo Here is my datanode log. https://issues.apache.org/jira/secure/attachment/13041276/ozone-root-datanode-9-29-173-57.log
In the test, I wrote 30 files concurrently. Each file is 128MB and the chunk size is 1MB

[szetszwo]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

[szetszwo]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Thanks for sharing the test code. It looks good.

[captainzmc]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

Thanks @szetszwo for update this. I had just test this patch, we still get same error. Just as the same as this.

[szetszwo]

Let's add an abstract class instead static methods. Then, KeyDataStreamOutput can extends it. If we need to add SmallFileDataStreamOutput, it can extends it too.

package org.apache.hadoop.ozone.client.io;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.protocol.exceptions.RaftRetryFailureException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
public abstract class AbstractDataStreamOutput implements ByteBufferStreamOutput {
private final Map<Class<? extends Throwable>, RetryPolicy> retryPolicyMap;
private int retryCount;
private boolean isException;
protected AbstractDataStreamOutput() {
this.retryPolicyMap = HddsClientUtils.getExceptionList()
.stream()
.collect(Collectors.toMap(Function.identity(),
e -> RetryPolicies.TRY_ONCE_THEN_FAIL));
}
@VisibleForTesting
public int getRetryCount() {
return retryCount;
}
void resetRetryCount() {
retryCount = 0;
}
boolean isException() {
return isException;
}
/**
* Checks if the provided exception signifies retry failure in ratis client.
* In case of retry failure, ratis client throws RaftRetryFailureException
* and all succeeding operations are failed with AlreadyClosedException.
*/
boolean checkForRetryFailure(Throwable t) {
return t instanceof RaftRetryFailureException
|| t instanceof AlreadyClosedException;
}
// Every container specific exception from datatnode will be seen as
// StorageContainerException
boolean checkIfContainerToExclude(Throwable t) {
return t instanceof StorageContainerException;
}
private void setExceptionAndThrow(IOException ioe) throws IOException {
isException = true;
throw ioe;
}
void handleRetry(IOException exception) throws IOException {
RetryPolicy retryPolicy = retryPolicyMap
.get(HddsClientUtils.checkForException(exception).getClass());
if (retryPolicy == null) {
retryPolicy = retryPolicyMap.get(Exception.class);
}
handleRetry(exception, retryPolicy);
}
private void handleRetry(IOException exception, RetryPolicy retryPolicy)
throws IOException {
RetryPolicy.RetryAction action = null;
try {
action = retryPolicy.shouldRetry(exception, retryCount, 0, true);
} catch (Exception e) {
setExceptionAndThrow(new IOException(e));
}
if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) {
String msg = "";
if (action.reason != null) {
msg = "Retry request failed. " + action.reason;
//LOG.error(msg, exception);
}
setExceptionAndThrow(new IOException(msg, exception));
}
// Throw the exception if the thread is interrupted
if (Thread.currentThread().isInterrupted()) {
//LOG.warn("Interrupted while trying for retry");
setExceptionAndThrow(exception);
}
Preconditions.checkArgument(
action.action == RetryPolicy.RetryAction.RetryDecision.RETRY);
if (action.delayMillis > 0) {
try {
Thread.sleep(action.delayMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
IOException ioe = (IOException) new InterruptedIOException(
"Interrupted: action=" + action + ", retry policy=" + retryPolicy)
.initCause(e);
setExceptionAndThrow(ioe);
}
}
retryCount++;
}
}

[guohao-rosicky]

Ok, I'll change it

[guohao-rosicky]

Has been modified. @szetszwo Please take a look.Thanks.

[szetszwo]

@guohao-rosicky , please review it yourself -- go through the change and remove the unused code (e.g. ConsumerWithIOException). Thanks.

[szetszwo]

@guohao-rosicky , thanks for the update. Some comments inlined.

[szetszwo]

runCommandAsync is not used. Let's add it later.

[szetszwo]

Let's remove LOG.error(..) since it already throws an exception.

[szetszwo]

Add final.

[szetszwo]

This method is not used anywhere. Let's remove it.

[szetszwo]

We should pass the retryPolicyMap to the super constructor.

[szetszwo]

The map is passed incorrectly. Below is the original map.

- this.retryPolicyMap = HddsClientUtils.getRetryPolicyByException(
- config.getMaxRetryCount(), config.getRetryInterval());

[captainzmc]

+1 Thanks @guohao-rosicky. The change looks good.

[szetszwo]

+1 the change looks good.

[guohao-rosicky]

What changes were proposed in this pull request?

Optimized the putBlock method of BlockDataStreamOutput

1. PutBlock does not use RaftRpc, but through the stream close() method

// putBlock metadata
ByteBuffer buf =
ContainerCommandRequestMessage.toMessage(putBlockRequestProto, null)
.getContent().asReadOnlyByteBuffer();
out.writeAsync(buf, StandardWriteOption.CLOSE); 

2.Mark the stream data boundary by sending an empty stream packet

// Mark the stream data boundary
out.writeAsync(ByteBuffer.allocateDirect(0).asReadOnlyBuffer()); 

3. Flush is also in the form of async RPC, because putBlock needs to fetch bcsId (Raft log index)

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-6137

How was this patch tested?

Use the existing UT

[captainzmc]

Thanks @guohao-rosicky Improve this. LGTM over all. TestBlockDataStreamOutput has some error, can you confirm?

[szetszwo]

@guohao-rosicky, Writing an empty buffer may not work as expected. In the streaming model, data is continuous but not block-by-block. Both sender and receiver must consider that the data is a stream of bytes.

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

Let me think about how to make it work.

[guohao-rosicky]

Thanks for your comment @szetszwo . I wrote a piece of pseudo-code, Ideas about the modification of org.apache.ratis.netty.server.DataStreamManagement#writeTo.

What do you think of this idea? There is no need to send this empty buffer rpc.

[szetszwo]

@guohao-rosicky , this is a standard problem in streaming. The standard solution is to write the length with the data. Let's don't change the API for it.

[guohao-rosicky]

Hi, @szetszwo
I have a question, about

For example, the sender may write 3 times with lengths 3, 4, 5. Since there could be buffers/encoder/decoder in between, the receiver may receive a single buffer with length 12.

I noticed that in HDFS, marking the end of stream is also represented by an empty package.
In Ratis Stream, sticky packets have already been dealt with.

org.apache.ratis.netty.NettyDataStreamUtils#encodeDataStreamRequestHeader
https://github.com/apache/ratis/blob/master/ratis-netty/src/main/java/org/apache/ratis/netty/NettyDataStreamUtils.java#L90

@captainzmc tested the code from some time ago and did not find this problem

[szetszwo]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

[guohao-rosicky]

@guohao-rosicky , the idea is to write the proto length at the end as shown below; see also https://issues.apache.org/jira/secure/attachment/13041230/3195_review.patch

 // readerIndex protoIndex lengthIndex readerIndex+readableBytes
// V V V V
// format: |--- data ---|--- proto ---|--- proto length (4 bytes) ---|

Thanks @szetszwo, Patch has been applied

[szetszwo]

@guohao-rosicky , please fix the checkstyle warnings and test the change manually. I have not tested it. Thanks.

[captainzmc]

Thanks @guohao-rosicky@szetszwo update this. I used this PR test and found that the file failed to write. In DataNode IndexOutOfBoundsException will appear.

[szetszwo]

@captainzmc , thanks. Let me test the serialization/deserialization code.

[szetszwo]

Found the bug: ByteBu.nioBuffers(index, length) somehow is not working. Let me fix it.

[szetszwo]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

[captainzmc]

@captainzmc , @guohao-rosicky, Fixed the bug; see https://issues.apache.org/jira/secure/attachment/13041255/3195_TestStreamDataChannel.patch

Thanks @szetszwo for the update. Using new patch I found IndexOutOfBoundsException still appear.
I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

[szetszwo]

@captainzmc , thanks a lot for testing it!

I added one more test for the serialization and buffers. It worked fine. See https://issues.apache.org/jira/secure/attachment/13041275/3195_testBuffers.patch

I found that the ERROR only appeared when writing with high concurrency. I was successful with 30 concurrent writes, but I get the exception with 600 concurrent writes.

It seems there are some synchronization problems. Let me check.

[szetszwo]

@captainzmc , could you post the test code?

[captainzmc]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Actually there a problem with my previous description. I noticed that the when I use 30 threads writing, we’ll get same error.

[captainzmc]

@szetszwo Here is my datanode log. https://issues.apache.org/jira/secure/attachment/13041276/ozone-root-datanode-9-29-173-57.log
In the test, I wrote 30 files concurrently. Each file is 128MB and the chunk size is 1MB

[szetszwo]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

[szetszwo]

@szetszwo yes here is my test code. Is written according to the ratis Filestore test.

Thanks for sharing the test code. It looks good.

[captainzmc]

@captainzmc , @guohao-rosicky , just found a bug in BlockDataStreamOutput. Could you test https://issues.apache.org/jira/secure/attachment/13041279/3195_bugfix.patch to see if it can fix the bug?

Thanks @szetszwo for update this. I had just test this patch, we still get same error. Just as the same as this.

[szetszwo]

Let's add an abstract class instead static methods. Then, KeyDataStreamOutput can extends it. If we need to add SmallFileDataStreamOutput, it can extends it too.

package org.apache.hadoop.ozone.client.io;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput;
import org.apache.hadoop.io.retry.RetryPolicies;
import org.apache.hadoop.io.retry.RetryPolicy;
import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
import org.apache.ratis.protocol.exceptions.RaftRetryFailureException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
public abstract class AbstractDataStreamOutput implements ByteBufferStreamOutput {
private final Map<Class<? extends Throwable>, RetryPolicy> retryPolicyMap;
private int retryCount;
private boolean isException;
protected AbstractDataStreamOutput() {
this.retryPolicyMap = HddsClientUtils.getExceptionList()
.stream()
.collect(Collectors.toMap(Function.identity(),
e -> RetryPolicies.TRY_ONCE_THEN_FAIL));
}
@VisibleForTesting
public int getRetryCount() {
return retryCount;
}
void resetRetryCount() {
retryCount = 0;
}
boolean isException() {
return isException;
}
/**
* Checks if the provided exception signifies retry failure in ratis client.
* In case of retry failure, ratis client throws RaftRetryFailureException
* and all succeeding operations are failed with AlreadyClosedException.
*/
boolean checkForRetryFailure(Throwable t) {
return t instanceof RaftRetryFailureException
|| t instanceof AlreadyClosedException;
}
// Every container specific exception from datatnode will be seen as
// StorageContainerException
boolean checkIfContainerToExclude(Throwable t) {
return t instanceof StorageContainerException;
}
private void setExceptionAndThrow(IOException ioe) throws IOException {
isException = true;
throw ioe;
}
void handleRetry(IOException exception) throws IOException {
RetryPolicy retryPolicy = retryPolicyMap
.get(HddsClientUtils.checkForException(exception).getClass());
if (retryPolicy == null) {
retryPolicy = retryPolicyMap.get(Exception.class);
}
handleRetry(exception, retryPolicy);
}
private void handleRetry(IOException exception, RetryPolicy retryPolicy)
throws IOException {
RetryPolicy.RetryAction action = null;
try {
action = retryPolicy.shouldRetry(exception, retryCount, 0, true);
} catch (Exception e) {
setExceptionAndThrow(new IOException(e));
}
if (action.action == RetryPolicy.RetryAction.RetryDecision.FAIL) {
String msg = "";
if (action.reason != null) {
msg = "Retry request failed. " + action.reason;
//LOG.error(msg, exception);
}
setExceptionAndThrow(new IOException(msg, exception));
}
// Throw the exception if the thread is interrupted
if (Thread.currentThread().isInterrupted()) {
//LOG.warn("Interrupted while trying for retry");
setExceptionAndThrow(exception);
}
Preconditions.checkArgument(
action.action == RetryPolicy.RetryAction.RetryDecision.RETRY);
if (action.delayMillis > 0) {
try {
Thread.sleep(action.delayMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
IOException ioe = (IOException) new InterruptedIOException(
"Interrupted: action=" + action + ", retry policy=" + retryPolicy)
.initCause(e);
setExceptionAndThrow(ioe);
}
}
retryCount++;
}
}

[guohao-rosicky]

Ok, I'll change it

[guohao-rosicky]

Has been modified. @szetszwo Please take a look.Thanks.

[szetszwo]

@guohao-rosicky , please review it yourself -- go through the change and remove the unused code (e.g. ConsumerWithIOException). Thanks.

[szetszwo]

@guohao-rosicky , thanks for the update. Some comments inlined.

[szetszwo]

runCommandAsync is not used. Let's add it later.

[szetszwo]

Let's remove LOG.error(..) since it already throws an exception.

[szetszwo]

Add final.

[szetszwo]

This method is not used anywhere. Let's remove it.

[szetszwo]

We should pass the retryPolicyMap to the super constructor.

[szetszwo]

The map is passed incorrectly. Below is the original map.

- this.retryPolicyMap = HddsClientUtils.getRetryPolicyByException(
- config.getMaxRetryCount(), config.getRetryInterval());

[captainzmc]

+1 Thanks @guohao-rosicky. The change looks good.

[szetszwo]

+1 the change looks good.