HBASE-30104 Refactor LruBlockCache to implement CacheEngine

dev-support/spotbugs-exclude.xml

@@ -247,6 +247,11 @@
     <Bug pattern="SC_START_IN_CTOR"/>
   </Match>
 
+  <Match>
+    <Class name="org.apache.hadoop.hbase.io.hfile.cache.LruCacheEngine"/>
+    <Bug pattern="SC_START_IN_CTOR"/>
+  </Match>
+
   <Match>
     <Class name="org.apache.hadoop.hbase.io.hfile.LruAdaptiveBlockCache"/>
     <Bug pattern="SC_START_IN_CTOR"/>

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

@@ -27,6 +27,9 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache;
+import org.apache.hadoop.hbase.io.hfile.cache.CacheEngine;
+import org.apache.hadoop.hbase.io.hfile.cache.CacheEngines;
+import org.apache.hadoop.hbase.io.hfile.cache.LruCacheEngine;
 import org.apache.hadoop.hbase.io.util.MemorySizeUtil;
 import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.util.ReflectionUtils;
@@ -71,8 +74,8 @@ public final class BlockCacheFactory {
    */
   public static final String BLOCKCACHE_BLOCKSIZE_KEY = "hbase.blockcache.minblocksize";
 
-  private static final String EXTERNAL_BLOCKCACHE_KEY = "hbase.blockcache.use.external";
-  private static final boolean EXTERNAL_BLOCKCACHE_DEFAULT = false;
+  public static final String EXTERNAL_BLOCKCACHE_KEY = "hbase.blockcache.use.external";
+  public static final boolean EXTERNAL_BLOCKCACHE_DEFAULT = false;
 
   private static final String EXTERNAL_BLOCKCACHE_CLASS_KEY = "hbase.blockcache.external.class";
 
@@ -135,6 +138,62 @@ public static BlockCache createBlockCache(Configuration conf) {
     return createBlockCache(conf, null);
   }
 
+  /**
+   * Creates the configured first-level cache as a cache engine.
+   * <p>
+   * Cache implementations that have been migrated to {@link CacheEngine} are instantiated directly.
+   * Legacy {@link FirstLevelBlockCache} implementations are adapted until their migration is
+   * complete.
+   * @param c cache configuration
+   * @return first-level cache engine, or {@code null} when the on-heap cache is disabled
+   */
+  public static CacheEngine createFirstLevelCacheEngine(final Configuration c) {
+    final long cacheSize = MemorySizeUtil.getOnHeapCacheSize(c);
+    if (cacheSize < 0) {
+      return null;
+    }
+
+    String policy = c.get(BLOCKCACHE_POLICY_KEY, BLOCKCACHE_POLICY_DEFAULT);
+    int blockSize = c.getInt(BLOCKCACHE_BLOCKSIZE_KEY, HConstants.DEFAULT_BLOCKSIZE);
+    LOG.info("Allocating CacheEngine size=" + StringUtils.byteDesc(cacheSize) + ", blockSize="
+      + StringUtils.byteDesc(blockSize));
+
+    if (policy.equalsIgnoreCase("LRU")) {
+      return new LruCacheEngine(cacheSize, blockSize, true, c);
+    } else if (policy.equalsIgnoreCase("IndexOnlyLRU")) {
+      return CacheEngines.fromBlockCache(new IndexOnlyLruBlockCache(cacheSize, blockSize, true, c));
+    } else if (policy.equalsIgnoreCase("TinyLFU")) {
+      return CacheEngines
+        .fromBlockCache(new TinyLfuBlockCache(cacheSize, blockSize, ForkJoinPool.commonPool(), c));
+    } else if (policy.equalsIgnoreCase("AdaptiveLRU")) {
+      return CacheEngines.fromBlockCache(new LruAdaptiveBlockCache(cacheSize, blockSize, true, c));
+    } else {
+      throw new IllegalArgumentException("Unknown policy: " + policy);
+    }
+  }
+
+  /**
+   * Creates the configured external second-level cache as a cache engine.
+   * @param c cache configuration
+   * @return external cache engine, or {@code null} when no external cache can be created
+   */
+  public static CacheEngine createExternalCacheEngine(Configuration c) {
+    BlockCache blockCache = createExternalBlockcache(c);
+    return blockCache == null ? null : CacheEngines.fromBlockCache(blockCache);
+  }
+
+  /**
+   * Creates the configured bucket cache as a cache engine.
+   * @param c             cache configuration
+   * @param onlineRegions currently online regions
+   * @return bucket cache engine, or {@code null} when BucketCache is disabled
+   */
+  public static CacheEngine createBucketCacheEngine(Configuration c,
+    Map<String, HRegion> onlineRegions) {
+    BucketCache bucketCache = createBucketCache(c, onlineRegions);
+    return bucketCache == null ? null : CacheEngines.fromBlockCache(bucketCache);
+  }
+
   private static FirstLevelBlockCache createFirstLevelCache(final Configuration c) {
     final long cacheSize = MemorySizeUtil.getOnHeapCacheSize(c);
     if (cacheSize < 0) {

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

@@ -30,6 +30,7 @@
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.ConcurrentSkipListSet;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.io.hfile.cache.CacheEngine;
 import org.apache.hadoop.hbase.metrics.impl.FastLongHistogram;
 import org.apache.hadoop.hbase.nio.ByteBuff;
 import org.apache.hadoop.hbase.regionserver.HRegion;
@@ -248,6 +249,51 @@ public static boolean shouldReplaceExistingCacheBlock(BlockCache blockCache,
     }
   }
 
+  /**
+   * Because of region splitting, it is possible that the split key is located in the middle of a
+   * block. As a result, both daughter regions may load the same block from their parent HFile.
+   * <p>
+   * When using positional reads, HBase does not force the read to include the complete next-block
+   * header. Therefore, when two threads try to cache the same block, one thread may have read the
+   * complete next-block header while the other did not. If the already cached block does not
+   * contain {@code nextBlockOnDiskSize} but the new block does, replacing the existing block with
+   * the new one improves subsequent read performance. See HBASE-20447.
+   * </p>
+   * @param cacheEngine cache engine to check
+   * @param cacheKey    block cache key
+   * @param newBlock    new block being considered for insertion
+   * @return {@code true} if the existing cached block should be replaced by {@code newBlock};
+   *         {@code false} if the existing cached block should be retained
+   */
+  public static boolean shouldReplaceExistingCacheBlock(CacheEngine cacheEngine,
+    BlockCacheKey cacheKey, Cacheable newBlock) {
+    // NOTICE: getBlock retains the existingBlock before returning it.
+    Cacheable existingBlock = cacheEngine.getBlock(cacheKey, false, false, false);
+    if (existingBlock == null) {
+      return true;
+    }
+
+    try {
+      int comparison = BlockCacheUtil.validateBlockAddition(existingBlock, newBlock, cacheKey);
+      if (comparison < 0) {
+        LOG.warn("Cached block contents differ by nextBlockOnDiskSize, the new block has "
+          + "nextBlockOnDiskSize set. Caching new block.");
+        return true;
+      } else if (comparison > 0) {
+        LOG.warn("Cached block contents differ by nextBlockOnDiskSize, the existing block has "
+          + "nextBlockOnDiskSize set. Keeping cached block.");
+        return false;
+      } else {
+        LOG.debug("Caching an already cached block: {}. This is harmless and can happen in rare "
+          + "cases (see HBASE-8547)", cacheKey);
+        return false;
+      }
+    } finally {
+      // Release the reference retained by CacheEngine#getBlock.
+      existingBlock.release();
+    }
+  }
+
   public static Set<String> listAllFilesNames(Map<String, HRegion> onlineRegions) {
     Set<String> files = new HashSet<>();
     onlineRegions.values().forEach(r -> {

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

@@ -320,6 +320,7 @@ private void initFromConf(Configuration conf, ColumnFamilyDescriptor family) {
 
   /**
    * Constructs a cache configuration copied from the specified configuration.
+   * @param cacheConf cache configuration to copy
    */
   public CacheConfig(CacheConfig cacheConf) {
     this.cacheDataOnRead = cacheConf.cacheDataOnRead;
@@ -337,9 +338,7 @@ public CacheConfig(CacheConfig cacheConf) {
     this.blockCache = cacheConf.blockCache;
     this.byteBuffAllocator = cacheConf.byteBuffAllocator;
     this.heapUsageThreshold = cacheConf.heapUsageThreshold;
-    this.cacheAccessService = blockCache != null
-      ? CacheAccessServices.fromBlockCache(blockCache)
-      : CacheAccessServices.disabled();
+    this.cacheAccessService = cacheConf.cacheAccessService;
   }
 
   private CacheConfig() {
@@ -551,8 +550,12 @@ public boolean shouldLockOnCacheMiss(BlockType blockType) {
   }
 
   /**
-   * Returns the block cache.
-   * @return the block cache, or null if caching is completely disabled
+   * Returns the legacy block cache when this configuration exposes one.
+   * <p>
+   * Native cache-engine configurations do not necessarily expose a {@link BlockCache}. Callers
+   * should use {@link #getCacheAccessService()} for cache access and diagnostics.
+   * </p>
+   * @return legacy block cache when available
    */
   public Optional<BlockCache> getBlockCache() {
     return Optional.ofNullable(this.blockCache);
@@ -561,10 +564,9 @@ public Optional<BlockCache> getBlockCache() {
   /**
    * Returns the cache access service used by HFile read/write path callers.
    * <p>
-   * This service is the migration-facing cache abstraction. For now it is backed by the existing
-   * {@link BlockCache} when block cache is configured, or by a disabled no-op implementation when
-   * block cache is unavailable. This keeps cache construction unchanged while allowing callers such
-   * as {@code HFileReaderImpl} to depend on {@link CacheAccessService}.
+   * The cache access service is the primary cache abstraction exposed by this configuration. It may
+   * be backed by native {@link CacheEngine} implementations, legacy {@link BlockCache}
+   * implementations adapted as cache engines, or a combination of both.
    * </p>
    * @return cache access service
    */
@@ -583,10 +585,10 @@ private static boolean isCombinedBlockCacheCompatible(CacheAccessService cacheAc
     if (!(cacheAccessService instanceof TopologyBackedCacheAccessService)) {
       return false;
     }
-
     TopologyBackedCacheAccessService service =
       (TopologyBackedCacheAccessService) cacheAccessService;
-    return service.getTopology().getType() == CacheTopologyType.TIERED_EXCLUSIVE;
+    CacheTopology topology = service.getTopology();
+    return topology.getTiers().contains(CacheTier.L1) && topology.getTiers().contains(CacheTier.L2);
   }
 
   public ByteBuffAllocator getByteBuffAllocator() {

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

@@ -2084,7 +2084,7 @@ public boolean equals(Object comparison) {
     return true;
   }
 
-  DataBlockEncoding getDataBlockEncoding() {
+  public DataBlockEncoding getDataBlockEncoding() {
     if (blockType == BlockType.ENCODED_DATA) {
       return DataBlockEncoding.getEncodingById(getDataBlockEncodingId());
     }
@@ -2211,7 +2211,7 @@ private static HFileBlock shallowClone(HFileBlock blk, ByteBuff newBuf) {
     return createBuilder(blk, newBuf).build();
   }
 
-  static HFileBlock deepCloneOnHeap(HFileBlock blk) {
+  public static HFileBlock deepCloneOnHeap(HFileBlock blk) {
     ByteBuff deepCloned = ByteBuff
       .wrap(ByteBuffer.wrap(blk.bufWithoutChecksum.toBytes(0, blk.bufWithoutChecksum.limit())));
     return createBuilder(blk, deepCloned).build();