HDDS-3155. Improved ozone client flush implementation to make it faster.

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java

@@ -85,7 +85,9 @@ public class BlockOutputStream extends OutputStream {
   private final int bytesPerChecksum;
   private int chunkIndex;
   private final AtomicLong chunkOffset = new AtomicLong();
+  private final int streamBufferSize;
   private final long streamBufferFlushSize;
+  private final boolean streamBufferFlushDelay;
   private final long streamBufferMaxSize;
   private final BufferPool bufferPool;
   // The IOException will be set by response handling thread in case there is an
@@ -131,10 +133,10 @@ public class BlockOutputStream extends OutputStream {
   @SuppressWarnings("parameternumber")
   public BlockOutputStream(BlockID blockID,
       XceiverClientManager xceiverClientManager, Pipeline pipeline,
-      long streamBufferFlushSize, long streamBufferMaxSize,
+      int streamBufferSize, long streamBufferFlushSize,
+      boolean streamBufferFlushDelay, long streamBufferMaxSize,
       BufferPool bufferPool, ChecksumType checksumType,
-      int bytesPerChecksum)
-      throws IOException {
+      int bytesPerChecksum) throws IOException {
     this.blockID = new AtomicReference<>(blockID);
     KeyValue keyValue =
         KeyValue.newBuilder().setKey("TYPE").setValue("KEY").build();
@@ -143,8 +145,10 @@ public BlockOutputStream(BlockID blockID,
             .addMetadata(keyValue);
     this.xceiverClientManager = xceiverClientManager;
     this.xceiverClient = xceiverClientManager.acquireClient(pipeline);
+    this.streamBufferSize = streamBufferSize;
     this.streamBufferFlushSize = streamBufferFlushSize;
     this.streamBufferMaxSize = streamBufferMaxSize;
+    this.streamBufferFlushDelay = streamBufferFlushDelay;
     this.bufferPool = bufferPool;
     this.bytesPerChecksum = bytesPerChecksum;
 
@@ -434,7 +438,9 @@ ContainerCommandResponseProto> executePutBlock(boolean close,
   @Override
   public void flush() throws IOException {
     if (xceiverClientManager != null && xceiverClient != null
-        && bufferPool != null && bufferPool.getSize() > 0) {
+        && bufferPool != null && bufferPool.getSize() > 0
+        && (!streamBufferFlushDelay ||
+            writtenDataLength - totalDataFlushedLength >= streamBufferSize)) {
       try {
         handleFlush(false);
       } catch (InterruptedException | ExecutionException e) {
@@ -447,7 +453,6 @@ public void flush() throws IOException {
     }
   }
 
-
   private void writeChunk(ChunkBuffer buffer)
       throws IOException {
     // This data in the buffer will be pushed to datanode and a reference will

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java

@@ -149,6 +149,17 @@ public final class OzoneConfigKeys {
   public static final TimeDuration OZONE_CLIENT_RETRY_INTERVAL_DEFAULT =
       TimeDuration.valueOf(0, TimeUnit.MILLISECONDS);
 
+  /**
+   * If this value is true, when the client calls the flush() method,
+   * we will checks whether the data in the buffer is greater than
+   * OZONE_CLIENT_STREAM_BUFFER_SIZE_DEFAULT. If greater than,
+   * send the data in the buffer to the datanode.
+   * */
+  public static final String OZONE_CLIENT_STREAM_BUFFER_FLUSH_DELAY =
+      "ozone.client.stream.buffer.flush.delay";
+  public static final boolean OOZONE_CLIENT_STREAM_BUFFER_FLUSH_DELAY_DEFAULT =
+      false;
+
   // This defines the overall connection limit for the connection pool used in
   // RestClient.
   public static final String OZONE_REST_CLIENT_HTTP_CONNECTION_MAX =

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -386,6 +386,15 @@
     <description>Connection timeout for Ozone client in milliseconds.
     </description>
   </property>
+  <property>
+    <name>ozone.client.stream.buffer.flush.delay</name>
+    <value>false</value>
+    <tag>OZONE, CLIENT</tag>
+    <description>If set true, when call flush()，we'll determine whether the
+      data in the current buffer is greater than ozone.client.stream.buffer.size.
+      if greater than it we'll sent buffer to the datanode.
+    </description>
+  </property>
   <property>
     <name>ozone.client.stream.buffer.size</name>
     <value>4MB</value>

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntry.java

@@ -55,7 +55,9 @@ public final class BlockOutputStreamEntry extends OutputStream {
   private long currentPosition;
   private Token<OzoneBlockTokenIdentifier> token;
 
+  private final int streamBufferSize;
   private final long streamBufferFlushSize;
+  private final boolean streamBufferFlushDelay;
   private final long streamBufferMaxSize;
   private final long watchTimeout;
   private BufferPool bufferPool;
@@ -64,7 +66,8 @@ public final class BlockOutputStreamEntry extends OutputStream {
   private BlockOutputStreamEntry(BlockID blockID, String key,
       XceiverClientManager xceiverClientManager,
       Pipeline pipeline, String requestId, int chunkSize,
-      long length, long streamBufferFlushSize, long streamBufferMaxSize,
+      long length, int streamBufferSize, long streamBufferFlushSize,
+      boolean streamBufferFlushDelay, long streamBufferMaxSize,
       long watchTimeout, BufferPool bufferPool,
       ChecksumType checksumType, int bytesPerChecksum,
       Token<OzoneBlockTokenIdentifier> token) {
@@ -77,7 +80,9 @@ private BlockOutputStreamEntry(BlockID blockID, String key,
     this.token = token;
     this.length = length;
     this.currentPosition = 0;
+    this.streamBufferSize = streamBufferSize;
     this.streamBufferFlushSize = streamBufferFlushSize;
+    this.streamBufferFlushDelay = streamBufferFlushDelay;
     this.streamBufferMaxSize = streamBufferMaxSize;
     this.watchTimeout = watchTimeout;
     this.bufferPool = bufferPool;
@@ -110,9 +115,9 @@ private void checkStream() throws IOException {
       }
       this.outputStream =
           new BlockOutputStream(blockID, xceiverClientManager,
-              pipeline, streamBufferFlushSize,
-              streamBufferMaxSize, bufferPool, checksumType,
-              bytesPerChecksum);
+              pipeline, streamBufferSize, streamBufferFlushSize,
+              streamBufferFlushDelay, streamBufferMaxSize, bufferPool,
+              checksumType, bytesPerChecksum);
     }
   }
 
@@ -215,7 +220,9 @@ public static class Builder {
     private String requestId;
     private int chunkSize;
     private long length;
+    private int  streamBufferSize;
     private long streamBufferFlushSize;
+    private boolean streamBufferFlushDelay;
     private long streamBufferMaxSize;
     private long watchTimeout;
     private BufferPool bufferPool;
@@ -269,11 +276,21 @@ public Builder setLength(long len) {
       return this;
     }
 
+    public Builder setStreamBufferSize(int bufferSize) {
+      this.streamBufferSize = bufferSize;
+      return this;
+    }
+
     public Builder setStreamBufferFlushSize(long bufferFlushSize) {
       this.streamBufferFlushSize = bufferFlushSize;
       return this;
     }
 
+    public Builder setStreamBufferFlushDelay(boolean bufferFlushDelay) {
+      this.streamBufferFlushDelay = bufferFlushDelay;
+      return this;
+    }
+
     public Builder setStreamBufferMaxSize(long bufferMaxSize) {
       this.streamBufferMaxSize = bufferMaxSize;
       return this;
@@ -297,7 +314,8 @@ public Builder setToken(Token<OzoneBlockTokenIdentifier> bToken) {
     public BlockOutputStreamEntry build() {
       return new BlockOutputStreamEntry(blockID, key,
           xceiverClientManager, pipeline, requestId, chunkSize,
-          length, streamBufferFlushSize, streamBufferMaxSize, watchTimeout,
+          length, streamBufferSize, streamBufferFlushSize,
+          streamBufferFlushDelay, streamBufferMaxSize, watchTimeout,
           bufferPool, checksumType, bytesPerChecksum, token);
     }
   }
@@ -331,10 +349,18 @@ public long getCurrentPosition() {
     return currentPosition;
   }
 
+  public int getStreamBufferSize() {
+    return streamBufferSize;
+  }
+
   public long getStreamBufferFlushSize() {
     return streamBufferFlushSize;
   }
 
+  public boolean getStreamBufferFlushDelay() {
+    return streamBufferFlushDelay;
+  }
+
   public long getStreamBufferMaxSize() {
     return streamBufferMaxSize;
   }

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockOutputStreamEntryPool.java

@@ -61,6 +61,7 @@ public class BlockOutputStreamEntryPool {
   private final String requestID;
   private final int streamBufferSize;
   private final long streamBufferFlushSize;
+  private final boolean streamBufferFlushDelay;
   private final long streamBufferMaxSize;
   private final long watchTimeout;
   private final long blockSize;
@@ -75,7 +76,8 @@ public class BlockOutputStreamEntryPool {
   public BlockOutputStreamEntryPool(OzoneManagerProtocol omClient,
       int chunkSize, String requestId, HddsProtos.ReplicationFactor factor,
       HddsProtos.ReplicationType type,
-      int bufferSize, long bufferFlushSize, long bufferMaxSize,
+      int bufferSize, long bufferFlushSize,
+      boolean bufferFlushDelay, long bufferMaxSize,
       long size, long watchTimeout, ContainerProtos.ChecksumType checksumType,
       int bytesPerChecksum, String uploadID, int partNumber,
       boolean isMultipart, OmKeyInfo info,
@@ -93,6 +95,7 @@ public BlockOutputStreamEntryPool(OzoneManagerProtocol omClient,
     this.requestID = requestId;
     this.streamBufferSize = bufferSize;
     this.streamBufferFlushSize = bufferFlushSize;
+    this.streamBufferFlushDelay = bufferFlushDelay;
     this.streamBufferMaxSize = bufferMaxSize;
     this.blockSize = size;
     this.watchTimeout = watchTimeout;
@@ -137,6 +140,7 @@ public BlockOutputStreamEntryPool(OzoneManagerProtocol omClient,
     requestID = null;
     streamBufferSize = 0;
     streamBufferFlushSize = 0;
+    streamBufferFlushDelay = false;
     streamBufferMaxSize = 0;
     bufferPool = new BufferPool(chunkSize, 1);
     watchTimeout = 0;
@@ -188,7 +192,9 @@ private void addKeyLocationInfo(OmKeyLocationInfo subKeyInfo)
             .setRequestId(requestID)
             .setChunkSize(chunkSize)
             .setLength(subKeyInfo.getLength())
+            .setStreamBufferSize(streamBufferSize)
             .setStreamBufferFlushSize(streamBufferFlushSize)
+            .setStreamBufferFlushDelay(streamBufferFlushDelay)
             .setStreamBufferMaxSize(streamBufferMaxSize)
             .setWatchTimeout(watchTimeout)
             .setbufferPool(bufferPool)

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyOutputStream.java

@@ -124,15 +124,16 @@ public KeyOutputStream(OpenKeySession handler,
       XceiverClientManager xceiverClientManager,
       OzoneManagerProtocol omClient, int chunkSize,
       String requestId, ReplicationFactor factor, ReplicationType type,
-      int bufferSize, long bufferFlushSize, long bufferMaxSize,
-      long size, long watchTimeout,
+      int bufferSize, long bufferFlushSize, boolean isBufferFlushDelay,
+      long bufferMaxSize, long size, long watchTimeout,
       ChecksumType checksumType, int bytesPerChecksum,
       String uploadID, int partNumber, boolean isMultipart,
       int maxRetryCount, long retryInterval) {
     OmKeyInfo info = handler.getKeyInfo();
     blockOutputStreamEntryPool =
         new BlockOutputStreamEntryPool(omClient, chunkSize, requestId, factor,
-            type, bufferSize, bufferFlushSize, bufferMaxSize, size,
+            type, bufferSize, bufferFlushSize, isBufferFlushDelay,
+            bufferMaxSize, size,
             watchTimeout, checksumType, bytesPerChecksum, uploadID, partNumber,
             isMultipart, info, xceiverClientManager, handler.getId());
     // Retrieve the file encryption key info, null if file is not in
@@ -542,6 +543,7 @@ public static class Builder {
     private ReplicationFactor factor;
     private int streamBufferSize;
     private long streamBufferFlushSize;
+    private boolean streamBufferFlushDelay;
     private long streamBufferMaxSize;
     private long blockSize;
     private long watchTimeout;
@@ -608,6 +610,11 @@ public Builder setStreamBufferFlushSize(long size) {
       return this;
     }
 
+    public Builder setStreamBufferFlushDelay(boolean isDelay) {
+      this.streamBufferFlushDelay = isDelay;
+      return this;
+    }
+
     public Builder setStreamBufferMaxSize(long size) {
       this.streamBufferMaxSize = size;
       return this;
@@ -646,7 +653,8 @@ public Builder setRetryInterval(long retryIntervalInMS) {
     public KeyOutputStream build() {
       return new KeyOutputStream(openHandler, xceiverManager, omClient,
           chunkSize, requestID, factor, type,
-          streamBufferSize, streamBufferFlushSize, streamBufferMaxSize,
+          streamBufferSize, streamBufferFlushSize, streamBufferFlushDelay,
+          streamBufferMaxSize,
           blockSize, watchTimeout, checksumType,
           bytesPerChecksum, multipartUploadID, multipartNumber, isMultipartKey,
           maxRetryCount, retryInterval);

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -126,6 +126,7 @@ public class RpcClient implements ClientProtocol {
   private final ACLType groupRights;
   private final int streamBufferSize;
   private final long streamBufferFlushSize;
+  private boolean streamBufferFlushDelay;
   private final long streamBufferMaxSize;
   private final long blockSize;
   private final ClientId clientId = ClientId.randomId();
@@ -185,6 +186,9 @@ public RpcClient(Configuration conf, String omServiceId) throws IOException {
         .getStorageSize(OzoneConfigKeys.OZONE_CLIENT_STREAM_BUFFER_FLUSH_SIZE,
             OzoneConfigKeys.OZONE_CLIENT_STREAM_BUFFER_FLUSH_SIZE_DEFAULT,
             StorageUnit.BYTES);
+    streamBufferFlushDelay = conf.getBoolean(
+        OzoneConfigKeys.OZONE_CLIENT_STREAM_BUFFER_FLUSH_DELAY,
+        OzoneConfigKeys.OOZONE_CLIENT_STREAM_BUFFER_FLUSH_DELAY_DEFAULT);
     streamBufferMaxSize = (long) conf
         .getStorageSize(OzoneConfigKeys.OZONE_CLIENT_STREAM_BUFFER_MAX_SIZE,
             OzoneConfigKeys.OZONE_CLIENT_STREAM_BUFFER_MAX_SIZE_DEFAULT,
@@ -1192,6 +1196,7 @@ private OzoneOutputStream createOutputStream(OpenKeySession openKey,
             .setFactor(HddsProtos.ReplicationFactor.valueOf(factor.getValue()))
             .setStreamBufferSize(streamBufferSize)
             .setStreamBufferFlushSize(streamBufferFlushSize)
+            .setStreamBufferFlushDelay(streamBufferFlushDelay)
             .setStreamBufferMaxSize(streamBufferMaxSize)
             .setBlockSize(blockSize)
             .setChecksumType(checksumType)