szetszwo commented on code in PR #11767:
URL: https://github.com/apache/ozone/pull/11767#discussion_r4380624844
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java:
##########
@@ -348,6 +348,16 @@ public class OzoneClientConfig {
tags = ConfigTag.CLIENT)
private boolean datastreamPutBlockWithoutRaftEnabled = false;
+ @Config(key = "ozone.client.datastream.hsync.enabled",
+ defaultValue = "false",
+ type = ConfigType.BOOLEAN,
+ description = "When enabled, hsync of a Ratis data stream makes the
datanodes sync the block data and " +
+ "the PutBlock to disk before replying. Otherwise, hsync returns once
the datanodes have written them " +
+ "without sync. It takes effect only when hsync is enabled
(ozone.fs.hsync.enabled) and the PutBlock " +
+ "is committed through the data stream
(ozone.client.datastream.putblock.without.raft.enabled).",
+ tags = ConfigTag.CLIENT)
+ private boolean datastreamHsyncEnabled = false;
Review Comment:
It should not be configurable -- hsync should always wait.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java:
##########
@@ -534,13 +547,21 @@ public void executePutBlock(boolean close,
* so far and the datanodes reply with the serialized {@link
ContainerCommandResponseProto}.
*
* @param blockData the block metadata to commit
- * @param byteBufferList the buffers covered by this PutBlock, to be
released once it is acknowledged
+ * @param byteBufferList the buffers covered by this PutBlock, to be
released once it is acknowledged;
+ * null if no data was written since the previous
PutBlock
+ * @param sync whether the datanodes sync the data and the PutBlock to disk
before replying
*/
private void executePutBlockCommand(BlockData blockData,
- List<StreamBuffer> byteBufferList) throws IOException {
- final ContainerCommandRequestProto putBlockRequest
+ List<StreamBuffer> byteBufferList, boolean sync) throws IOException {
+ ContainerCommandRequestProto putBlockRequest
= ContainerProtocolCalls.getPutBlockRequest(
xceiverClient.getPipeline(), blockData, false, tokenString);
+ if (sync) {
+ final ContainerCommandRequestProto.Builder builder =
putBlockRequest.toBuilder();
+ builder.getPutBlockBuilder().setSync(true);
+ putBlockRequest = builder.build();
+ }
Review Comment:
We should add a `sync` parameter to getPutBlockRequest instead of rebuilding
it.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java:
##########
@@ -434,6 +436,17 @@ public void watchForCommit(boolean bufferFull) throws
IOException {
*/
public void executePutBlock(boolean close,
boolean force) throws IOException {
+ executePutBlock(close, force, false);
+ }
+
+ /**
+ * @param close whether putBlock is happening as part of closing the stream
+ * @param force true if no data was written since most recent putBlock and
+ * either the stream is being closed or the previous putBlock is
to be synced
+ * @param sync whether the datanodes sync the data and the PutBlock to disk
before replying;
+ * it applies only to a PutBlock committed in the middle of the
stream without Raft
+ */
+ private void executePutBlock(boolean close, boolean force, boolean sync)
throws IOException {
Review Comment:
With sync, it seems that `force` is not longer needed.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java:
##########
@@ -699,6 +699,11 @@ ContainerCommandResponseProto handlePutBlock(
dispatcherContext == null ? 0 : dispatcherContext.getLogIndex();
blockData.setBlockCommitSequenceId(bcsId);
blockManager.putBlock(kvContainer, blockData, endOfBlock);
+ if (request.getPutBlock().getSync()) {
+ try (DBHandle db = BlockUtils.getDB(kvContainer.getContainerData(),
conf)) {
+ db.getStore().flushLog(true);
+ }
Review Comment:
Call kvContainer.flushAndSyncDB(). Need to remove `private`.
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java:
##########
@@ -699,6 +699,11 @@ ContainerCommandResponseProto handlePutBlock(
dispatcherContext == null ? 0 : dispatcherContext.getLogIndex();
blockData.setBlockCommitSequenceId(bcsId);
blockManager.putBlock(kvContainer, blockData, endOfBlock);
+ if (request.getPutBlock().getSync()) {
+ try (DBHandle db = BlockUtils.getDB(kvContainer.getContainerData(),
conf)) {
+ db.getStore().flushLog(true);
+ }
Review Comment:
Call kvContainer.flushAndSyncDB(). Need to remove `private`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
