HDFS-17496. DataNode supports more fine-grained dataset lock based on blockid. (#7280). Contributed by hfutatzhanghb.

Signed-off-by: He Xiaoqiao <[email protected]>

Author: hfutatzhanghb

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/common/DataNodeLockManager.java

@@ -24,12 +24,17 @@
 public interface DataNodeLockManager<T extends AutoCloseDataSetLock> {
 
   /**
-   * Acquire block pool level first if you want to Acquire volume lock.
+   * Acquire block pool level and volume level lock first if you want to acquire dir lock.
    * Or only acquire block pool level lock.
+   * There are several locking sequential patterns as below:
+   * 1. block pool
+   * 2. block poll -> volume
+   * 3. block pool level -> volume -> dir
    */
   enum LockLevel {
     BLOCK_POOl,
-    VOLUME
+    VOLUME,
+    DIR
   }
 
   /**

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNodeLayoutSubLockStrategy.java

@@ -0,0 +1,33 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.server.datanode;
+
+import java.util.List;
+
+public class DataNodeLayoutSubLockStrategy implements DataSetSubLockStrategy {
+  @Override
+  public String blockIdToSubLock(long blockid) {
+    return DatanodeUtil.idToBlockDirSuffix(blockid);
+  }
+
+  @Override
+  public List<String> getAllSubLockNames() {
+    return DatanodeUtil.getAllSubDirNameForDataSetLock();
+  }
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataSetLockManager.java

@@ -96,6 +96,13 @@ private String generateLockName(LockLevel level, String... resources) {
             + resources[0] + "volume lock :" + resources[1]);
       }
       return resources[0] + resources[1];
+    } else if (resources.length == 3 && level == LockLevel.DIR) {
+      if (resources[0] == null || resources[1] == null || resources[2] == null) {
+        throw new IllegalArgumentException("acquire a null dataset lock : "
+            + resources[0] + ",volume lock :" + resources[1]
+        + ",subdir lock :" + resources[2]);
+      }
+      return resources[0] + resources[1] + resources[2];
     } else {
       throw new IllegalArgumentException("lock level do not match resource");
     }
@@ -156,7 +163,7 @@ public DataSetLockManager(Configuration conf, DataNode dn) {
   public AutoCloseDataSetLock readLock(LockLevel level, String... resources) {
     if (level == LockLevel.BLOCK_POOl) {
       return getReadLock(level, resources[0]);
-    } else {
+    } else if (level == LockLevel.VOLUME){
       AutoCloseDataSetLock bpLock = getReadLock(LockLevel.BLOCK_POOl, resources[0]);
       AutoCloseDataSetLock volLock = getReadLock(level, resources);
       volLock.setParentLock(bpLock);
@@ -165,14 +172,25 @@ public AutoCloseDataSetLock readLock(LockLevel level, String... resources) {
             resources[0]);
       }
       return volLock;
+    } else {
+      AutoCloseDataSetLock bpLock = getReadLock(LockLevel.BLOCK_POOl, resources[0]);
+      AutoCloseDataSetLock volLock = getReadLock(LockLevel.VOLUME, resources[0], resources[1]);
+      volLock.setParentLock(bpLock);
+      AutoCloseDataSetLock dirLock = getReadLock(level, resources);
+      dirLock.setParentLock(volLock);
+      if (openLockTrace) {
+        LOG.debug("Sub lock " + resources[0] + resources[1] + resources[2] + " parent lock " +
+            resources[0] + resources[1]);
+      }
+      return dirLock;
     }
   }
 
   @Override
   public AutoCloseDataSetLock writeLock(LockLevel level, String... resources) {
     if (level == LockLevel.BLOCK_POOl) {
       return getWriteLock(level, resources[0]);
-    } else {
+    } else if (level == LockLevel.VOLUME) {
       AutoCloseDataSetLock bpLock = getReadLock(LockLevel.BLOCK_POOl, resources[0]);
       AutoCloseDataSetLock volLock = getWriteLock(level, resources);
       volLock.setParentLock(bpLock);
@@ -181,6 +199,17 @@ public AutoCloseDataSetLock writeLock(LockLevel level, String... resources) {
             resources[0]);
       }
       return volLock;
+    } else {
+      AutoCloseDataSetLock bpLock = getReadLock(LockLevel.BLOCK_POOl, resources[0]);
+      AutoCloseDataSetLock volLock = getReadLock(LockLevel.VOLUME, resources[0], resources[1]);
+      volLock.setParentLock(bpLock);
+      AutoCloseDataSetLock dirLock = getWriteLock(level, resources);
+      dirLock.setParentLock(volLock);
+      if (openLockTrace) {
+        LOG.debug("Sub lock " + resources[0] + resources[1] + resources[2] + " parent lock " +
+            resources[0] + resources[1]);
+      }
+      return dirLock;
     }
   }
 
@@ -235,8 +264,13 @@ public void addLock(LockLevel level, String... resources) {
     String lockName = generateLockName(level, resources);
     if (level == LockLevel.BLOCK_POOl) {
       lockMap.addLock(lockName, new ReentrantReadWriteLock(isFair));
+    } else if (level == LockLevel.VOLUME) {
+      lockMap.addLock(resources[0], new ReentrantReadWriteLock(isFair));
+      lockMap.addLock(lockName, new ReentrantReadWriteLock(isFair));
     } else {
       lockMap.addLock(resources[0], new ReentrantReadWriteLock(isFair));
+      lockMap.addLock(generateLockName(LockLevel.VOLUME, resources[0], resources[1]),
+          new ReentrantReadWriteLock(isFair));
       lockMap.addLock(lockName, new ReentrantReadWriteLock(isFair));
     }
   }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataSetSubLockStrategy.java

@@ -0,0 +1,36 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdfs.server.datanode;
+
+import java.util.List;
+
+/**
+ * This interface is used to generate sub lock name for a blockid.
+ */
+public interface DataSetSubLockStrategy {
+
+  /**
+   * Generate sub lock name for the given blockid.
+   * @param blockid the block id.
+   * @return sub lock name for the input blockid.
+   */
+  String blockIdToSubLock(long blockid);
+
+  List<String> getAllSubLockNames();
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DatanodeUtil.java

@@ -21,6 +21,8 @@
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.hdfs.protocol.Block;
@@ -37,6 +39,7 @@ public class DatanodeUtil {
   public static final String DISK_ERROR = "Possible disk error: ";
 
   private static final String SEP = System.getProperty("file.separator");
+  private static final long MASK = 0x1F;
 
   /** Get the cause of an I/O exception if caused by a possible disk error
    * @param ioe an I/O exception
@@ -112,6 +115,21 @@ public static boolean dirNoFilesRecursive(
     return true;
   }
 
+  /**
+   * Take an example.
+   * We hava a block with blockid mapping to:
+   * "/data1/hadoop/hdfs/datanode/current/BP-xxxx/current/finalized/subdir0/subdir1"
+   * We return "subdir0/subdir0".
+   * @param blockId the block id.
+   * @return two-level subdir string where block will be stored.
+   */
+  public static String idToBlockDirSuffix(long blockId) {
+    int d1 = (int) ((blockId >> 16) & MASK);
+    int d2 = (int) ((blockId >> 8) & MASK);
+    return DataStorage.BLOCK_SUBDIR_PREFIX + d1 + SEP +
+        DataStorage.BLOCK_SUBDIR_PREFIX + d2;
+  }
+
   /**
    * Get the directory where a finalized block with this ID should be stored.
    * Do not attempt to create the directory.
@@ -120,13 +138,21 @@ public static boolean dirNoFilesRecursive(
    * @return
    */
   public static File idToBlockDir(File root, long blockId) {
-    int d1 = (int) ((blockId >> 16) & 0x1F);
-    int d2 = (int) ((blockId >> 8) & 0x1F);
-    String path = DataStorage.BLOCK_SUBDIR_PREFIX + d1 + SEP +
-        DataStorage.BLOCK_SUBDIR_PREFIX + d2;
+    String path = idToBlockDirSuffix(blockId);
     return new File(root, path);
   }
 
+  public static List<String> getAllSubDirNameForDataSetLock() {
+    List<String> res = new ArrayList<>();
+    for (int d1 = 0; d1 <= MASK; d1++) {
+      for (int d2 = 0; d2 <= MASK; d2++) {
+        res.add(DataStorage.BLOCK_SUBDIR_PREFIX + d1 + SEP +
+            DataStorage.BLOCK_SUBDIR_PREFIX + d2);
+      }
+    }
+    return res;
+  }
+
   /**
    * @return the FileInputStream for the meta data of the given block.
    * @throws FileNotFoundException