HBASE-22114 Port HBASE-15560 (TinyLFU-based BlockCache) to branch-1

HBASE-15560 W-TinyLFU based BlockCache (Ben Manes)

Author: apurtell

hbase-common/src/main/java/org/apache/hadoop/hbase/io/util/HeapMemorySizeUtil.java

@@ -180,7 +180,7 @@ public static float getL2BlockCacheHeapPercent(Configuration conf) {
    * @return the number of bytes to use for LRU, negative if disabled.
    * @throws IllegalArgumentException if HFILE_BLOCK_CACHE_SIZE_KEY is > 1.0
    */
-  public static long getLruCacheSize(final Configuration conf) {
+  public static long getFirstLevelCacheSize(final Configuration conf, final long xmx) {
     float cachePercentage = conf.getFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY,
       HConstants.HFILE_BLOCK_CACHE_SIZE_DEFAULT);
     if (cachePercentage <= 0.0001f) {

hbase-common/src/main/resources/hbase-default.xml

@@ -806,6 +806,15 @@ possible configurations would overwhelm and obscure the important.
     <description>
       The default thread pool size if parallel-seeking feature enabled.</description>
   </property>
+  <property>
+    <name>hfile.block.cache.policy</name>
+    <value>LRU</value>
+    <description>The eviction policy for the L1 block cache (LRU or TinyLFU). If you want to
+      use TinyLFU you must build with JDK8+, run with JRE8+, and have both the
+      hbase-tinylfu-blockcache module and its dependency the caffiene library installed into
+      the classpath.
+    </description>
+  </property>
   <property>
     <name>hfile.block.cache.size</name>
     <value>0.4</value>

hbase-it/pom.xml

@@ -362,6 +362,21 @@
   </dependencies>
 
   <profiles>
+
+    <profile>
+      <id>build-with-jdk8</id>
+      <activation>
+        <jdk>1.8</jdk>
+      </activation>
+      <dependencies>
+        <dependency>
+          <groupId>org.apache.hbase</groupId>
+          <artifactId>hbase-tinylfu-blockcache</artifactId>
+          <version>${project.version}</version>
+        </dependency>
+      </dependencies>
+    </profile>
+
     <profile>
       <id>rsgroup</id>
       <activation>

hbase-resource-bundle/src/main/resources/supplemental-models.xml

@@ -342,6 +342,20 @@ under the License.
       </licenses>
     </project>
   </supplement>
+  <supplement>
+    <project>
+      <groupId>com.github.ben-manes.caffeine</groupId>
+      <artifactId>caffeine</artifactId>
+
+      <licenses>
+        <license>
+          <name>Apache License, Version 2.0</name>
+          <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+          <distribution>repo</distribution>
+        </license>
+      </licenses>
+    </project>
+  </supplement>
   <supplement>
     <project>
       <groupId>com.lmax</groupId>
@@ -2181,7 +2195,7 @@ Mozilla Public License Version 2.0
     means any form of the work other than Source Code Form.
 
 1.7. "Larger Work"
-    means a work that combines Covered Software with other material, in 
+    means a work that combines Covered Software with other material, in
     a separate file or files, that is not Covered Software.
 
 1.8. "License"
@@ -2612,4 +2626,17 @@ Copyright (c) 2007-2011 The JRuby project
       </licenses>
     </project>
   </supplement>
+  <supplement>
+    <project>
+      <groupId>com.google.errorprone</groupId>
+      <artifactId>error_prone_annotations</artifactId>
+      <licenses>
+        <license>
+          <name>Apache License, Version 2.0</name>
+          <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+          <distribution>repo</distribution>
+        </license>
+      </licenses>
+    </project>
+  </supplement>
 </supplementalDataModels>

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

@@ -22,7 +22,8 @@
 
 import java.io.IOException;
 import java.lang.management.ManagementFactory;
-import java.lang.management.MemoryUsage;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -45,6 +46,12 @@
 public class CacheConfig {
   private static final Log LOG = LogFactory.getLog(CacheConfig.class.getName());
 
+  /**
+   * Configuration key to cache block policy (Lru, TinyLfu).
+   */
+  public static final String HFILE_BLOCK_CACHE_POLICY_KEY = "hfile.block.cache.policy";
+  public static final String HFILE_BLOCK_CACHE_POLICY_DEFAULT = "LRU";
+
   /**
    * Configuration key to cache data blocks on write. There are separate
    * switches for bloom blocks and non-root index blocks.
@@ -92,19 +99,19 @@ public class CacheConfig {
    * is an in-memory map that needs to be persisted across restarts. Where to store this
    * in-memory state is what you supply here: e.g. <code>/tmp/bucketcache.map</code>.
    */
-  public static final String BUCKET_CACHE_PERSISTENT_PATH_KEY = 
+  public static final String BUCKET_CACHE_PERSISTENT_PATH_KEY =
       "hbase.bucketcache.persistent.path";
 
   /**
    * If the bucket cache is used in league with the lru on-heap block cache (meta blocks such
    * as indices and blooms are kept in the lru blockcache and the data blocks in the
    * bucket cache).
    */
-  public static final String BUCKET_CACHE_COMBINED_KEY = 
+  public static final String BUCKET_CACHE_COMBINED_KEY =
       "hbase.bucketcache.combinedcache.enabled";
 
   public static final String BUCKET_CACHE_WRITER_THREADS_KEY = "hbase.bucketcache.writer.threads";
-  public static final String BUCKET_CACHE_WRITER_QUEUE_KEY = 
+  public static final String BUCKET_CACHE_WRITER_QUEUE_KEY =
       "hbase.bucketcache.writer.queuelength";
 
   /**
@@ -177,6 +184,7 @@ private static enum ExternalBlockCaches {
     memcached("org.apache.hadoop.hbase.io.hfile.MemcachedBlockCache");
     // TODO(eclark): Consider more. Redis, etc.
     Class<? extends BlockCache> clazz;
+    @SuppressWarnings("unchecked")
     ExternalBlockCaches(String clazzName) {
       try {
         clazz = (Class<? extends BlockCache>) Class.forName(clazzName);
@@ -511,7 +519,9 @@ public boolean shouldCacheDataCompressed() {
    * @return true if this {@link BlockCategory} should be compressed in blockcache, false otherwise
    */
   public boolean shouldCacheCompressed(BlockCategory category) {
-    if (!isBlockCacheEnabled()) return false;
+    if (!isBlockCacheEnabled()) {
+      return false;
+    }
     switch (category) {
       case DATA:
         return this.cacheDataCompressed;
@@ -614,28 +624,62 @@ public String toString() {
   // Clear this if in tests you'd make more than one block cache instance.
   @VisibleForTesting
   static BlockCache GLOBAL_BLOCK_CACHE_INSTANCE;
-  private static LruBlockCache GLOBAL_L1_CACHE_INSTANCE = null;
-  private static BlockCache GLOBAL_L2_CACHE_INSTANCE = null;
+  private static FirstLevelBlockCache GLOBAL_L1_CACHE_INSTANCE;
+  private static BlockCache GLOBAL_L2_CACHE_INSTANCE;
+  private static ForkJoinPool GLOBAL_FORKJOIN_POOL;
 
   /** Boolean whether we have disabled the block cache entirely. */
   @VisibleForTesting
   static boolean blockCacheDisabled = false;
 
   /**
-   * @param c Configuration to use.
-   * @return An L1 instance.  Currently an instance of LruBlockCache.
+   * @param c Configuration to use
+   * @return An L1 instance
+   */
+  public static FirstLevelBlockCache getL1(final Configuration c) {
+    long xmx = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getMax();
+    long l1CacheSize = HeapMemorySizeUtil.getFirstLevelCacheSize(c, xmx);
+    return getL1(l1CacheSize, c);
+  }
+
+  /**
+   * @param c Configuration to use
+   * @param xmx Max heap memory
+   * @return An L1 instance
    */
-  private static synchronized LruBlockCache getL1(final Configuration c) {
+
+  private synchronized static FirstLevelBlockCache getL1(long cacheSize, Configuration c) {
     if (GLOBAL_L1_CACHE_INSTANCE != null) return GLOBAL_L1_CACHE_INSTANCE;
-    final long lruCacheSize = HeapMemorySizeUtil.getLruCacheSize(c);
-    if (lruCacheSize < 0) {
-      blockCacheDisabled = true;
+    if (cacheSize < 0) {
+      return null;
     }
-    if (blockCacheDisabled) return null;
+    String policy = c.get(HFILE_BLOCK_CACHE_POLICY_KEY, HFILE_BLOCK_CACHE_POLICY_DEFAULT);
     int blockSize = c.getInt(BLOCKCACHE_BLOCKSIZE_KEY, HConstants.DEFAULT_BLOCKSIZE);
-    LOG.info("Allocating LruBlockCache size=" +
-      StringUtils.byteDesc(lruCacheSize) + ", blockSize=" + StringUtils.byteDesc(blockSize));
-    GLOBAL_L1_CACHE_INSTANCE = new LruBlockCache(lruCacheSize, blockSize, true, c);
+    LOG.info("Allocating BlockCache size=" +
+        StringUtils.byteDesc(cacheSize) + ", blockSize=" + StringUtils.byteDesc(blockSize));
+    if (policy.equalsIgnoreCase("LRU")) {
+      GLOBAL_L1_CACHE_INSTANCE = new LruBlockCache(cacheSize, blockSize, true, c);
+    } else if (policy.equalsIgnoreCase("TinyLFU")) {
+      if (GLOBAL_FORKJOIN_POOL == null) {
+        GLOBAL_FORKJOIN_POOL = new ForkJoinPool();
+      }
+      Class<?> tinyLFUClass;
+      try {
+        tinyLFUClass = Class.forName("org.apache.hadoop.hbase.io.hfile.TinyLfuBlockCache");
+        GLOBAL_L1_CACHE_INSTANCE = (FirstLevelBlockCache)
+            tinyLFUClass.getDeclaredConstructor(long.class, long.class, Executor.class,
+              Configuration.class)
+            .newInstance(cacheSize, blockSize, GLOBAL_FORKJOIN_POOL, c);
+      } catch (Exception e) {
+        throw new RuntimeException(
+          "Unable to instantiate the TinyLfuBlockCache block cache policy." +
+          "If you want to use TinyLFU you must build with JDK8+, run with JRE8+, and have both " +
+          "the hbase-tinylfu-blockcache module and its dependency the caffiene library " +
+          "installed into the classpath.", e);
+      }
+    } else {
+      throw new IllegalArgumentException("Unknown block cache policy " + policy);
+    }
     return GLOBAL_L1_CACHE_INSTANCE;
   }
 
@@ -676,7 +720,7 @@ public CacheStats getL2Stats() {
   }
 
   private static BlockCache getExternalBlockcache(Configuration c) {
-    Class klass = null;
+    Class<?> klass = null;
 
     // Get the class, from the config. s
     try {
@@ -704,7 +748,9 @@ private static BlockCache getExternalBlockcache(Configuration c) {
   private static BlockCache getBucketCache(Configuration c) {
     // Check for L2.  ioengine name must be non-null.
     String bucketCacheIOEngineName = c.get(BUCKET_CACHE_IOENGINE_KEY, null);
-    if (bucketCacheIOEngineName == null || bucketCacheIOEngineName.length() <= 0) return null;
+    if (bucketCacheIOEngineName == null || bucketCacheIOEngineName.length() <= 0) {
+      return null;
+    }
 
     int blockSize = c.getInt(BLOCKCACHE_BLOCKSIZE_KEY, HConstants.DEFAULT_BLOCKSIZE);
     final long bucketCacheSize = HeapMemorySizeUtil.getBucketCacheSize(c);
@@ -762,33 +808,35 @@ private static BlockCache getBucketCache(Configuration c) {
    * @return The block cache or <code>null</code>.
    */
   public static synchronized BlockCache instantiateBlockCache(Configuration conf) {
-    if (GLOBAL_BLOCK_CACHE_INSTANCE != null) return GLOBAL_BLOCK_CACHE_INSTANCE;
-    if (blockCacheDisabled) return null;
+    if (GLOBAL_BLOCK_CACHE_INSTANCE != null) {
+      return GLOBAL_BLOCK_CACHE_INSTANCE;
+    }
+    if (blockCacheDisabled) {
+      return null;
+    }
     if (conf.get(DEPRECATED_BLOCKCACHE_BLOCKSIZE_KEY) != null) {
       LOG.warn("The config key " + DEPRECATED_BLOCKCACHE_BLOCKSIZE_KEY +
           " is deprecated now, instead please use " + BLOCKCACHE_BLOCKSIZE_KEY  +". "
           + "In future release we will remove the deprecated config.");
     }
-    LruBlockCache l1 = getL1(conf);
-    // blockCacheDisabled is set as a side-effect of getL1Internal(), so check it again after the call.
-    if (blockCacheDisabled) return null;
+    FirstLevelBlockCache l1 = getL1(conf);
     BlockCache l2 = getL2(conf);
     if (l2 == null) {
       GLOBAL_BLOCK_CACHE_INSTANCE = l1;
     } else {
       boolean useExternal = conf.getBoolean(EXTERNAL_BLOCKCACHE_KEY, EXTERNAL_BLOCKCACHE_DEFAULT);
-      boolean combinedWithLru = conf.getBoolean(BUCKET_CACHE_COMBINED_KEY,
+      boolean combinedWithL1 = conf.getBoolean(BUCKET_CACHE_COMBINED_KEY,
         DEFAULT_BUCKET_CACHE_COMBINED);
       if (useExternal) {
         GLOBAL_BLOCK_CACHE_INSTANCE = new InclusiveCombinedBlockCache(l1, l2);
       } else {
-        if (combinedWithLru) {
+        if (combinedWithL1) {
           GLOBAL_BLOCK_CACHE_INSTANCE = new CombinedBlockCache(l1, l2);
         } else {
-          // L1 and L2 are not 'combined'.  They are connected via the LruBlockCache victimhandler
-          // mechanism.  It is a little ugly but works according to the following: when the
-          // background eviction thread runs, blocks evicted from L1 will go to L2 AND when we get
-          // a block from the L1 cache, if not in L1, we will search L2.
+          // L1 and L2 are not 'combined'. They are connected via the FirstLevelBlockCache
+          // victimhandler mechanism. It is a little ugly but works according to the following:
+          // when the background eviction thread runs, blocks evicted from L1 will go to L2 AND when
+          // we get a block from the L1 cache, if not in L1, we will search L2.
           GLOBAL_BLOCK_CACHE_INSTANCE = l1;
         }
       }