HDDS-12560. Reclaimable Filter for Snaphost Garbage Collections

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

@@ -0,0 +1,30 @@
+/*
+ * 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.ozone.util;
+
+/**
+ * Represents a function that accepts one argument and produces a result.
+ * This is a functional interface whose functional method is apply(Object).
+ * Type parameters:
+ * <T> – the type of the input to the function
+ * <R> – the type of the result of the function
+ * <E> - the type of exception thrown.
+ */
+public interface CheckedFunction<T, R, E extends Exception> {
+  R apply(T t) throws E;
+}

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.om.lock;
 
 import com.google.common.annotations.VisibleForTesting;
+import java.util.Collection;
 
 /**
  * Interface for OM Metadata locks.
@@ -27,19 +28,31 @@ public interface IOzoneManagerLock {
   OMLockDetails acquireReadLock(OzoneManagerLock.Resource resource,
                                 String... resources);
 
+  OMLockDetails acquireReadLocks(OzoneManagerLock.Resource resource, Collection<String[]> resources);
+
+
   OMLockDetails acquireWriteLock(OzoneManagerLock.Resource resource,
                                  String... resources);
 
+  OMLockDetails acquireWriteLocks(OzoneManagerLock.Resource resource,
+                                 Collection<String[]> resources);
+
   boolean acquireMultiUserLock(String firstUser, String secondUser);
 
   void releaseMultiUserLock(String firstUser, String secondUser);
 
   OMLockDetails releaseWriteLock(OzoneManagerLock.Resource resource,
                         String... resources);
 
+  OMLockDetails releaseWriteLocks(OzoneManagerLock.Resource resource,
+                                 Collection<String[]> resources);
+
   OMLockDetails releaseReadLock(OzoneManagerLock.Resource resource,
                                 String... resources);
 
+  OMLockDetails releaseReadLocks(OzoneManagerLock.Resource resource,
+                                Collection<String[]> resources);
+
   @VisibleForTesting
   int getReadHoldCount(OzoneManagerLock.Resource resource,
       String... resources);

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

@@ -20,6 +20,7 @@
 import static org.apache.hadoop.ozone.om.lock.OMLockDetails.EMPTY_DETAILS_LOCK_ACQUIRED;
 import static org.apache.hadoop.ozone.om.lock.OMLockDetails.EMPTY_DETAILS_LOCK_NOT_ACQUIRED;
 
+import java.util.Collection;
 import org.apache.hadoop.ozone.om.lock.OzoneManagerLock.Resource;
 
 /**
@@ -34,12 +35,22 @@ public OMLockDetails acquireReadLock(Resource resource, String... resources) {
     return EMPTY_DETAILS_LOCK_ACQUIRED;
   }
 
+  @Override
+  public OMLockDetails acquireReadLocks(Resource resource, Collection<String[]> resources) {
+    return EMPTY_DETAILS_LOCK_ACQUIRED;
+  }
+
   @Override
   public OMLockDetails acquireWriteLock(Resource resource,
       String... resources) {
     return EMPTY_DETAILS_LOCK_NOT_ACQUIRED;
   }
 
+  @Override
+  public OMLockDetails acquireWriteLocks(Resource resource, Collection<String[]> resources) {
+    return EMPTY_DETAILS_LOCK_NOT_ACQUIRED;
+  }
+
   @Override
   public boolean acquireMultiUserLock(String firstUser, String secondUser) {
     return false;
@@ -56,11 +67,21 @@ public OMLockDetails releaseWriteLock(Resource resource,
     return EMPTY_DETAILS_LOCK_NOT_ACQUIRED;
   }
 
+  @Override
+  public OMLockDetails releaseWriteLocks(Resource resource, Collection<String[]> resources) {
+    return EMPTY_DETAILS_LOCK_NOT_ACQUIRED;
+  }
+
   @Override
   public OMLockDetails releaseReadLock(Resource resource, String... resources) {
     return EMPTY_DETAILS_LOCK_NOT_ACQUIRED;
   }
 
+  @Override
+  public OMLockDetails releaseReadLocks(Resource resource, Collection<String[]> resources) {
+    return EMPTY_DETAILS_LOCK_NOT_ACQUIRED;
+  }
+
   @Override
   public int getReadHoldCount(Resource resource, String... resources) {
     return 0;

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

@@ -27,14 +27,18 @@
 import com.google.common.util.concurrent.Striped;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.utils.CompositeKey;
 import org.apache.hadoop.hdds.utils.SimpleStriped;
 import org.apache.hadoop.ipc.ProcessingDetails.Timing;
 import org.apache.hadoop.ipc.Server;
@@ -122,6 +126,12 @@ private Striped<ReadWriteLock> createStripeLock(Resource r,
     return SimpleStriped.readWriteLock(size, fair);
   }
 
+  private Iterable<ReadWriteLock> bulkGetLock(Resource resource, Collection<String[]> keys) {
+    Striped<ReadWriteLock> striped = stripedLockByResource.get(resource);
+    return striped.bulkGet(keys.stream().filter(Objects::nonNull)
+        .map(CompositeKey::combineKeys).collect(Collectors.toList()));
+  }
+
   private ReentrantReadWriteLock getLock(Resource resource, String... keys) {
     Striped<ReadWriteLock> striped = stripedLockByResource.get(resource);
     Object key = combineKeys(keys);
@@ -150,6 +160,28 @@ public OMLockDetails acquireReadLock(Resource resource, String... keys) {
     return acquireLock(resource, true, keys);
   }
 
+  /**
+   * Acquire read locks on a list of resources.
+   *
+   * For S3_BUCKET_LOCK, VOLUME_LOCK, BUCKET_LOCK type resource, same
+   * thread acquiring lock again is allowed.
+   *
+   * For USER_LOCK, PREFIX_LOCK, S3_SECRET_LOCK type resource, same thread
+   * acquiring lock again is not allowed.
+   *
+   * Special Note for USER_LOCK: Single thread can acquire single user lock/
+   * multi user lock. But not both at the same time.
+   * @param resource - Type of the resource.
+   * @param keys - A list of Resource names on which user want to acquire locks.
+   * For Resource type BUCKET_LOCK, first param should be volume, second param
+   * should be bucket name. For remaining all resource only one param should
+   * be passed.
+   */
+  @Override
+  public OMLockDetails acquireReadLocks(Resource resource, Collection<String[]> keys) {
+    return acquireLocks(resource, true, keys);
+  }
+
   /**
    * Acquire write lock on resource.
    *
@@ -172,6 +204,54 @@ public OMLockDetails acquireWriteLock(Resource resource, String... keys) {
     return acquireLock(resource, false, keys);
   }
 
+  /**
+   * Acquire write locks on a list of resources.
+   *
+   * For S3_BUCKET_LOCK, VOLUME_LOCK, BUCKET_LOCK type resource, same
+   * thread acquiring lock again is allowed.
+   *
+   * For USER_LOCK, PREFIX_LOCK, S3_SECRET_LOCK type resource, same thread
+   * acquiring lock again is not allowed.
+   *
+   * Special Note for USER_LOCK: Single thread can acquire single user lock/
+   * multi user lock. But not both at the same time.
+   * @param resource - Type of the resource.
+   * @param keys - A list of Resource names on which user want to acquire lock.
+   * For Resource type BUCKET_LOCK, first param should be volume, second param
+   * should be bucket name. For remaining all resource only one param should
+   * be passed.
+   */
+  @Override
+  public OMLockDetails acquireWriteLocks(Resource resource, Collection<String[]> keys) {
+    return acquireLocks(resource, false, keys);
+  }
+
+  private OMLockDetails acquireLocks(Resource resource, boolean isReadLock,
+                                    Collection<String[]> keys) {
+    omLockDetails.get().clear();
+    if (!resource.canLock(lockSet.get())) {
+      String errorMessage = getErrorMessage(resource);
+      LOG.error(errorMessage);
+      throw new RuntimeException(errorMessage);
+    }
+
+    long startWaitingTimeNanos = Time.monotonicNowNanos();
+
+    for (ReadWriteLock lock : bulkGetLock(resource, keys)) {
+      if (isReadLock) {
+        lock.readLock().lock();
+        updateReadLockMetrics(resource, (ReentrantReadWriteLock) lock, startWaitingTimeNanos);
+      } else {
+        lock.writeLock().lock();
+        updateWriteLockMetrics(resource, (ReentrantReadWriteLock) lock, startWaitingTimeNanos);
+      }
+    }
+
+    lockSet.set(resource.setLock(lockSet.get()));
+    omLockDetails.get().setLockAcquired(true);
+    return omLockDetails.get();
+  }
+
   private OMLockDetails acquireLock(Resource resource, boolean isReadLock,
       String... keys) {
     omLockDetails.get().clear();
@@ -317,6 +397,11 @@ public OMLockDetails releaseWriteLock(Resource resource, String... keys) {
     return releaseLock(resource, false, keys);
   }
 
+  @Override
+  public OMLockDetails releaseWriteLocks(Resource resource, Collection<String[]> keys) {
+    return releaseLocks(resource, false, keys);
+  }
+
   /**
    * Release read lock on resource.
    * @param resource - Type of the resource.
@@ -330,6 +415,19 @@ public OMLockDetails releaseReadLock(Resource resource, String... keys) {
     return releaseLock(resource, true, keys);
   }
 
+  /**
+   * Release read locks on a list of resources.
+   * @param resource - Type of the resource.
+   * @param keys - Resource names on which user want to acquire lock.
+   * For Resource type BUCKET_LOCK, first param should be volume, second param
+   * should be bucket name. For remaining all resource only one param should
+   * be passed.
+   */
+  @Override
+  public OMLockDetails releaseReadLocks(Resource resource, Collection<String[]> keys) {
+    return releaseLocks(resource, true, keys);
+  }
+
   private OMLockDetails releaseLock(Resource resource, boolean isReadLock,
       String... keys) {
     omLockDetails.get().clear();
@@ -347,6 +445,26 @@ private OMLockDetails releaseLock(Resource resource, boolean isReadLock,
     return omLockDetails.get();
   }
 
+  private OMLockDetails releaseLocks(Resource resource, boolean isReadLock,
+                                    Collection<String[]> keys) {
+    omLockDetails.get().clear();
+    Iterable<ReadWriteLock> locks = bulkGetLock(resource, keys);
+
+    for (ReadWriteLock lock : locks) {
+      if (isReadLock) {
+        lock.readLock().unlock();
+        updateReadUnlockMetrics(resource, (ReentrantReadWriteLock) lock);
+      } else {
+        boolean isWriteLocked = ((ReentrantReadWriteLock)lock).isWriteLockedByCurrentThread();
+        lock.writeLock().unlock();
+        updateWriteUnlockMetrics(resource, (ReentrantReadWriteLock) lock, isWriteLocked);
+      }
+    }
+
+    lockSet.set(resource.clearLock(lockSet.get()));
+    return omLockDetails.get();
+  }
+
   private void updateReadUnlockMetrics(Resource resource,
       ReentrantReadWriteLock lock) {
     /*
@@ -453,7 +571,8 @@ public enum Resource {
     S3_SECRET_LOCK((byte) 4, "S3_SECRET_LOCK"), // 31
     KEY_PATH_LOCK((byte) 5, "KEY_PATH_LOCK"), //63
     PREFIX_LOCK((byte) 6, "PREFIX_LOCK"), //127
-    SNAPSHOT_LOCK((byte) 7, "SNAPSHOT_LOCK"); // = 255
+    SNAPSHOT_LOCK((byte) 7, "SNAPSHOT_LOCK"), // = 255
+    SNAPSHOT_GC_LOCK((byte) 8, "SNAPSHOT_GC_LOCK");
 
     // level of the resource
     private byte lockLevel;

hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/om/lock/TestOzoneManagerLock.java

@@ -25,6 +25,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Stack;
 import java.util.UUID;
@@ -287,6 +288,37 @@ void testLockResourceParallel() throws Exception {
 
   }
 
+  @Test
+  void testMultiLocksResourceParallel() throws Exception {
+    OzoneManagerLock lock = new OzoneManagerLock(new OzoneConfiguration());
+
+    for (Resource resource : Resource.values()) {
+      final List<String[]> resourceName = Arrays.asList(generateResourceName(resource),
+          generateResourceName(resource), generateResourceName(resource));
+      lock.acquireWriteLocks(resource, resourceName.subList(1, resourceName.size()));
+
+      AtomicBoolean gotLock = new AtomicBoolean(false);
+      new Thread(() -> {
+        lock.acquireWriteLocks(resource, resourceName.subList(0, 2));
+        gotLock.set(true);
+        lock.releaseWriteLocks(resource, resourceName.subList(0, 2));
+      }).start();
+      // Let's give some time for the new thread to run
+      Thread.sleep(100);
+      // Since the new thread is trying to get lock on same resource,
+      // it will wait.
+      assertFalse(gotLock.get());
+      lock.releaseWriteLocks(resource, resourceName.subList(1, resourceName.size()));
+      // Since we have released the lock, the new thread should have the lock
+      // now.
+      // Let's give some time for the new thread to run
+      Thread.sleep(100);
+      assertTrue(gotLock.get());
+    }
+
+  }
+
+
   @Test
   void testMultiLockResourceParallel() throws Exception {
     OzoneManagerLock lock = new OzoneManagerLock(new OzoneConfiguration());