HDDS-10156. Optimize Snapshot Cache get and eviction

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -4265,6 +4265,16 @@
     </description>
   </property>
 
+  <property>
+    <name>ozone.om.snapshot.cache.cleanup.service.run.internal</name>
+    <value>1m</value>
+    <tag>OZONE, OM</tag>
+    <description>
+      Interval at which snapshot cache clean up will run.
+      Uses millisecond by default when no time unit is specified.
+    </description>
+  </property>
+
   <property>
     <name>ozone.om.snapshot.sst_dumptool.pool.size</name>
     <value>1</value>

hadoop-hdds/rocksdb-checkpoint-differ/src/main/java/org/apache/ozone/rocksdiff/RocksDBCheckpointDiffer.java

@@ -301,7 +301,7 @@ private void createSstBackUpDir() {
   }
 
   @Override
-  public void close() throws Exception {
+  public void close() {
     if (!closed) {
       synchronized (this) {
         if (!closed) {

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

@@ -580,12 +580,19 @@ private OMConfigKeys() {
 
   public static final String OZONE_OM_SNAPSHOT_DIFF_CLEANUP_SERVICE_RUN_INTERVAL
       = "ozone.om.snapshot.diff.cleanup.service.run.internal";
+  public static final String
+      OZONE_OM_SNAPSHOT_CACHE_CLEANUP_SERVICE_RUN_INTERVAL
+      = "ozone.om.snapshot.cache.cleanup.service.run.internal";
   public static final long
       OZONE_OM_SNAPSHOT_DIFF_CLEANUP_SERVICE_RUN_INTERVAL_DEFAULT
       = TimeUnit.MINUTES.toMillis(1);
+  public static final long
+      OZONE_OM_SNAPSHOT_CACHE_CLEANUP_SERVICE_RUN_INTERVAL_DEFAULT
+      = TimeUnit.MINUTES.toMillis(1);
 
   public static final String OZONE_OM_SNAPSHOT_DIFF_CLEANUP_SERVICE_TIMEOUT
       = "ozone.om.snapshot.diff.cleanup.service.timeout";
+
   public static final long
       OZONE_OM_SNAPSHOT_DIFF_CLEANUP_SERVICE_TIMEOUT_DEFAULT
       = TimeUnit.MINUTES.toMillis(5);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java

@@ -1665,10 +1665,10 @@ private boolean versionExistsInPreviousSnapshot(OmKeyInfo omKeyInfo,
   /**
    * Get the latest OmSnapshot for a snapshot path.
    */
-  public ReferenceCounted<
-      IOmMetadataReader, SnapshotCache> getLatestActiveSnapshot(
-          String volumeName, String bucketName,
-          OmSnapshotManager snapshotManager)
+  public ReferenceCounted<IOmMetadataReader,
+      SnapshotCache> getLatestActiveSnapshot(String volumeName,
+                                             String bucketName,
+                                             OmSnapshotManager snapshotManager)
       throws IOException {
 
     String snapshotPath = volumeName + OM_KEY_PREFIX + bucketName;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmSnapshotManager.java

@@ -82,6 +82,8 @@
 import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_CHECKPOINT_DIR;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_DIFF_DB_NAME;
 import static org.apache.hadoop.ozone.OzoneConsts.OM_SNAPSHOT_INDICATOR;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_CACHE_CLEANUP_SERVICE_RUN_INTERVAL;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_CACHE_CLEANUP_SERVICE_RUN_INTERVAL_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_CACHE_MAX_SIZE_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_DB_MAX_OPEN_FILES;
@@ -270,8 +272,14 @@ public OmSnapshotManager(OzoneManager ozoneManager) {
       }
     };
 
+
     // Init snapshot cache
-    this.snapshotCache = new SnapshotCache(this, loader, softCacheSize);
+    long cacheCleanupServiceInterval = ozoneManager.getConfiguration()
+        .getTimeDuration(OZONE_OM_SNAPSHOT_CACHE_CLEANUP_SERVICE_RUN_INTERVAL,
+            OZONE_OM_SNAPSHOT_CACHE_CLEANUP_SERVICE_RUN_INTERVAL_DEFAULT,
+            TimeUnit.MILLISECONDS);
+    this.snapshotCache = new SnapshotCache(this, loader,
+        softCacheSize, cacheCleanupServiceInterval);
 
     this.snapshotDiffManager = new SnapshotDiffManager(snapshotDiffDb, differ,
         ozoneManager, snapshotCache, snapDiffJobCf, snapDiffReportCf,
@@ -896,6 +904,7 @@ public void close() {
     }
     if (snapshotCache != null) {
       snapshotCache.invalidateAll();
+      snapshotCache.close();
     }
     if (snapshotDiffCleanupService != null) {
       snapshotDiffCleanupService.shutdown();

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -485,7 +485,8 @@ private enum State {
   private OmMetadataReader omMetadataReader;
   // Wrap active DB metadata reader in ReferenceCounted once to avoid
   // instance creation every single time.
-  private ReferenceCounted<IOmMetadataReader, SnapshotCache> rcOmMetadataReader;
+  private ReferenceCounted<IOmMetadataReader, SnapshotCache>
+      rcOmMetadataReader;
   private OmSnapshotManager omSnapshotManager;
 
   @SuppressWarnings("methodlength")
@@ -2580,8 +2581,8 @@ public boolean getAllowListAllVolumes() {
     return allowListAllVolumes;
   }
 
-  public ReferenceCounted<
-      IOmMetadataReader, SnapshotCache> getOmMetadataReader() {
+  public ReferenceCounted<IOmMetadataReader,
+      SnapshotCache> getOmMetadataReader() {
     return rcOmMetadataReader;
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/ReferenceCounted.java

@@ -25,7 +25,7 @@
 /**
  * Add reference counter to an object instance.
  */
-public class ReferenceCounted<T, U>
+public class ReferenceCounted<T, U extends ReferenceCountedCallback>
     implements AutoCloseable {
 
   /**
@@ -126,7 +126,9 @@ public long decrementRefCount() {
       Preconditions.checkState(newValTotal >= 0L,
           "Total reference count underflow");
     }
-
+    if (refCount.get() == 0) {
+      this.parentWithCallback.callback(this);
+    }
     return refCount.get();
   }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/ReferenceCountedCallback.java

@@ -0,0 +1,25 @@
+/*
+ * 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.om.snapshot;
+
+/**
+ * Callback interface for ReferenceCounted.
+ */
+public interface ReferenceCountedCallback {
+  void callback(ReferenceCounted referenceCounted);
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotCache.java

@@ -19,6 +19,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.cache.CacheLoader;
+import org.apache.hadoop.hdds.utils.Scheduler;
 import org.apache.hadoop.ozone.om.IOmMetadataReader;
 import org.apache.hadoop.ozone.om.OmSnapshot;
 import org.apache.hadoop.ozone.om.OmSnapshotManager;
@@ -29,15 +30,17 @@
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.FILE_NOT_FOUND;
 import static org.apache.hadoop.ozone.om.helpers.SnapshotInfo.SnapshotStatus.SNAPSHOT_ACTIVE;
 
 /**
  * Thread-safe custom unbounded LRU cache to manage open snapshot DB instances.
  */
-public class SnapshotCache {
+public class SnapshotCache implements ReferenceCountedCallback, AutoCloseable {
 
   static final Logger LOG = LoggerFactory.getLogger(SnapshotCache.class);
 
@@ -53,14 +56,29 @@ public class SnapshotCache {
   // opened on the OM.
   private final int cacheSizeLimit;
 
+  private final Set<String> pendingEvictionQueue;
+  private final Scheduler scheduler;
+  private static final String SNAPSHOT_CACHE_CLEANUP_SERVICE =
+      "SnapshotCacheCleanupService";
+
+
   public SnapshotCache(
       OmSnapshotManager omSnapshotManager,
       CacheLoader<String, OmSnapshot> cacheLoader,
-      int cacheSizeLimit) {
+      int cacheSizeLimit, long cleanupInterval) {
     this.dbMap = new ConcurrentHashMap<>();
     this.omSnapshotManager = omSnapshotManager;
     this.cacheLoader = cacheLoader;
     this.cacheSizeLimit = cacheSizeLimit;
+    this.pendingEvictionQueue = ConcurrentHashMap.newKeySet();
+    if (cleanupInterval > 0) {
+      this.scheduler = new Scheduler(SNAPSHOT_CACHE_CLEANUP_SERVICE,
+          true, 1);
+      this.scheduler.scheduleWithFixedDelay(this::cleanup, cleanupInterval,
+          cleanupInterval, TimeUnit.MILLISECONDS);
+    } else {
+      this.scheduler = null;
+    }
   }
 
   @VisibleForTesting
@@ -114,6 +132,13 @@ public void invalidateAll() {
     }
   }
 
+  @Override
+  public void close() {
+    if (this.scheduler != null) {
+      this.scheduler.close();
+    }
+  }
+
   /**
    * State the reason the current thread is getting the OmSnapshot instance.
    * Unused for now.
@@ -178,18 +203,12 @@ public ReferenceCounted<IOmMetadataReader, SnapshotCache> get(String key, boolea
     // when called from SDT (and some) if the snapshot is not active anymore.
     if (!skipActiveCheck && !omSnapshotManager.isSnapshotStatus(key, SNAPSHOT_ACTIVE)) {
       // Ref count was incremented. Need to decrement on exception here.
-      rcOmSnapshot.decrementRefCount();
+      rcOmSnapshot.close();
       throw new OMException("Unable to load snapshot. " +
           "Snapshot with table key '" + key + "' is no longer active",
           FILE_NOT_FOUND);
     }
 
-    // Check if any entries can be cleaned up.
-    // At this point, cache size might temporarily exceed cacheSizeLimit
-    // even if there are entries that can be evicted, which is fine since it
-    // is a soft limit.
-    cleanup();
-
     return rcOmSnapshot;
   }
 
@@ -198,18 +217,12 @@ public ReferenceCounted<IOmMetadataReader, SnapshotCache> get(String key, boolea
    * @param key snapshot table key
    */
   public void release(String key) {
-    dbMap.compute(key, (k, v) -> {
-      if (v == null) {
-        throw new IllegalArgumentException("Key '" + key + "' does not exist in cache.");
-      } else {
-        v.decrementRefCount();
-      }
-      return v;
-    });
-
-    // The cache size might have already exceeded the soft limit
-    // Thus triggering cleanup() to check and evict if applicable
-    cleanup();
+    ReferenceCounted<IOmMetadataReader, SnapshotCache> val = dbMap.get(key);
+    if (val == null) {
+      throw new IllegalArgumentException("Key '" + key + "' does not " +
+          "exist in cache.");
+    }
+    val.decrementRefCount();
   }
 
   /**
@@ -221,11 +234,25 @@ public void release(OmSnapshot omSnapshot) {
     release(snapshotTableKey);
   }
 
+  /**
+   * Callback method used to enqueue or dequeue ReferenceCounted from
+   * pendingEvictionList.
+   * @param referenceCounted ReferenceCounted object
+   */
+  @Override
+  public void callback(ReferenceCounted referenceCounted) {
+    if (referenceCounted.getTotalRefCount() == 0L) {
+      // Reference count reaches zero, add to pendingEvictionList
+      pendingEvictionQueue.add(((OmSnapshot) referenceCounted.get())
+          .getSnapshotTableKey());
+    }
+  }
+
   /**
    * Wrapper for cleanupInternal() that is synchronized to prevent multiple
    * threads from interleaving into the cleanup method.
    */
-  private synchronized void cleanup() {
+  private void cleanup() {
     if (dbMap.size() > cacheSizeLimit) {
       cleanupInternal();
     }
@@ -237,8 +264,9 @@ private synchronized void cleanup() {
    * TODO: [SNAPSHOT] Add new ozone debug CLI command to trigger this directly.
    */
   private void cleanupInternal() {
-    for (Map.Entry<String, ReferenceCounted<IOmMetadataReader, SnapshotCache>> entry : dbMap.entrySet()) {
-      dbMap.compute(entry.getKey(), (k, v) -> {
+    for (String evictionKey : pendingEvictionQueue) {
+      dbMap.compute(evictionKey, (k, v) -> {
+        pendingEvictionQueue.remove(k);
         if (v == null) {
           throw new IllegalStateException("Key '" + k + "' does not exist in cache. The RocksDB " +
               "instance of the Snapshot may not be closed properly.");
@@ -261,4 +289,6 @@ private void cleanupInternal() {
       });
     }
   }
+
+
 }

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotCache.java

@@ -82,7 +82,7 @@ static void beforeAll() throws Exception {
   void setUp() {
     // Reset cache for each test case
     snapshotCache = new SnapshotCache(
-        omSnapshotManager, cacheLoader, CACHE_SIZE_LIMIT);
+        omSnapshotManager, cacheLoader, CACHE_SIZE_LIMIT, 0);
   }
 
   @AfterEach

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java

@@ -394,7 +394,8 @@ public OmSnapshot load(@Nonnull String key) {
     omSnapshotManager = mock(OmSnapshotManager.class);
     when(omSnapshotManager.isSnapshotStatus(
         any(), any())).thenReturn(true);
-    snapshotCache = new SnapshotCache(omSnapshotManager, loader, 10);
+    snapshotCache = new SnapshotCache(omSnapshotManager, loader,
+        10, 0);
 
     snapshotDiffManager = new SnapshotDiffManager(db, differ, ozoneManager,
         snapshotCache, snapDiffJobTable, snapDiffReportTable,