HDDS-13983. Support full resource lock in PoolBasedHierarchicalLockManager (#9353)

Author: swamirishi

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/lock/HierarchicalResourceLockManager.java

@@ -48,6 +48,20 @@ public interface HierarchicalResourceLockManager extends AutoCloseable {
    */
   HierarchicalResourceLock acquireWriteLock(DAGLeveledResource resource, String key) throws IOException;
 
+  /**
+   * Acquires a write lock on the specified hierarchical resource.
+   * This method is typically used for managing write operations on resources
+   * that are part of a directed acyclic graph (DAG) structure, ensuring a
+   * deterministic and deadlock-free locking order.
+   *
+   * @param resource the {@code DAGLeveledResource} representing the hierarchical
+   *                 resource for which the write lock is to be acquired
+   * @return a {@code HierarchicalResourceLock} instance to manage the lifecycle
+   *         of the acquired lock for the specified resource
+   * @throws IOException if an I/O error occurs during the lock acquisition process
+   */
+  HierarchicalResourceLock acquireResourceWriteLock(DAGLeveledResource resource) throws IOException;
+
   /**
    * Retrieves a stream of all currently locked resources in the system by the thread calling this method.
    *

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/lock/PoolBasedHierarchicalResourceLockManager.java

@@ -25,6 +25,7 @@
 import com.google.common.base.Preconditions;
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.EnumMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
@@ -34,12 +35,14 @@
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.commons.pool2.BasePooledObjectFactory;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
 import org.apache.commons.pool2.impl.GenericObjectPool;
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.ratis.util.UncheckedAutoCloseable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -56,7 +59,7 @@ public class PoolBasedHierarchicalResourceLockManager implements HierarchicalRes
 
   private final GenericObjectPool<ReadWriteLock> lockPool;
   private final ResourceLockTracker<DAGLeveledResource> resourceLockTracker;
-  private final Map<DAGLeveledResource, Map<String, LockReferenceCountPair>> lockMap;
+  private final Map<DAGLeveledResource, Pair<ReadWriteLock, Map<String, LockReferenceCountPair>>> lockMap;
 
   public PoolBasedHierarchicalResourceLockManager(OzoneConfiguration conf) {
     int softLimit = conf.getInt(OZONE_OM_HIERARCHICAL_RESOURCE_LOCKS_SOFT_LIMIT,
@@ -68,15 +71,18 @@ public PoolBasedHierarchicalResourceLockManager(OzoneConfiguration conf) {
     config.setMaxTotal(hardLimit);
     config.setBlockWhenExhausted(true);
     this.lockPool = new GenericObjectPool<>(new ReadWriteLockFactory(), config);
-    this.lockMap = new ConcurrentHashMap<>();
+    this.lockMap = new EnumMap<>(DAGLeveledResource.class);
+    for (DAGLeveledResource resource : DAGLeveledResource.values()) {
+      this.lockMap.put(resource, Pair.of(new ReentrantReadWriteLock(), new ConcurrentHashMap<>(0)));
+    }
     this.resourceLockTracker = DAGResourceLockTracker.get();
   }
 
   private ReadWriteLock operateOnLock(DAGLeveledResource resource, String key,
       Consumer<LockReferenceCountPair> function) throws IOException {
     AtomicReference<IOException> exception = new AtomicReference<>();
-    Map<String, LockReferenceCountPair> resourceLockMap =
-        this.lockMap.computeIfAbsent(resource, k -> new ConcurrentHashMap<>());
+    Pair<ReadWriteLock, Map<String, LockReferenceCountPair>> resourceLockPair = this.lockMap.get(resource);
+    Map<String, LockReferenceCountPair> resourceLockMap = resourceLockPair.getValue();
     LockReferenceCountPair lockRef = resourceLockMap.compute(key, (k, v) -> {
       if (v == null) {
         try {
@@ -111,6 +117,16 @@ public HierarchicalResourceLock acquireWriteLock(DAGLeveledResource resource, St
     return acquireLock(resource, key, false);
   }
 
+  @Override
+  public HierarchicalResourceLock acquireResourceWriteLock(DAGLeveledResource resource) throws IOException {
+    if (!resourceLockTracker.canLockResource(resource)) {
+      String errorMessage = getErrorMessage(resource);
+      LOG.error(errorMessage);
+      throw new RuntimeException(errorMessage);
+    }
+    return new PoolBasedHierarchicalResourceLock(resource);
+  }
+
   private String getErrorMessage(IOzoneManagerLock.Resource resource) {
     return "Thread '" + Thread.currentThread().getName() + "' cannot " +
         "acquire " + resource.getName() + " lock while holding " +
@@ -135,7 +151,7 @@ private HierarchicalResourceLock acquireLock(DAGLeveledResource resource, String
       throw new IOException("Unable to acquire " + (isReadLock ? "read" : "write") + " lock on resource "
           + resource + " and key " + key);
     }
-    return new PoolBasedHierarchicalResourceLock(resource, key,
+    return new PoolBasedHierarchicalResourceKeyLock(resource, key,
         isReadLock ? readWriteLock.readLock() : readWriteLock.writeLock());
   }
 
@@ -144,6 +160,40 @@ public void close() {
     this.lockPool.close();
   }
 
+  /**
+   * The PoolBasedHierachicalResourceLock class implements the HierarchicalResourceLock
+   * and UncheckedAutoCloseable interfaces to manage hierarchical resource locks from
+   * a shared pool. It ensures proper lock acquisition and release during its lifecycle.
+   */
+  private final class PoolBasedHierarchicalResourceLock implements HierarchicalResourceLock,
+      UncheckedAutoCloseable {
+    private final DAGLeveledResource resource;
+    private final Lock resourceLock;
+    private boolean lockAcquired;
+
+    private PoolBasedHierarchicalResourceLock(DAGLeveledResource resource) {
+      this.resource = resource;
+      this.resourceLock = lockMap.get(this.resource).getKey().writeLock();
+      resourceLock.lock();
+      resourceLockTracker.lockResource(this.resource);
+      lockAcquired = true;
+    }
+
+    @Override
+    public boolean isLockAcquired() {
+      return lockAcquired;
+    }
+
+    @Override
+    public synchronized void close() {
+      if (lockAcquired) {
+        resourceLock.unlock();
+        resourceLockTracker.unlockResource(this.resource);
+        lockAcquired = false;
+      }
+    }
+  }
+
   /**
    * Represents a hierarchical resource lock mechanism that operates
    * using a resource pool for acquiring and releasing locks. This class
@@ -159,20 +209,23 @@ public void close() {
    * class, {@code PoolBasedHierarchicalResourceLockManager}, which oversees
    * the lifecycle of multiple such locks.
    */
-  private final class PoolBasedHierarchicalResourceLock implements HierarchicalResourceLock, Closeable {
+  private final class PoolBasedHierarchicalResourceKeyLock implements HierarchicalResourceLock, Closeable {
 
     private boolean isLockAcquired;
-    private final Lock lock;
+    private final Lock resourceLock;
+    private final Lock keyLock;
     private final DAGLeveledResource resource;
     private final String key;
 
-    private PoolBasedHierarchicalResourceLock(DAGLeveledResource resource, String key, Lock lock) {
-      this.isLockAcquired = true;
-      this.lock = lock;
+    private PoolBasedHierarchicalResourceKeyLock(DAGLeveledResource resource, String key, Lock lock) {
+      this.keyLock = lock;
       this.resource = resource;
       this.key = key;
-      this.lock.lock();
+      this.resourceLock = lockMap.get(resource).getKey().readLock();
+      this.resourceLock.lock();
+      this.keyLock.lock();
       resourceLockTracker.lockResource(resource);
+      this.isLockAcquired = true;
     }
 
     @Override
@@ -183,7 +236,8 @@ public boolean isLockAcquired() {
     @Override
     public synchronized void close() throws IOException {
       if (isLockAcquired) {
-        this.lock.unlock();
+        this.keyLock.unlock();
+        this.resourceLock.unlock();
         resourceLockTracker.unlockResource(resource);
         operateOnLock(resource, key, (LockReferenceCountPair::decrement));
         isLockAcquired = false;
@@ -193,7 +247,7 @@ public synchronized void close() throws IOException {
 
   private static final class LockReferenceCountPair {
     private int count;
-    private ReadWriteLock lock;
+    private final ReadWriteLock lock;
 
     private LockReferenceCountPair(ReadWriteLock lock) {
       this.count = 0;
@@ -220,7 +274,7 @@ private ReadWriteLock getLock() {
   private static class ReadWriteLockFactory extends BasePooledObjectFactory<ReadWriteLock>  {
 
     @Override
-    public ReadWriteLock create() throws Exception {
+    public ReadWriteLock create() {
       return new ReentrantReadWriteLock();
     }
 

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/lock/ReadOnlyHierarchicalResourceLockManager.java

@@ -58,6 +58,11 @@ public HierarchicalResourceLock acquireWriteLock(DAGLeveledResource resource, St
     return EMPTY_LOCK_NOT_ACQUIRED;
   }
 
+  @Override
+  public HierarchicalResourceLock acquireResourceWriteLock(DAGLeveledResource resource) throws IOException {
+    return EMPTY_LOCK_NOT_ACQUIRED;
+  }
+
   @Override
   public Stream<DAGLeveledResource> getCurrentLockedResources() {
     return Stream.empty();