Uh oh!
There was an error while loading. Please reload this page.
HDDS-4474. [Ozone-Streaming] Use WriteSmallFile to write small file. - #2860
HDDS-4474. [Ozone-Streaming] Use WriteSmallFile to write small file.#2860guohao-rosicky wants to merge 37 commits into
Conversation
guohao-rosicky
commented
Nov 23, 2021
@szetszwo@captainzmc please take a look, PutSmallFile contains writeChunk and putBlock, which is transmitted to Datanode through stream write. There is no applyTransaction that goes through containerStateMachine. Please check my PR,I don't know how to get the BCSID. |
szetszwo
left a comment
There was a problem hiding this comment.
@guohao-rosicky , thanks for the update. The change look mostly good. Some comments:
- Let's move the common code KeyValueStreamDataChannel and SmallFileStreamDataChannel to a common base class, say StreamDataChannelBase.
- Please add some tests.
See also https://issues.apache.org/jira/secure/attachment/13036532/2860_review.patch .
There was a problem hiding this comment.
Move randomAccessFile, file, containerData and metrics and the shared code to a new base class.
There was a problem hiding this comment.
Let's call it StreamDataChannel.
There was a problem hiding this comment.
Why changing data to optional?
There was a problem hiding this comment.
Data in PutSmallFileRequestProto is null in setupStream
ContainerProtos.PutSmallFileRequestProtoputSmallFileRequest =
ContainerProtos.PutSmallFileRequestProto.newBuilder()
.setChunkInfo(chunk)
.setBlock(createBlockRequest) // data is empty
.build();
Stringid = xceiverClient.getPipeline().getFirstNode().getUuidString();
ContainerProtos.ContainerCommandRequestProto.Builderbuilder =
ContainerProtos.ContainerCommandRequestProto.newBuilder()
.setCmdType(ContainerProtos.Type.StreamInit)
.setContainerID(blockID.get().getContainerID())
.setDatanodeUuid(id)
.setPutSmallFile(putSmallFileRequest);There was a problem hiding this comment.
is this backward compatible?
There was a problem hiding this comment.
Yes, changing required to optional is backward compatible since the old code always provides the value and the new code can handle it.
However, it is not forward compatible since the old code may not be able to handle the case that the value is missing.
szetszwo
commented
Nov 24, 2021
@guohao-rosicky, The change looks good but the new TestSmallFileDataStreamOutput failed. Please take a look. Thanks. |
captainzmc
commented
Nov 25, 2021
Thanks @guohao-rosicky for the contribution, |
szetszwo
commented
Nov 26, 2021
@captainzmc , this change is an optimization of createStreamKey(..) and createStreamFile(..) in RpcClient for the case that the given size is smaller than the chunk size. Using the Async API will have fewer RPC calls and we should implement it. However, the data have to go through the leader so that the network path won't be optimal and the leader may become a hotspot. |
Thanks @szetszwo for your explanation, agree with you. let's using Streaming instead of async write. |
szetszwo
left a comment
There was a problem hiding this comment.
@guohao-rosicky , we need support zero buffer copying so that we should not put the data inside PutSmallFileRequestProto. We should send the header and then send the raw data in the stream.
There was a problem hiding this comment.
This method copies buffer twice so that it is not zero buffer copying.
There was a problem hiding this comment.
We cannot call array() since the ByteBuffer may not have an array.
Also, using an array means there must be buffer copying inside the code. This is the reason that we use ByteBuffer but not byte[].
There was a problem hiding this comment.
Do not set the data inside the proto. We should send only the header in the proto and then send the raw data to the stream.
bshashikant
commented
Dec 2, 2021
@guohao-rosicky , can you please rebase? |
0501406 to
6f1e274Compareguohao-rosicky
commented
Dec 15, 2021
done |
6f1e274 to
a8600bbComparea8600bb to
89a8a2aCompare75c37a0 to
f86291cCompare8ce34b3 to
b0a5fe5Comparedd4d27d to
6537d0dCompare
captainzmc
left a comment
There was a problem hiding this comment.
Thanks @guohao-rosicky for the update. The change looks good.
guohao-rosicky
commented
Jan 20, 2022
@szetszwo@captainzmc please take a look. |
szetszwo
commented
Jan 22, 2022
@guohao-rosicky , the change grew from 13kB to 65kB so that it becomes hard to review. The SmallFileDataStreamOutput class is really long. How about we move the refactoring to a separated JIRA? |
How about splitting it into two parts @szetszwo
|
szetszwo
commented
Jan 24, 2022
@guohao-rosicky , sure, please do it. Thanks. |
6537d0d to
b0a7307Compareaa8903d to
76096cfCompareguohao-rosicky
commented
Mar 22, 2022
@szetszwo please take a look. Thanks. |
captainzmc
commented
Mar 23, 2022
The CI run was successful. @szetszwo Could you help take another look? |
szetszwo
commented
Mar 23, 2022
@captainzmc , @guohao-rosicky , sure, I am reviewing this. |
szetszwo
left a comment
There was a problem hiding this comment.
With HDDS-6137, we may not need to add much code to implement WriteSmallFile. The data is buffered at the client side. When the data size is small, all the data can be sent in a single write call with close. What do you think?
Uh oh!
There was an error while loading. Please reload this page.
| public static byte[] getFixedLengthBytes(int length) { | ||
| byte[] bytes = new byte[length]; | ||
| Random random = new Random(); | ||
| random.nextBytes(bytes); | ||
| return bytes; | ||
| } | ||
There was a problem hiding this comment.
Use ThreadLocalRandom and support non-random data as below:
public static byte[] generateData(int length, boolean random) {
final byte[] data = new byte[length];
if (random) {
ThreadLocalRandom.current().nextBytes(data);
} else {
for (int i = 0; i < length; i++) {
data[i] = (byte) i;
}
}
return data;
}
| private CompletableFuture<ContainerCommandResponseProto> link( | ||
| LogEntryProto entry, SmallFileStreamDataChannel smallFileChannel) { | ||
| return CompletableFuture.supplyAsync(() -> { | ||
| final DispatcherContext context = new DispatcherContext.Builder() | ||
| .setTerm(entry.getTerm()) | ||
| .setLogIndex(entry.getIndex()) | ||
| .setStage(DispatcherContext.WriteChunkStage.COMMIT_DATA) | ||
| .setContainer2BCSIDMap(container2BCSIDMap) | ||
| .build(); | ||
| return runCommand(smallFileChannel.getPutBlockRequest(), context); | ||
| }, executor); |
There was a problem hiding this comment.
Pass ContainerCommandRequestProto instead and rename it to runCommandAsync(..)
private CompletableFuture<ContainerCommandResponseProto> runCommandAsync(
ContainerCommandRequestProto requestProto, LogEntryProto entry) {
return CompletableFuture.supplyAsync(() -> {
final DispatcherContext context = new DispatcherContext.Builder()
.setTerm(entry.getTerm())
.setLogIndex(entry.getIndex())
.setStage(DispatcherContext.WriteChunkStage.COMMIT_DATA)
.setContainer2BCSIDMap(container2BCSIDMap)
.build();
return runCommand(requestProto, context);
}, executor);
}
| * <p> | ||
| * TODO : currently not support multi-thread access. | ||
| */ | ||
| public class SmallFileDataStreamOutput implements ByteBufferStreamOutput { |
There was a problem hiding this comment.
It seems that the code in this class is copied from BlockDataStreamOutputEntryPool and KeyDataStreamOutput. We should reuse the code but not copy them. Otherwise, it is very hard to maintain.
szetszwo
left a comment
There was a problem hiding this comment.
@guohao-rosicky , have you seen this comment #2860 (review) ?
guohao-rosicky
commented
Mar 24, 2022
@szetszwo |
That's why KeyDataStreamOutput is more powerful than SmallFileStreamOutput since it works even if the data size is unknown. |
@szetszwo Ok, how can we do the following work better? Does this PR code help us to achieve this function? I can split it. |
szetszwo
commented
Mar 24, 2022
Yes, I actually suggest you to split and move the common code to #3195 in this comment #3195 (comment) |
captainzmc
commented
Apr 11, 2022
Hi @szetszwo , will you take this task? If so, I will close the PR. |
szetszwo
commented
Apr 11, 2022
@captainzmc , I was thinking to support a BufferedDataStreamOutput in Ratis (similar to java.io.BufferedOutputStream). Then, Ozone could use it. It will require only a small change in Ozone. We may close this pull request and one a new pull request later. |
captainzmc
commented
Apr 11, 2022
OK, let's close this PR. |
Hi @szetszwo , Is it this ratis Jira? https://issues.apache.org/jira/browse/RATIS-1157 |
jira: https://issues.apache.org/jira/browse/HDDS-4474