Revert "HDFS-17496. DataNode supports more fine-grained dataset lock …

…based on…" (apache#7279)

This reverts commit 94d6a77.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -1744,10 +1744,6 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final boolean
       DFS_DATANODE_LOCKMANAGER_TRACE_DEFAULT = false;
 
-  public static final String DFS_DATANODE_DATASET_SUBLOCK_COUNT_KEY =
-      "dfs.datanode.dataset.sublock.count";
-  public static final long DFS_DATANODE_DATASET_SUBLOCK_COUNT_DEFAULT = 1000L;
-
   // dfs.client.retry confs are moved to HdfsClientConfigKeys.Retry
   @Deprecated
   public static final String  DFS_CLIENT_RETRY_POLICY_ENABLED_KEY

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

@@ -29,8 +29,7 @@ public interface DataNodeLockManager<T extends AutoCloseDataSetLock> {
    */
   enum LockLevel {
     BLOCK_POOl,
-    VOLUME,
-    DIR
+    VOLUME
   }
 
   /**

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

@@ -96,13 +96,6 @@ 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");
     }
@@ -163,7 +156,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 if (level == LockLevel.VOLUME){
+    } else {
       AutoCloseDataSetLock bpLock = getReadLock(LockLevel.BLOCK_POOl, resources[0]);
       AutoCloseDataSetLock volLock = getReadLock(level, resources);
       volLock.setParentLock(bpLock);
@@ -172,25 +165,14 @@ 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 if (level == LockLevel.VOLUME) {
+    } else {
       AutoCloseDataSetLock bpLock = getReadLock(LockLevel.BLOCK_POOl, resources[0]);
       AutoCloseDataSetLock volLock = getWriteLock(level, resources);
       volLock.setParentLock(bpLock);
@@ -199,17 +181,6 @@ 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;
     }
   }
 
@@ -264,13 +235,8 @@ 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));
     }
   }