HBASE-28170 Put the cached time at the beginning of the block; run cache validation in the background when retrieving the persistent cache (#5471)

Signed-off-by: Peter Somogyi <psomogyi@apache.org>

(cherry-picked from commit 41057bb)

Author: wchevreuil

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFilePreadReader.java

@@ -40,7 +40,9 @@ public HFilePreadReader(ReaderContext context, HFileInfo fileInfo, CacheConfig c
     Configuration conf) throws IOException {
     super(context, fileInfo, cacheConf, conf);
     final MutableBoolean fileAlreadyCached = new MutableBoolean(false);
-    BucketCache.getBuckedCacheFromCacheConfig(cacheConf).ifPresent(bc -> fileAlreadyCached
+    Optional<BucketCache> bucketCacheOptional =
+      BucketCache.getBucketCacheFromCacheConfig(cacheConf);
+    bucketCacheOptional.ifPresent(bc -> fileAlreadyCached
       .setValue(bc.getFullyCachedFiles().get(path.getName()) == null ? false : true));
     // Prefetch file blocks upon open if requested
     if (
@@ -65,8 +67,6 @@ public void run() {
             if (LOG.isTraceEnabled()) {
               LOG.trace("Prefetch start " + getPathOffsetEndStr(path, offset, end));
             }
-            Optional<BucketCache> bucketCacheOptional =
-              BucketCache.getBuckedCacheFromCacheConfig(cacheConf);
             // Don't use BlockIterator here, because it's designed to read load-on-open section.
             long onDiskSizeOfNextBlock = -1;
             while (offset < end) {
@@ -78,21 +78,24 @@ public void run() {
               // to the next block without actually going read all the way to the cache.
               if (bucketCacheOptional.isPresent()) {
                 BucketCache cache = bucketCacheOptional.get();
-                BlockCacheKey cacheKey = new BlockCacheKey(name, offset);
-                BucketEntry entry = cache.getBackingMap().get(cacheKey);
-                if (entry != null) {
-                  cacheKey = new BlockCacheKey(name, offset);
-                  entry = cache.getBackingMap().get(cacheKey);
-                  if (entry == null) {
-                    LOG.debug("No cache key {}, we'll read and cache it", cacheKey);
+                if (cache.getBackingMapValidated().get()) {
+                  BlockCacheKey cacheKey = new BlockCacheKey(name, offset);
+                  BucketEntry entry = cache.getBackingMap().get(cacheKey);
+                  if (entry != null) {
+                    cacheKey = new BlockCacheKey(name, offset);
+                    entry = cache.getBackingMap().get(cacheKey);
+                    if (entry == null) {
+                      LOG.debug("No cache key {}, we'll read and cache it", cacheKey);
+                    } else {
+                      offset += entry.getOnDiskSizeWithHeader();
+                      LOG.debug(
+                        "Found cache key {}. Skipping prefetch, the block is already cached.",
+                        cacheKey);
+                      continue;
+                    }
                   } else {
-                    offset += entry.getOnDiskSizeWithHeader();
-                    LOG.debug("Found cache key {}. Skipping prefetch, the block is already cached.",
-                      cacheKey);
-                    continue;
+                    LOG.debug("No entry in the backing map for cache key {}", cacheKey);
                   }
-                } else {
-                  LOG.debug("No entry in the backing map for cache key {}", cacheKey);
                 }
               }
               // Perhaps we got our block from cache? Unlikely as this may be, if it happens, then
@@ -111,9 +114,7 @@ public void run() {
                 block.release();
               }
             }
-            BucketCache.getBuckedCacheFromCacheConfig(cacheConf)
-              .ifPresent(bc -> bc.fileCacheCompleted(path.getName()));
-
+            bucketCacheOptional.ifPresent(bc -> bc.fileCacheCompleted(path.getName()));
           } catch (IOException e) {
             // IOExceptions are probably due to region closes (relocation, etc.)
             if (LOG.isTraceEnabled()) {

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java

@@ -144,6 +144,8 @@ public class BucketCache implements BlockCache, HeapSize {
   // In this map, store the block's meta data like offset, length
   transient Map<BlockCacheKey, BucketEntry> backingMap;
 
+  private AtomicBoolean backingMapValidated = new AtomicBoolean(false);
+
   /** Set of files for which prefetch is completed */
   final Map<String, Boolean> fullyCachedFiles = new ConcurrentHashMap<>();
 
@@ -301,7 +303,6 @@ public BucketCache(String ioEngineName, long capacity, int blockSize, int[] buck
 
     this.allocFailLogPrevTs = 0;
 
-    bucketAllocator = new BucketAllocator(capacity, bucketSizes);
     for (int i = 0; i < writerThreads.length; ++i) {
       writerQueues.add(new ArrayBlockingQueue<>(writerQLen));
     }
@@ -318,10 +319,14 @@ public BucketCache(String ioEngineName, long capacity, int blockSize, int[] buck
       try {
         retrieveFromFile(bucketSizes);
       } catch (IOException ioex) {
+        LOG.error("Can't restore from file[{}] because of ", persistencePath, ioex);
         backingMap.clear();
         fullyCachedFiles.clear();
-        LOG.error("Can't restore from file[" + persistencePath + "] because of ", ioex);
+        backingMapValidated.set(true);
+        bucketAllocator = new BucketAllocator(capacity, bucketSizes);
       }
+    } else {
+      bucketAllocator = new BucketAllocator(capacity, bucketSizes);
     }
     final String threadName = Thread.currentThread().getName();
     this.cacheEnabled = true;
@@ -374,6 +379,7 @@ protected void startWriterThreads() {
   }
 
   void startBucketCachePersisterThread() {
+    LOG.info("Starting BucketCachePersisterThread");
     cachePersister = new BucketCachePersister(this, bucketcachePersistInterval);
     cachePersister.setDaemon(true);
     cachePersister.start();
@@ -529,6 +535,7 @@ protected void cacheBlockWithWaitInternal(BlockCacheKey cacheKey, Cacheable cach
     } else {
       this.blockNumber.increment();
       this.heapSize.add(cachedItem.heapSize());
+      blocksByHFile.add(cacheKey);
     }
   }
 
@@ -589,6 +596,7 @@ public Cacheable getBlock(BlockCacheKey key, boolean caching, boolean repeat,
         // the cache map state might differ from the actual cache. If we reach this block,
         // we should remove the cache key entry from the backing map
         backingMap.remove(key);
+        fullyCachedFiles.remove(key.getHfileName());
         LOG.debug("Failed to fetch block for cache key: {}.", key, hioex);
       } catch (IOException ioex) {
         LOG.error("Failed reading block " + key + " from bucket cache", ioex);
@@ -684,6 +692,7 @@ private boolean doEvictBlock(BlockCacheKey cacheKey, BucketEntry bucketEntry,
     } else {
       return bucketEntryToUse.withWriteLock(offsetLock, () -> {
         if (backingMap.remove(cacheKey, bucketEntryToUse)) {
+          LOG.debug("removed key {} from back map in the evict process", cacheKey);
           blockEvicted(cacheKey, bucketEntryToUse, !existedInRamCache, evictedByEvictionProcess);
           return true;
         }
@@ -1255,30 +1264,43 @@ static List<RAMQueueEntry> getRAMQueueEntries(BlockingQueue<RAMQueueEntry> q,
   @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "OBL_UNSATISFIED_OBLIGATION",
       justification = "false positive, try-with-resources ensures close is called.")
   void persistToFile() throws IOException {
+    LOG.debug("Thread {} started persisting bucket cache to file",
+      Thread.currentThread().getName());
     if (!isCachePersistent()) {
       throw new IOException("Attempt to persist non-persistent cache mappings!");
     }
     File tempPersistencePath = new File(persistencePath + EnvironmentEdgeManager.currentTime());
     try (FileOutputStream fos = new FileOutputStream(tempPersistencePath, false)) {
       fos.write(ProtobufMagic.PB_MAGIC);
       BucketProtoUtils.toPB(this).writeDelimitedTo(fos);
+    } catch (IOException e) {
+      LOG.error("Failed to persist bucket cache to file", e);
+      throw e;
     }
+    LOG.debug("Thread {} finished persisting bucket cache to file, renaming",
+      Thread.currentThread().getName());
     if (!tempPersistencePath.renameTo(new File(persistencePath))) {
       LOG.warn("Failed to commit cache persistent file. We might lose cached blocks if "
         + "RS crashes/restarts before we successfully checkpoint again.");
     }
   }
 
-  private boolean isCachePersistent() {
+  public boolean isCachePersistent() {
     return ioEngine.isPersistent() && persistencePath != null;
   }
 
   /**
    * @see #persistToFile()
    */
   private void retrieveFromFile(int[] bucketSizes) throws IOException {
+    LOG.info("Started retrieving bucket cache from file");
     File persistenceFile = new File(persistencePath);
     if (!persistenceFile.exists()) {
+      LOG.warn("Persistence file missing! "
+        + "It's ok if it's first run after enabling persistent cache.");
+      bucketAllocator = new BucketAllocator(cacheCapacity, bucketSizes, backingMap, realCacheSize);
+      blockNumber.add(backingMap.size());
+      backingMapValidated.set(true);
       return;
     }
     assert !cacheEnabled;
@@ -1296,6 +1318,7 @@ private void retrieveFromFile(int[] bucketSizes) throws IOException {
       parsePB(BucketCacheProtos.BucketCacheEntry.parseDelimitedFrom(in));
       bucketAllocator = new BucketAllocator(cacheCapacity, bucketSizes, backingMap, realCacheSize);
       blockNumber.add(backingMap.size());
+      LOG.info("Bucket cache retrieved from file successfully");
     }
   }
 
@@ -1368,27 +1391,43 @@ private void parsePB(BucketCacheProtos.BucketCacheEntry proto) throws IOExceptio
       try {
         ((PersistentIOEngine) ioEngine).verifyFileIntegrity(proto.getChecksum().toByteArray(),
           algorithm);
+        backingMapValidated.set(true);
       } catch (IOException e) {
         LOG.warn("Checksum for cache file failed. "
-          + "We need to validate each cache key in the backing map. This may take some time...");
-        long startTime = EnvironmentEdgeManager.currentTime();
-        int totalKeysOriginally = backingMap.size();
-        for (Map.Entry<BlockCacheKey, BucketEntry> keyEntry : backingMap.entrySet()) {
-          try {
-            ((FileIOEngine) ioEngine).checkCacheTime(keyEntry.getValue());
-          } catch (IOException e1) {
-            LOG.debug("Check for key {} failed. Removing it from map.", keyEntry.getKey());
-            backingMap.remove(keyEntry.getKey());
-            fullyCachedFiles.remove(keyEntry.getKey().getHfileName());
+          + "We need to validate each cache key in the backing map. "
+          + "This may take some time, so we'll do it in a background thread,");
+        Runnable cacheValidator = () -> {
+          while (bucketAllocator == null) {
+            try {
+              Thread.sleep(50);
+            } catch (InterruptedException ex) {
+              throw new RuntimeException(ex);
+            }
           }
-        }
-        LOG.info("Finished validating {} keys in the backing map. Recovered: {}. This took {}ms.",
-          totalKeysOriginally, backingMap.size(),
-          (EnvironmentEdgeManager.currentTime() - startTime));
+          long startTime = EnvironmentEdgeManager.currentTime();
+          int totalKeysOriginally = backingMap.size();
+          for (Map.Entry<BlockCacheKey, BucketEntry> keyEntry : backingMap.entrySet()) {
+            try {
+              ((FileIOEngine) ioEngine).checkCacheTime(keyEntry.getValue());
+            } catch (IOException e1) {
+              LOG.debug("Check for key {} failed. Evicting.", keyEntry.getKey());
+              evictBlock(keyEntry.getKey());
+              fullyCachedFiles.remove(keyEntry.getKey().getHfileName());
+            }
+          }
+          backingMapValidated.set(true);
+          LOG.info("Finished validating {} keys in the backing map. Recovered: {}. This took {}ms.",
+            totalKeysOriginally, backingMap.size(),
+            (EnvironmentEdgeManager.currentTime() - startTime));
+        };
+        Thread t = new Thread(cacheValidator);
+        t.setDaemon(true);
+        t.start();
       }
     } else {
       // if has not checksum, it means the persistence file is old format
       LOG.info("Persistent file is old format, it does not support verifying file integrity!");
+      backingMapValidated.set(true);
     }
     verifyCapacityAndClasses(proto.getCacheCapacity(), proto.getIoClass(), proto.getMapClass());
   }
@@ -1417,6 +1456,7 @@ private void checkIOErrorIsTolerated() {
    */
   private void disableCache() {
     if (!cacheEnabled) return;
+    LOG.info("Disabling cache");
     cacheEnabled = false;
     ioEngine.shutdown();
     this.scheduleThreadPool.shutdown();
@@ -1441,11 +1481,15 @@ public void shutdown() {
     LOG.info("Shutdown bucket cache: IO persistent=" + ioEngine.isPersistent() + "; path to write="
       + persistencePath);
     if (ioEngine.isPersistent() && persistencePath != null) {
-      if (cachePersister != null) {
-        cachePersister.interrupt();
-      }
       try {
         join();
+        if (cachePersister != null) {
+          LOG.info("Shutting down cache persister thread.");
+          cachePersister.shutdown();
+          while (cachePersister.isAlive()) {
+            Thread.sleep(10);
+          }
+        }
         persistToFile();
       } catch (IOException ex) {
         LOG.error("Unable to persist data on exit: " + ex.toString(), ex);
@@ -1650,17 +1694,17 @@ public BucketEntry writeToCache(final IOEngine ioEngine, final BucketAllocator a
           HFileBlock block = (HFileBlock) data;
           ByteBuff sliceBuf = block.getBufferReadOnly();
           block.getMetaData(metaBuff);
-          ioEngine.write(sliceBuf, offset);
-          // adds the cache time after the block and metadata part
+          // adds the cache time prior to the block and metadata part
           if (isCachePersistent) {
-            ioEngine.write(metaBuff, offset + len - metaBuff.limit() - Long.BYTES);
             ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES);
             buffer.putLong(bucketEntry.getCachedTime());
             buffer.rewind();
-            ioEngine.write(buffer, (offset + len - Long.BYTES));
+            ioEngine.write(buffer, offset);
+            ioEngine.write(sliceBuf, (offset + Long.BYTES));
           } else {
-            ioEngine.write(metaBuff, offset + len - metaBuff.limit());
+            ioEngine.write(sliceBuf, offset);
           }
+          ioEngine.write(metaBuff, offset + len - metaBuff.limit());
         } else {
           // Only used for testing.
           ByteBuffer bb = ByteBuffer.allocate(len);
@@ -1902,11 +1946,15 @@ public Map<BlockCacheKey, BucketEntry> getBackingMap() {
     return backingMap;
   }
 
+  public AtomicBoolean getBackingMapValidated() {
+    return backingMapValidated;
+  }
+
   public Map<String, Boolean> getFullyCachedFiles() {
     return fullyCachedFiles;
   }
 
-  public static Optional<BucketCache> getBuckedCacheFromCacheConfig(CacheConfig cacheConf) {
+  public static Optional<BucketCache> getBucketCacheFromCacheConfig(CacheConfig cacheConf) {
     if (cacheConf.getBlockCache().isPresent()) {
       BlockCache bc = cacheConf.getBlockCache().get();
       if (bc instanceof CombinedBlockCache) {

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCachePersister.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hbase.io.hfile.bucket;
 
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -28,6 +29,8 @@ public class BucketCachePersister extends Thread {
   private final long intervalMillis;
   private static final Logger LOG = LoggerFactory.getLogger(BucketCachePersister.class);
 
+  private AtomicBoolean shutdown = new AtomicBoolean(false);
+
   public BucketCachePersister(BucketCache cache, long intervalMillis) {
     super("bucket-cache-persister");
     this.cache = cache;
@@ -36,20 +39,34 @@ public BucketCachePersister(BucketCache cache, long intervalMillis) {
   }
 
   public void run() {
-    while (true) {
-      try {
-        Thread.sleep(intervalMillis);
-        if (cache.isCacheInconsistent()) {
-          LOG.debug("Cache is inconsistent, persisting to disk");
-          cache.persistToFile();
-          cache.setCacheInconsistent(false);
+    try {
+      while (true) {
+        try {
+          Thread.sleep(intervalMillis);
+          if (cache.isCacheInconsistent()) {
+            LOG.debug("Cache is inconsistent, persisting to disk");
+            cache.persistToFile();
+            cache.setCacheInconsistent(false);
+          }
+          // Thread.interrupt may cause an InterruptException inside util method used for checksum
+          // calculation in persistToFile. This util currently swallows the exception, causing this
+          // thread to net get interrupt, so we added this flag to indicate the persister thread
+          // should stop.
+          if (shutdown.get()) {
+            break;
+          }
+        } catch (IOException e) {
+          LOG.warn("Exception in BucketCachePersister.", e);
         }
-      } catch (IOException e) {
-        LOG.warn("IOException in BucketCachePersister {} ", e.getMessage());
-      } catch (InterruptedException iex) {
-        LOG.warn("InterruptedException in BucketCachePersister {} ", iex.getMessage());
-        break;
       }
+      LOG.info("Finishing cache persister thread.");
+    } catch (InterruptedException e) {
+      LOG.warn("Interrupting BucketCachePersister thread.", e);
     }
   }
+
+  public void shutdown() {
+    this.shutdown.set(true);
+    this.interrupt();
+  }
 }