HDFS-16208. Support Delete API with FGL.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/PartitionedGSet.java

@@ -345,4 +345,21 @@ public void latchWriteLock(K[] keys) {
     assert pLock != null : "pLock is null";
     pLock.writeTopUnlock();
   }
+
+  /**
+   * Verify whether the current thread holds the lock on given key / partition.
+   * @param key INode for which lock to be verified.
+   * @return true if the current thread holds the partition lock.
+   */
+  public boolean hasWriteLock(final K key) {
+    Entry<K, PartitionEntry> partEntry = partitions.floorEntry(key);
+    if (partEntry == null) {
+      return false;
+    }
+    PartitionEntry part = partEntry.getValue();
+    if (part == null) {
+      throw new IllegalStateException("Null partition for key: " + key);
+    }
+    return part.partLock.hasWriteChildLock();
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirDeleteOp.java

@@ -243,6 +243,9 @@ private static boolean unprotectedDelete(FSDirectory fsd, INodesInPath iip,
     final int latestSnapshot = iip.getLatestSnapshotId();
     targetNode.recordModification(latestSnapshot);
 
+    // switch the locks
+    fsd.getINodeMap().latchWriteLock(iip.getParentINodesInPath(), new INode[] {iip.getLastINode()});
+
     // Remove the node from the namespace
     long removed = fsd.removeLastINode(iip);
     if (removed == -1) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java

@@ -1607,6 +1607,18 @@ void moveInodes() throws IOException {
    */
   public final void removeFromInodeMap(List<? extends INode> inodes) {
     if (inodes != null) {
+      for (INode inode : inodes) {
+        if (inodeMap.hasWriteLock(inode)) {
+          // Consider /hive/blr to be removed, and it contains children as below
+          // /hive/blr/A1, /hive/blr/A2, /hive/blr/A3 all are in same
+          // partition P1
+          // /hive/blr/A3/a1, /hive/blr/A3/a2 and in Partition P2
+          // Here we iterate through all inodes (5), and acquire lock on each
+          // Partition only once
+          continue;
+        }
+        inodeMap.latchWriteLock(inode);
+      }
       for (INode inode : inodes) {
         if (inode != null && inode instanceof INodeWithAdditionalFields) {
           inodeMap.remove(inode);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -2785,7 +2785,7 @@ private HdfsFileStatus startFileInt(String src,
       if (!skipSync) {
         getEditLog().logSync();
         if (toRemoveBlocks != null) {
-          removeBlocks(toRemoveBlocks);
+          removeBlocksWithFGL(toRemoveBlocks, iip.getPath(), pc);
           toRemoveBlocks.clear();
         }
       }
@@ -3349,11 +3349,55 @@ boolean delete(String src, boolean recursive, boolean logRetryCache)
     getEditLog().logSync();
     logAuditEvent(ret, operationName, src);
     if (toRemovedBlocks != null) {
-      removeBlocks(toRemovedBlocks); // Incremental deletion of blocks
+      removeBlocksWithFGL(toRemovedBlocks, src, pc); // Incremental deletion of blocks
     }
     return ret;
   }
 
+  /** Duplicate of removeBlocks(blocks), difference is it acquire partition
+   * lock instead of global write lock.
+   * @param blocks blocks to be removed.
+   * @param src    Source file on which partition lock held.
+   * @param pc     Permission checker for resolving the path.
+   * @throws IOException
+   * @see this.removeBlocks(BlocksMapUpdateInfo ), duplicate method.
+   * Remove the above one once all APIs implemented FGL.
+   */
+  private void removeBlocksWithFGL(BlocksMapUpdateInfo blocks, String src,
+      FSPermissionChecker pc) throws IOException {
+    List<BlockInfo> toDeleteList = blocks.getToDeleteList();
+    Iterator<BlockInfo> iter = toDeleteList.iterator();
+    while (iter.hasNext()) {
+      writeLock();
+      try {
+        switchToPartitionLock(src, pc);
+        for (int i = 0; i < blockDeletionIncrement && iter.hasNext(); i++) {
+          blockManager.removeBlock(iter.next());
+        }
+      } finally {
+        writeUnlock("removeBlocks");
+      }
+    }
+  }
+
+  /**
+   * Switch global write lock with the partition lock.
+   *
+   * @param src source file on which lock to be acquired
+   * @param pc  PermissionChecker to resolve the source
+   * @throws IOException when fails to access the path.
+   */
+  private void switchToPartitionLock(String src, FSPermissionChecker pc)
+      throws IOException {
+    INodesInPath iip = dir.resolvePath(pc, src, DirOp.WRITE_LINK);
+    iip = iip.getExistingINodes();
+    if (null != iip.getParentINodesInPath()) {
+      // switch the locks
+      // Only content gets deleted, acquire only file / directory level lock.
+      dir.getINodeMap().latchWriteLock(iip, new INode[0]);
+    }
+  }
+
   FSPermissionChecker getPermissionChecker()
       throws AccessControlException {
     return dir.getPermissionChecker();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeMap.java

@@ -329,4 +329,22 @@ public void latchWriteLock(INodesInPath iip, INode[] missing) {
     ((PartitionedGSet<INode, INodeWithAdditionalFields>)
         map).latchWriteLock(allINodes);
   }
+
+  /**
+   * Take the partition lock on the given INode key.
+   * @param iNode INode to acquire the partition lock.
+   */
+  public void latchWriteLock(INode iNode) {
+    if (!(map instanceof PartitionedGSet)) {
+      return;
+    }
+
+    ((PartitionedGSet<INode, INodeWithAdditionalFields>) map).latchWriteLock(
+        new INode[] {iNode });
+  }
+
+  public boolean hasWriteLock(final INode key) {
+    return ((PartitionedGSet<INode, INodeWithAdditionalFields>) map).
+        hasWriteLock(key);
+  }
 }