HDDS-10488. Datanode OOM due to run out of mmap handler

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java

@@ -144,6 +144,10 @@ public final class ScmConfigKeys {
       "ozone.chunk.read.mapped.buffer.threshold";
   public static final String OZONE_CHUNK_READ_MAPPED_BUFFER_THRESHOLD_DEFAULT =
       "32KB";
+  public static final String OZONE_CHUNK_READ_MAPPED_BUFFER_MAX_COUNT_KEY =
+      "ozone.chunk.read.mapped.buffer.max.count";
+  // this max_count could not be greater than Linux platform max_map_count which by default is 65530.
+  public static final int OZONE_CHUNK_READ_MAPPED_BUFFER_MAX_COUNT_DEFAULT = 0;
 
   public static final String OZONE_SCM_CONTAINER_LAYOUT_KEY =
       "ozone.scm.container.layout";

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

@@ -22,6 +22,7 @@
 import java.nio.channels.GatheringByteChannel;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -60,13 +61,24 @@ static ChunkBuffer wrap(ByteBuffer buffer) {
     return new ChunkBufferImplWithByteBuffer(buffer);
   }
 
+  /** Wrap the given {@link ByteBuffer} as a {@link ChunkBuffer}, with a function called when buffer is released. */
+  static ChunkBuffer wrap(ByteBuffer buffer, Consumer<Integer> function) {
+    return new ChunkBufferImplWithByteBuffer(buffer, function);
+  }
+
   /** Wrap the given list of {@link ByteBuffer}s as a {@link ChunkBuffer}. */
   static ChunkBuffer wrap(List<ByteBuffer> buffers) {
+    return wrap(buffers, null);
+  }
+
+  /** Wrap the given list of {@link ByteBuffer}s as a {@link ChunkBuffer},
+   * with a function called when buffers are released.*/
+  static ChunkBuffer wrap(List<ByteBuffer> buffers, Consumer<Integer> function) {
     Objects.requireNonNull(buffers, "buffers == null");
     if (buffers.size() == 1) {
-      return wrap(buffers.get(0));
+      return wrap(buffers.get(0), function);
     }
-    return new ChunkBufferImplWithByteBufferList(buffers);
+    return new ChunkBufferImplWithByteBufferList(buffers, function);
   }
 
   /** Similar to {@link ByteBuffer#position()}. */

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

@@ -19,12 +19,14 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
 import java.nio.channels.GatheringByteChannel;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.Objects;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
@@ -34,14 +36,24 @@
 final class ChunkBufferImplWithByteBuffer implements ChunkBuffer {
   private final ByteBuffer buffer;
   private final UncheckedAutoCloseable underlying;
+  private final Consumer<Integer> releaseCallback;
 
   ChunkBufferImplWithByteBuffer(ByteBuffer buffer) {
-    this(buffer, null);
+    this.buffer = Objects.requireNonNull(buffer, "buffer == null");
+    this.releaseCallback = null;
+    this.underlying = null;
+  }
+
+  ChunkBufferImplWithByteBuffer(ByteBuffer buffer, Consumer<Integer> releaseFunction) {
+    this.buffer = Objects.requireNonNull(buffer, "buffer == null");
+    this.releaseCallback = releaseFunction;
+    this.underlying = null;
   }
 
   ChunkBufferImplWithByteBuffer(ByteBuffer buffer, UncheckedAutoCloseable underlying) {
     this.buffer = Objects.requireNonNull(buffer, "buffer == null");
     this.underlying = underlying;
+    this.releaseCallback = null;
   }
 
   @Override
@@ -163,4 +175,12 @@ public String toString() {
     return getClass().getSimpleName() + ":limit=" + buffer.limit()
         + "@" + Integer.toHexString(hashCode());
   }
+
+  @Override
+  protected void finalize() throws Throwable {
+    if (releaseCallback != null && buffer instanceof MappedByteBuffer) {
+      releaseCallback.accept(1);
+    }
+    super.finalize();
+  }
 }

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

@@ -20,6 +20,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 
+import java.nio.MappedByteBuffer;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
@@ -32,6 +33,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 /**
@@ -46,6 +48,7 @@ public class ChunkBufferImplWithByteBufferList implements ChunkBuffer {
   /** Buffer list backing the ChunkBuffer. */
   private final List<ByteBuffer> buffers;
   private final int limit;
+  private final Consumer<Integer> releaseCallback;
 
   private int limitPrecedingCurrent;
   private int currentIndex;
@@ -55,7 +58,16 @@ public class ChunkBufferImplWithByteBufferList implements ChunkBuffer {
     this.buffers = !buffers.isEmpty() ? ImmutableList.copyOf(buffers) :
         EMPTY_BUFFER;
     this.limit = buffers.stream().mapToInt(ByteBuffer::limit).sum();
+    this.releaseCallback = null;
+    findCurrent();
+  }
 
+  ChunkBufferImplWithByteBufferList(List<ByteBuffer> buffers, Consumer<Integer> releaseFunction) {
+    Objects.requireNonNull(buffers, "buffers == null");
+    this.buffers = !buffers.isEmpty() ? ImmutableList.copyOf(buffers) :
+        EMPTY_BUFFER;
+    this.limit = buffers.stream().mapToInt(ByteBuffer::limit).sum();
+    this.releaseCallback = releaseFunction;
     findCurrent();
   }
 
@@ -245,6 +257,15 @@ public String toString() {
         + ":l=" + limit();
   }
 
+  @Override
+  protected void finalize() throws Throwable {
+    // The assumption here is all the buffers in buffers list are of same buffer type, so only check the first buffer
+    if (releaseCallback != null && !buffers.isEmpty() && buffers.get(0) instanceof MappedByteBuffer) {
+      releaseCallback.accept(buffers.size());
+    }
+    super.finalize();
+  }
+
   private static int relativeToRange(int value, int min, int max) {
     final int pos;
     if (value <= min) {

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

@@ -860,6 +860,15 @@
       The default read threshold to use memory mapped buffers.
     </description>
   </property>
+  <property>
+    <name>ozone.chunk.read.mapped.buffer.max.count</name>
+    <value>0</value>
+    <tag>OZONE, SCM, CONTAINER, PERFORMANCE</tag>
+    <description>
+      The default max count of memory mapped buffers allowed for a DN.
+      Default 0 means no mapped buffers allowed for data read.
+    </description>
+  </property>
   <property>
     <name>ozone.scm.container.layout</name>
     <value>FILE_PER_BLOCK</value>

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/ChunkUtils.java

@@ -40,6 +40,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Semaphore;
 import java.util.function.Supplier;
 import java.util.function.ToLongFunction;
 
@@ -187,11 +188,12 @@ private static long writeDataToChannel(FileChannel channel, ChunkBuffer data,
     }
   }
 
+  @SuppressWarnings("checkstyle:parameternumber")
   public static ChunkBuffer readData(long len, int bufferCapacity,
-      File file, long off, HddsVolume volume, int readMappedBufferThreshold)
-      throws StorageContainerException {
-    if (len > readMappedBufferThreshold) {
-      return readData(file, bufferCapacity, off, len, volume);
+      File file, long off, HddsVolume volume, int readMappedBufferThreshold, boolean mmapEnabled,
+      Semaphore semaphore) throws StorageContainerException {
+    if (mmapEnabled && len > readMappedBufferThreshold && bufferCapacity > readMappedBufferThreshold) {
+      return readData(file, bufferCapacity, off, len, volume, semaphore);
     } else if (len == 0) {
       return ChunkBuffer.wrap(Collections.emptyList());
     }
@@ -252,25 +254,39 @@ private static void readData(File file, long offset, long len,
    * @return a list of {@link MappedByteBuffer} containing the data.
    */
   private static ChunkBuffer readData(File file, int chunkSize,
-      long offset, long length, HddsVolume volume)
+      long offset, long length, HddsVolume volume, Semaphore semaphore)
       throws StorageContainerException {
 
-    final List<ByteBuffer> buffers = new ArrayList<>(
-        Math.toIntExact((length - 1) / chunkSize) + 1);
-    readData(file, offset, length, channel -> {
-      long readLen = 0;
-      while (readLen < length) {
-        final int n = Math.toIntExact(Math.min(length - readLen, chunkSize));
-        final ByteBuffer mapped = channel.map(
-            FileChannel.MapMode.READ_ONLY, offset + readLen, n);
-        LOG.debug("mapped: offset={}, readLen={}, n={}, {}",
-            offset, readLen, n, mapped.getClass());
-        readLen += mapped.remaining();
-        buffers.add(mapped);
-      }
-      return readLen;
-    }, volume);
-    return ChunkBuffer.wrap(buffers);
+    final int bufferNum = Math.toIntExact((length - 1) / chunkSize) + 1;
+    if (!semaphore.tryAcquire(bufferNum)) {
+      // proceed with normal buffer
+      final ByteBuffer[] buffers = BufferUtils.assignByteBuffers(length,
+          chunkSize);
+      readData(file, offset, length, c -> c.position(offset).read(buffers), volume);
+      Arrays.stream(buffers).forEach(ByteBuffer::flip);
+      return ChunkBuffer.wrap(Arrays.asList(buffers));
+    } else {
+      // proceed with mapped buffer
+      LOG.debug("mapped buffer permits decreased by {} to total {}", bufferNum, semaphore.availablePermits());
+      final List<ByteBuffer> buffers = new ArrayList<>(bufferNum);
+      readData(file, offset, length, channel -> {
+        long readLen = 0;
+        while (readLen < length) {
+          final int n = Math.toIntExact(Math.min(length - readLen, chunkSize));
+          final ByteBuffer mapped = channel.map(
+              FileChannel.MapMode.READ_ONLY, offset + readLen, n);
+          LOG.debug("mapped: offset={}, readLen={}, n={}, {}",
+              offset, readLen, n, mapped.getClass());
+          readLen += mapped.remaining();
+          buffers.add(mapped);
+        }
+        return readLen;
+      }, volume);
+      return ChunkBuffer.wrap(buffers, permits -> {
+        semaphore.release(permits);
+        LOG.debug("mapped buffer permits increased by {} to total {}", permits, semaphore.availablePermits());
+      });
+    }
   }
 
   /**

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java

@@ -60,6 +60,7 @@ public class BlockManagerImpl implements BlockManager {
   // Default Read Buffer capacity when Checksum is not present
   private final int defaultReadBufferCapacity;
   private final int readMappedBufferThreshold;
+  private final int readMappedBufferMaxCount;
 
   /**
    * Constructs a Block Manager.
@@ -75,6 +76,9 @@ public BlockManagerImpl(ConfigurationSource conf) {
     this.readMappedBufferThreshold = config.getBufferSize(
         ScmConfigKeys.OZONE_CHUNK_READ_MAPPED_BUFFER_THRESHOLD_KEY,
         ScmConfigKeys.OZONE_CHUNK_READ_MAPPED_BUFFER_THRESHOLD_DEFAULT);
+    this.readMappedBufferMaxCount = config.getInt(
+        ScmConfigKeys.OZONE_CHUNK_READ_MAPPED_BUFFER_MAX_COUNT_KEY,
+        ScmConfigKeys.OZONE_CHUNK_READ_MAPPED_BUFFER_MAX_COUNT_DEFAULT);
   }
 
   @Override
@@ -264,6 +268,11 @@ public int getReadMappedBufferThreshold() {
     return readMappedBufferThreshold;
   }
 
+  /** @return the max count of memory mapped buffers for read. */
+  public int getReadMappedBufferMaxCount() {
+    return readMappedBufferMaxCount;
+  }
+
   /**
    * Deletes an existing block.
    * As Deletion is handled by BlockDeletingService,

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/FilePerBlockStrategy.java

@@ -54,6 +54,7 @@
 import java.nio.channels.FileChannel;
 import java.time.Duration;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Semaphore;
 
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.UNSUPPORTED_REQUEST;
 import static org.apache.hadoop.ozone.container.common.impl.ContainerLayoutVersion.FILE_PER_BLOCK;
@@ -75,6 +76,8 @@ public class FilePerBlockStrategy implements ChunkManager {
   private final OpenFiles files = new OpenFiles();
   private final int defaultReadBufferCapacity;
   private final int readMappedBufferThreshold;
+  private final int readMappedBufferMaxCount;
+  private final Semaphore semaphore;
   private final VolumeSet volumeSet;
 
   public FilePerBlockStrategy(boolean sync, BlockManager manager,
@@ -84,7 +87,15 @@ public FilePerBlockStrategy(boolean sync, BlockManager manager,
         manager.getDefaultReadBufferCapacity();
     this.readMappedBufferThreshold = manager == null ? 0
         : manager.getReadMappedBufferThreshold();
+    this.readMappedBufferMaxCount = manager == null ? 0
+        : manager.getReadMappedBufferMaxCount();
+    LOG.info("ozone.chunk.read.mapped.buffer.max.count is load with {}", readMappedBufferMaxCount);
     this.volumeSet = volSet;
+    if (this.readMappedBufferMaxCount > 0) {
+      semaphore = new Semaphore(this.readMappedBufferMaxCount);
+    } else {
+      semaphore = null;
+    }
   }
 
   private static void checkLayoutVersion(Container container) {
@@ -192,10 +203,10 @@ public ChunkBuffer readChunk(Container container, BlockID blockID,
 
     final long len = info.getLen();
     long offset = info.getOffset();
-    int bufferCapacity =  ChunkManager.getBufferCapacityForChunkRead(info,
+    int bufferCapacity = ChunkManager.getBufferCapacityForChunkRead(info,
         defaultReadBufferCapacity);
     return ChunkUtils.readData(len, bufferCapacity, chunkFile, offset, volume,
-        readMappedBufferThreshold);
+        readMappedBufferThreshold, readMappedBufferMaxCount > 0, semaphore);
   }
 
   @Override

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/FilePerChunkStrategy.java

@@ -50,6 +50,7 @@
 import java.nio.file.StandardCopyOption;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Semaphore;
 
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.UNSUPPORTED_REQUEST;
 import static org.apache.hadoop.ozone.container.common.impl.ContainerLayoutVersion.FILE_PER_CHUNK;
@@ -67,6 +68,8 @@ public class FilePerChunkStrategy implements ChunkManager {
   private final BlockManager blockManager;
   private final int defaultReadBufferCapacity;
   private final int readMappedBufferThreshold;
+  private final int readMappedBufferMaxCount;
+  private final Semaphore semaphore;
   private final VolumeSet volumeSet;
 
   public FilePerChunkStrategy(boolean sync, BlockManager manager,
@@ -77,7 +80,15 @@ public FilePerChunkStrategy(boolean sync, BlockManager manager,
         manager.getDefaultReadBufferCapacity();
     this.readMappedBufferThreshold = manager == null ? 0
         : manager.getReadMappedBufferThreshold();
+    this.readMappedBufferMaxCount = manager == null ? 0
+        : manager.getReadMappedBufferMaxCount();
+    LOG.info("ozone.chunk.read.mapped.buffer.max.count is load with {}", readMappedBufferMaxCount);
     this.volumeSet = volSet;
+    if (this.readMappedBufferMaxCount > 0) {
+      semaphore = new Semaphore(this.readMappedBufferMaxCount);
+    } else {
+      semaphore = null;
+    }
   }
 
   private static void checkLayoutVersion(Container container) {
@@ -265,7 +276,7 @@ public ChunkBuffer readChunk(Container container, BlockID blockID,
           long offset = info.getOffset() - chunkFileOffset;
           Preconditions.checkState(offset >= 0);
           return ChunkUtils.readData(len, bufferCapacity, file, offset, volume,
-              readMappedBufferThreshold);
+              readMappedBufferThreshold, readMappedBufferMaxCount > 0, semaphore);
         }
       } catch (StorageContainerException ex) {
         //UNABLE TO FIND chunk is not a problem as we will try with the

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/interfaces/BlockManager.java

@@ -96,6 +96,9 @@ long getCommittedBlockLength(Container container, BlockID blockID)
   /** @return the threshold to read using memory mapped buffers. */
   int getReadMappedBufferThreshold();
 
+  /** @return the max count of memory mapped buffers to read. */
+  int getReadMappedBufferMaxCount();
+
   /**
    * Shutdown ContainerManager.
    */