accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dlmar...@apache.org
Subject [10/13] accumulo git commit: ACCUMULO-4463: Make block cache implementation pluggable
Date Tue, 23 May 2017 12:22:56 GMT
ACCUMULO-4463: Make block cache implementation pluggable


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/452732c8
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/452732c8
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/452732c8

Branch: refs/heads/IGNITE
Commit: 452732c8ad91c4a73a556cdc2030e0a5cbbd173b
Parents: 3f1b0f3
Author: Dave Marion <dlmarion@apache.org>
Authored: Mon May 22 13:08:41 2017 -0400
Committer: Dave Marion <dlmarion@apache.org>
Committed: Mon May 22 13:08:41 2017 -0400

----------------------------------------------------------------------
 core/src/main/findbugs/exclude-filter.xml       |   4 +-
 .../core/client/rfile/RFileScanner.java         |  57 +-
 .../org/apache/accumulo/core/conf/Property.java |   4 +-
 .../core/file/blockfile/cache/BlockCache.java   |   6 +
 .../cache/BlockCacheConfiguration.java          |  91 +++
 .../file/blockfile/cache/BlockCacheManager.java | 107 +++
 .../core/file/blockfile/cache/CacheType.java    |  24 +
 .../core/file/blockfile/cache/CachedBlock.java  |   2 +-
 .../file/blockfile/cache/LruBlockCache.java     | 703 -------------------
 .../file/blockfile/cache/TinyLfuBlockCache.java | 141 ----
 .../file/blockfile/cache/lru/LruBlockCache.java | 625 +++++++++++++++++
 .../cache/lru/LruBlockCacheConfiguration.java   | 212 ++++++
 .../cache/lru/LruBlockCacheManager.java         |  49 ++
 .../cache/tinylfu/TinyLfuBlockCache.java        | 150 ++++
 .../tinylfu/TinyLfuBlockCacheConfiguration.java |  32 +
 .../cache/tinylfu/TinyLfuBlockCacheManager.java |  37 +
 .../accumulo/core/summary/SummaryReader.java    |   5 +
 .../blockfile/cache/BlockCacheFactoryTest.java  |  53 ++
 .../cache/BlockConfigurationHelperTest.java     |  32 +
 .../file/blockfile/cache/TestLruBlockCache.java | 194 ++---
 .../accumulo/core/file/rfile/RFileTest.java     |  25 +-
 .../tserver/TabletServerResourceManager.java    |  42 +-
 22 files changed, 1625 insertions(+), 970 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/findbugs/exclude-filter.xml
----------------------------------------------------------------------
diff --git a/core/src/main/findbugs/exclude-filter.xml b/core/src/main/findbugs/exclude-filter.xml
index 95aca41..56d3659 100644
--- a/core/src/main/findbugs/exclude-filter.xml
+++ b/core/src/main/findbugs/exclude-filter.xml
@@ -67,12 +67,12 @@
   </Match>
   <Match>
     <!-- bad practice to start a thread in constructor; we should be careful using this class -->
-    <Class name="org.apache.accumulo.core.file.blockfile.cache.LruBlockCache" />
+    <Class name="org.apache.accumulo.core.file.blockfile.cache.lru.LruBlockCache" />
     <Bug code="SC" pattern="SC_START_IN_CTOR" />
   </Match>
   <Match>
     <!-- locking is confusing, but probably correct -->
-    <Class name="org.apache.accumulo.core.file.blockfile.cache.LruBlockCache$EvictionThread" />
+    <Class name="org.apache.accumulo.core.file.blockfile.cache.lru.LruBlockCache$EvictionThread" />
     <Or>
       <Bug code="NN" pattern="NN_NAKED_NOTIFY" />
       <Bug code="UW" pattern="UW_UNCOND_WAIT" />

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java
index c95531b..3e6239d 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileScanner.java
@@ -43,8 +43,9 @@ import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Range;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.file.blockfile.cache.BlockCache;
+import org.apache.accumulo.core.file.blockfile.cache.BlockCacheManager;
 import org.apache.accumulo.core.file.blockfile.cache.CacheEntry;
-import org.apache.accumulo.core.file.blockfile.cache.LruBlockCache;
+import org.apache.accumulo.core.file.blockfile.cache.CacheType;
 import org.apache.accumulo.core.file.blockfile.impl.CachableBlockFile;
 import org.apache.accumulo.core.file.rfile.RFile;
 import org.apache.accumulo.core.file.rfile.RFile.Reader;
@@ -67,6 +68,7 @@ class RFileScanner extends ScannerOptions implements Scanner {
   private static final Range EMPTY_RANGE = new Range();
 
   private Range range;
+  private BlockCacheManager blockCacheManager = null;
   private BlockCache dataCache = null;
   private BlockCache indexCache = null;
   private Opts opts;
@@ -109,6 +111,11 @@ class RFileScanner extends ScannerOptions implements Scanner {
     }
 
     @Override
+    public long getMaxHeapSize() {
+      return getMaxSize();
+    }
+
+    @Override
     public long getMaxSize() {
       return Integer.MAX_VALUE;
     }
@@ -135,15 +142,34 @@ class RFileScanner extends ScannerOptions implements Scanner {
     }
 
     this.opts = opts;
-    if (opts.indexCacheSize > 0) {
-      this.indexCache = new LruBlockCache(opts.indexCacheSize, CACHE_BLOCK_SIZE);
-    } else {
+
+    if (opts.indexCacheSize > 0 || opts.dataCacheSize > 0) {
+      ConfigurationCopy cc = new ConfigurationCopy(DefaultConfiguration.getInstance());
+      if (null != opts.tableConfig) {
+        opts.tableConfig.forEach(cc::set);
+      }
+
+      try {
+        blockCacheManager = BlockCacheManager.getClientInstance(cc);
+        if (opts.indexCacheSize > 0) {
+          cc.set(Property.TSERV_INDEXCACHE_SIZE, Long.toString(opts.indexCacheSize));
+        }
+        if (opts.dataCacheSize > 0) {
+          cc.set(Property.TSERV_DATACACHE_SIZE, Long.toString(opts.dataCacheSize));
+        }
+        blockCacheManager.start(cc);
+        this.indexCache = blockCacheManager.getBlockCache(CacheType.INDEX);
+        this.dataCache = blockCacheManager.getBlockCache(CacheType.DATA);
+      } catch (RuntimeException e) {
+        throw e;
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    }
+    if (null == indexCache) {
       this.indexCache = new NoopCache();
     }
-
-    if (opts.dataCacheSize > 0) {
-      this.dataCache = new LruBlockCache(opts.dataCacheSize, CACHE_BLOCK_SIZE);
-    } else {
+    if (null == this.dataCache) {
       this.dataCache = new NoopCache();
     }
   }
@@ -326,14 +352,6 @@ class RFileScanner extends ScannerOptions implements Scanner {
 
   @Override
   public void close() {
-    if (dataCache instanceof LruBlockCache) {
-      ((LruBlockCache) dataCache).shutdown();
-    }
-
-    if (indexCache instanceof LruBlockCache) {
-      ((LruBlockCache) indexCache).shutdown();
-    }
-
     try {
       for (RFileSource source : opts.in.getSources()) {
         source.getInputStream().close();
@@ -341,5 +359,12 @@ class RFileScanner extends ScannerOptions implements Scanner {
     } catch (IOException e) {
       throw new RuntimeException(e);
     }
+    try {
+      if (null != this.blockCacheManager) {
+        this.blockCacheManager.stop();
+      }
+    } catch (Exception e1) {
+      throw new RuntimeException(e1);
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/conf/Property.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/Property.java b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
index 5480867..a506e07 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/Property.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/Property.java
@@ -245,7 +245,9 @@ public enum Property {
   TSERV_PREFIX("tserver.", null, PropertyType.PREFIX, "Properties in this category affect the behavior of the tablet servers"),
   TSERV_CLIENT_TIMEOUT("tserver.client.timeout", "3s", PropertyType.TIMEDURATION, "Time to wait for clients to continue scans before closing a session."),
   TSERV_DEFAULT_BLOCKSIZE("tserver.default.blocksize", "1M", PropertyType.BYTES, "Specifies a default blocksize for the tserver caches"),
-  TSERV_CACHE_POLICY("tserver.cache.policy", "LRU", PropertyType.STRING, "Specifies the eviction policy of the file data caches (LRU or TinyLFU)."),
+  TSERV_CACHE_MANAGER_IMPL("tserver.cache.manager.class", "org.apache.accumulo.core.file.blockfile.cache.lru.LruBlockCacheManager", PropertyType.STRING,
+      "Specifies the class name of the block cache factory implementation. Alternative implementation is "
+          + "org.apache.accumulo.core.file.blockfile.cache.tinylfu.TinyLfuBlockCacheManager"),
   TSERV_DATACACHE_SIZE("tserver.cache.data.size", "10%", PropertyType.MEMORY, "Specifies the size of the cache for file data blocks."),
   TSERV_INDEXCACHE_SIZE("tserver.cache.index.size", "25%", PropertyType.MEMORY, "Specifies the size of the cache for file indices."),
   TSERV_SUMMARYCACHE_SIZE("tserver.cache.summary.size", "10%", PropertyType.MEMORY, "Specifies the size of the cache for summary data on each tablet server."),

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCache.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCache.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCache.java
index 82f8b1e..b27c918 100644
--- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCache.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCache.java
@@ -21,6 +21,7 @@ package org.apache.accumulo.core.file.blockfile.cache;
  * Block cache interface.
  */
 public interface BlockCache {
+
   /**
    * Add block to cache.
    *
@@ -53,6 +54,11 @@ public interface BlockCache {
   CacheEntry getBlock(String blockName);
 
   /**
+   * Get the maximum amount of on heap memory this cache will use.
+   */
+  long getMaxHeapSize();
+
+  /**
    * Get the maximum size of this cache.
    *
    * @return max size in bytes

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCacheConfiguration.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCacheConfiguration.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCacheConfiguration.java
new file mode 100644
index 0000000..efab628
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCacheConfiguration.java
@@ -0,0 +1,91 @@
+/*
+ * 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.accumulo.core.file.blockfile.cache;
+
+import java.util.Map;
+import java.util.Optional;
+
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+
+public class BlockCacheConfiguration {
+
+  public static final String CACHE_PROPERTY_BASE = Property.GENERAL_ARBITRARY_PROP_PREFIX + "cache.block.";
+
+  /** Maximum allowable size of cache (block put if size > max, evict) */
+  private final long maxSize;
+
+  /** Approximate block size */
+  private final long blockSize;
+
+  private final Map<String,String> genProps;
+
+  private final String prefix;
+
+  private final String defaultPrefix;
+
+  public BlockCacheConfiguration(AccumuloConfiguration conf, CacheType type, String implName) {
+    defaultPrefix = getDefaultPrefix(implName);
+    prefix = getPrefix(type, implName);
+    genProps = conf.getAllPropertiesWithPrefix(Property.GENERAL_ARBITRARY_PROP_PREFIX);
+
+    switch (type) {
+      case INDEX:
+        this.maxSize = conf.getAsBytes(Property.TSERV_INDEXCACHE_SIZE);
+        break;
+      case DATA:
+        this.maxSize = conf.getAsBytes(Property.TSERV_DATACACHE_SIZE);
+        break;
+      case SUMMARY:
+        this.maxSize = conf.getAsBytes(Property.TSERV_SUMMARYCACHE_SIZE);
+        break;
+      default:
+        throw new IllegalArgumentException("Unknown block cache type");
+    }
+    this.blockSize = conf.getAsBytes(Property.TSERV_DEFAULT_BLOCKSIZE);
+  }
+
+  public long getMaxSize() {
+    return this.maxSize;
+  }
+
+  public long getBlockSize() {
+    return this.blockSize;
+  }
+
+  protected Optional<String> get(String suffix) {
+    String val = genProps.get(prefix + suffix);
+    if (val == null) {
+      val = genProps.get(defaultPrefix + suffix);
+    }
+    return Optional.ofNullable(val);
+  }
+
+  public static String getDefaultPrefix(String implName) {
+    return CACHE_PROPERTY_BASE + implName + ".default.";
+  }
+
+  public static String getPrefix(CacheType type, String implName) {
+    return CACHE_PROPERTY_BASE + implName + "." + type.name().toLowerCase() + ".";
+  }
+
+  @Override
+  public String toString() {
+    return "maxSize: " + getMaxSize() + ", blockSize: " + getBlockSize();
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCacheManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCacheManager.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCacheManager.java
new file mode 100644
index 0000000..be77ee2
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/BlockCacheManager.java
@@ -0,0 +1,107 @@
+/*
+ * 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.accumulo.core.file.blockfile.cache;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.conf.Property;
+import org.apache.accumulo.start.classloader.vfs.AccumuloVFSClassLoader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class BlockCacheManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(BlockCacheManager.class);
+
+  private final Map<CacheType,BlockCache> caches = new HashMap<>();
+
+  /**
+   * Initialize the caches for each CacheType based on the configuration
+   *
+   * @param conf
+   *          accumulo configuration
+   */
+  public void start(AccumuloConfiguration conf) {
+    for (CacheType type : CacheType.values()) {
+      BlockCache cache = this.createCache(conf, type);
+      this.caches.put(type, cache);
+    }
+  }
+
+  /**
+   * Stop caches and release resources
+   */
+  public void stop() {
+    this.caches.clear();
+  }
+
+  /**
+   * Get the block cache of the given type
+   *
+   * @param type
+   *          block cache type
+   * @return BlockCache or null if not enabled
+   */
+  public BlockCache getBlockCache(CacheType type) {
+    return caches.get(type);
+  }
+
+  /**
+   * Create a block cache using the supplied configuration
+   *
+   * @param conf
+   *          cache configuration
+   * @return configured block cache
+   */
+  protected abstract BlockCache createCache(AccumuloConfiguration conf, CacheType type);
+
+  /**
+   * Get the BlockCacheFactory specified by the property 'tserver.cache.factory.class' using the AccumuloVFSClassLoader
+   *
+   * @param conf
+   *          accumulo configuration
+   * @return block cache manager instance
+   * @throws Exception
+   *           error loading block cache manager implementation class
+   */
+  public static synchronized BlockCacheManager getInstance(AccumuloConfiguration conf) throws Exception {
+    String impl = conf.get(Property.TSERV_CACHE_MANAGER_IMPL);
+    Class<? extends BlockCacheManager> clazz = AccumuloVFSClassLoader.loadClass(impl, BlockCacheManager.class);
+    LOG.info("Created new block cache manager of type: {}", clazz.getSimpleName());
+    return (BlockCacheManager) clazz.newInstance();
+  }
+
+  /**
+   * Get the BlockCacheFactory specified by the property 'tserver.cache.factory.class'
+   *
+   * @param conf
+   *          accumulo configuration
+   * @return block cache manager instance
+   * @throws Exception
+   *           error loading block cache manager implementation class
+   */
+  public static synchronized BlockCacheManager getClientInstance(AccumuloConfiguration conf) throws Exception {
+    String impl = conf.get(Property.TSERV_CACHE_MANAGER_IMPL);
+    Class<? extends BlockCacheManager> clazz = Class.forName(impl).asSubclass(BlockCacheManager.class);
+    LOG.info("Created new block cache factory of type: {}", clazz.getSimpleName());
+    return (BlockCacheManager) clazz.newInstance();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/CacheType.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/CacheType.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/CacheType.java
new file mode 100644
index 0000000..9e26c05
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/CacheType.java
@@ -0,0 +1,24 @@
+/*
+ * 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.accumulo.core.file.blockfile.cache;
+
+public enum CacheType {
+
+  DATA, INDEX, SUMMARY;
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/CachedBlock.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/CachedBlock.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/CachedBlock.java
index c67b4c7..44cea6b 100644
--- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/CachedBlock.java
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/CachedBlock.java
@@ -31,7 +31,7 @@ public class CachedBlock implements HeapSize, Comparable<CachedBlock>, CacheEntr
   public final static long PER_BLOCK_OVERHEAD = ClassSize.align(ClassSize.OBJECT + (3 * ClassSize.REFERENCE) + (2 * SizeConstants.SIZEOF_LONG)
       + ClassSize.STRING + ClassSize.BYTE_BUFFER);
 
-  static enum BlockPriority {
+  public static enum BlockPriority {
     /**
      * Accessed a single time (used for scan-resistance)
      */

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/LruBlockCache.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/LruBlockCache.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/LruBlockCache.java
deleted file mode 100644
index cbdaca5..0000000
--- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/LruBlockCache.java
+++ /dev/null
@@ -1,703 +0,0 @@
-/*
- * 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.accumulo.core.file.blockfile.cache;
-
-import java.lang.ref.WeakReference;
-import java.util.Objects;
-import java.util.PriorityQueue;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.apache.accumulo.core.util.NamingThreadFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A block cache implementation that is memory-aware using {@link HeapSize}, memory-bound using an LRU eviction algorithm, and concurrent: backed by a
- * {@link ConcurrentHashMap} and with a non-blocking eviction thread giving constant-time {@link #cacheBlock} and {@link #getBlock} operations.
- *
- * <p>
- * Contains three levels of block priority to allow for scan-resistance and in-memory families. A block is added with an inMemory flag if necessary, otherwise a
- * block becomes a single access priority. Once a blocked is accessed again, it changes to multiple access. This is used to prevent scans from thrashing the
- * cache, adding a least-frequently-used element to the eviction algorithm.
- *
- * <p>
- * Each priority is given its own chunk of the total cache to ensure fairness during eviction. Each priority will retain close to its maximum size, however, if
- * any priority is not using its entire chunk the others are able to grow beyond their chunk size.
- *
- * <p>
- * Instantiated at a minimum with the total size and average block size. All sizes are in bytes. The block size is not especially important as this cache is
- * fully dynamic in its sizing of blocks. It is only used for pre-allocating data structures and in initial heap estimation of the map.
- *
- * <p>
- * The detailed constructor defines the sizes for the three priorities (they should total to the maximum size defined). It also sets the levels that trigger and
- * control the eviction thread.
- *
- * <p>
- * The acceptable size is the cache size level which triggers the eviction process to start. It evicts enough blocks to get the size below the minimum size
- * specified.
- *
- * <p>
- * Eviction happens in a separate thread and involves a single full-scan of the map. It determines how many bytes must be freed to reach the minimum size, and
- * then while scanning determines the fewest least-recently-used blocks necessary from each of the three priorities (would be 3 times bytes to free). It then
- * uses the priority chunk sizes to evict fairly according to the relative sizes and usage.
- */
-public class LruBlockCache implements BlockCache, HeapSize {
-
-  private static final Logger log = LoggerFactory.getLogger(LruBlockCache.class);
-
-  /** Default Configuration Parameters */
-
-  /** Backing Concurrent Map Configuration */
-  static final float DEFAULT_LOAD_FACTOR = 0.75f;
-  static final int DEFAULT_CONCURRENCY_LEVEL = 16;
-
-  /** Eviction thresholds */
-  static final float DEFAULT_MIN_FACTOR = 0.75f;
-  static final float DEFAULT_ACCEPTABLE_FACTOR = 0.85f;
-
-  /** Priority buckets */
-  static final float DEFAULT_SINGLE_FACTOR = 0.25f;
-  static final float DEFAULT_MULTI_FACTOR = 0.50f;
-  static final float DEFAULT_MEMORY_FACTOR = 0.25f;
-
-  /** Statistics thread */
-  static final int statThreadPeriod = 60;
-
-  /** Concurrent map (the cache) */
-  private final ConcurrentHashMap<String,CachedBlock> map;
-
-  /** Eviction lock (locked when eviction in process) */
-  private final ReentrantLock evictionLock = new ReentrantLock(true);
-
-  /** Volatile boolean to track if we are in an eviction process or not */
-  private volatile boolean evictionInProgress = false;
-
-  /** Eviction thread */
-  private final EvictionThread evictionThread;
-
-  /** Statistics thread schedule pool (for heavy debugging, could remove) */
-  private final ScheduledExecutorService scheduleThreadPool = Executors.newScheduledThreadPool(1, new NamingThreadFactory("LRUBlockCacheStats"));
-
-  /** Current size of cache */
-  private final AtomicLong size;
-
-  /** Current number of cached elements */
-  private final AtomicLong elements;
-
-  /** Cache access count (sequential ID) */
-  private final AtomicLong count;
-
-  /** Cache statistics */
-  private final CacheStats stats;
-
-  /** Maximum allowable size of cache (block put if size > max, evict) */
-  private long maxSize;
-
-  /** Approximate block size */
-  private long blockSize;
-
-  /** Acceptable size of cache (no evictions if size < acceptable) */
-  private float acceptableFactor;
-
-  /** Minimum threshold of cache (when evicting, evict until size < min) */
-  private float minFactor;
-
-  /** Single access bucket size */
-  private float singleFactor;
-
-  /** Multiple access bucket size */
-  private float multiFactor;
-
-  /** In-memory bucket size */
-  private float memoryFactor;
-
-  /** Overhead of the structure itself */
-  private long overhead;
-
-  /**
-   * Default constructor. Specify maximum size and expected average block size (approximation is fine).
-   *
-   * <p>
-   * All other factors will be calculated based on defaults specified in this class.
-   *
-   * @param maxSize
-   *          maximum size of cache, in bytes
-   * @param blockSize
-   *          approximate size of each block, in bytes
-   */
-  public LruBlockCache(long maxSize, long blockSize) {
-    this(maxSize, blockSize, true);
-  }
-
-  /**
-   * Constructor used for testing. Allows disabling of the eviction thread.
-   */
-  public LruBlockCache(long maxSize, long blockSize, boolean evictionThread) {
-    this(maxSize, blockSize, evictionThread, (int) Math.ceil(1.2 * maxSize / blockSize), DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL, DEFAULT_MIN_FACTOR,
-        DEFAULT_ACCEPTABLE_FACTOR, DEFAULT_SINGLE_FACTOR, DEFAULT_MULTI_FACTOR, DEFAULT_MEMORY_FACTOR);
-  }
-
-  /**
-   * Configurable constructor. Use this constructor if not using defaults.
-   *
-   * @param maxSize
-   *          maximum size of this cache, in bytes
-   * @param blockSize
-   *          expected average size of blocks, in bytes
-   * @param evictionThread
-   *          whether to run evictions in a bg thread or not
-   * @param mapInitialSize
-   *          initial size of backing ConcurrentHashMap
-   * @param mapLoadFactor
-   *          initial load factor of backing ConcurrentHashMap
-   * @param mapConcurrencyLevel
-   *          initial concurrency factor for backing CHM
-   * @param minFactor
-   *          percentage of total size that eviction will evict until
-   * @param acceptableFactor
-   *          percentage of total size that triggers eviction
-   * @param singleFactor
-   *          percentage of total size for single-access blocks
-   * @param multiFactor
-   *          percentage of total size for multiple-access blocks
-   * @param memoryFactor
-   *          percentage of total size for in-memory blocks
-   */
-  public LruBlockCache(long maxSize, long blockSize, boolean evictionThread, int mapInitialSize, float mapLoadFactor, int mapConcurrencyLevel, float minFactor,
-      float acceptableFactor, float singleFactor, float multiFactor, float memoryFactor) {
-    if (singleFactor + multiFactor + memoryFactor != 1) {
-      throw new IllegalArgumentException("Single, multi, and memory factors " + " should total 1.0");
-    }
-    if (minFactor >= acceptableFactor) {
-      throw new IllegalArgumentException("minFactor must be smaller than acceptableFactor");
-    }
-    if (minFactor >= 1.0f || acceptableFactor >= 1.0f) {
-      throw new IllegalArgumentException("all factors must be < 1");
-    }
-    this.maxSize = maxSize;
-    this.blockSize = blockSize;
-    map = new ConcurrentHashMap<>(mapInitialSize, mapLoadFactor, mapConcurrencyLevel);
-    this.minFactor = minFactor;
-    this.acceptableFactor = acceptableFactor;
-    this.singleFactor = singleFactor;
-    this.multiFactor = multiFactor;
-    this.memoryFactor = memoryFactor;
-    this.stats = new CacheStats();
-    this.count = new AtomicLong(0);
-    this.elements = new AtomicLong(0);
-    this.overhead = calculateOverhead(maxSize, blockSize, mapConcurrencyLevel);
-    this.size = new AtomicLong(this.overhead);
-
-    if (evictionThread) {
-      this.evictionThread = new EvictionThread(this);
-      this.evictionThread.start();
-      while (!this.evictionThread.running()) {
-        try {
-          Thread.sleep(10);
-        } catch (InterruptedException ex) {
-          throw new RuntimeException(ex);
-        }
-      }
-    } else {
-      this.evictionThread = null;
-    }
-    this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this), statThreadPeriod, statThreadPeriod, TimeUnit.SECONDS);
-  }
-
-  public void setMaxSize(long maxSize) {
-    this.maxSize = maxSize;
-    if (this.size.get() > acceptableSize() && !evictionInProgress) {
-      runEviction();
-    }
-  }
-
-  // BlockCache implementation
-
-  /**
-   * Cache the block with the specified name and buffer.
-   * <p>
-   * It is assumed this will NEVER be called on an already cached block. If that is done, it is assumed that you are reinserting the same exact block due to a
-   * race condition and will update the buffer but not modify the size of the cache.
-   *
-   * @param blockName
-   *          block name
-   * @param buf
-   *          block buffer
-   * @param inMemory
-   *          if block is in-memory
-   */
-  @Override
-  public CacheEntry cacheBlock(String blockName, byte buf[], boolean inMemory) {
-    CachedBlock cb = map.get(blockName);
-    if (cb != null) {
-      stats.duplicateReads();
-      cb.access(count.incrementAndGet());
-    } else {
-      cb = new CachedBlock(blockName, buf, count.incrementAndGet(), inMemory);
-      CachedBlock currCb = map.putIfAbsent(blockName, cb);
-      if (currCb != null) {
-        stats.duplicateReads();
-        cb = currCb;
-        cb.access(count.incrementAndGet());
-      } else {
-        // Actually added block to cache
-        long newSize = size.addAndGet(cb.heapSize());
-        elements.incrementAndGet();
-        if (newSize > acceptableSize() && !evictionInProgress) {
-          runEviction();
-        }
-      }
-    }
-
-    return cb;
-  }
-
-  /**
-   * Cache the block with the specified name and buffer.
-   * <p>
-   * It is assumed this will NEVER be called on an already cached block. If that is done, it is assumed that you are reinserting the same exact block due to a
-   * race condition and will update the buffer but not modify the size of the cache.
-   *
-   * @param blockName
-   *          block name
-   * @param buf
-   *          block buffer
-   */
-  @Override
-  public CacheEntry cacheBlock(String blockName, byte buf[]) {
-    return cacheBlock(blockName, buf, false);
-  }
-
-  /**
-   * Get the buffer of the block with the specified name.
-   *
-   * @param blockName
-   *          block name
-   * @return buffer of specified block name, or null if not in cache
-   */
-  @Override
-  public CachedBlock getBlock(String blockName) {
-    CachedBlock cb = map.get(blockName);
-    if (cb == null) {
-      stats.miss();
-      return null;
-    }
-    stats.hit();
-    cb.access(count.incrementAndGet());
-    return cb;
-  }
-
-  protected long evictBlock(CachedBlock block) {
-    map.remove(block.getName());
-    size.addAndGet(-1 * block.heapSize());
-    elements.decrementAndGet();
-    stats.evicted();
-    return block.heapSize();
-  }
-
-  /**
-   * Multi-threaded call to run the eviction process.
-   */
-  private void runEviction() {
-    if (evictionThread == null) {
-      evict();
-    } else {
-      evictionThread.evict();
-    }
-  }
-
-  /**
-   * Eviction method.
-   */
-  void evict() {
-
-    // Ensure only one eviction at a time
-    if (!evictionLock.tryLock())
-      return;
-
-    try {
-      evictionInProgress = true;
-
-      long bytesToFree = size.get() - minSize();
-
-      log.trace("Block cache LRU eviction started.  Attempting to free {} bytes", bytesToFree);
-
-      if (bytesToFree <= 0)
-        return;
-
-      // Instantiate priority buckets
-      BlockBucket bucketSingle = new BlockBucket(bytesToFree, blockSize, singleSize());
-      BlockBucket bucketMulti = new BlockBucket(bytesToFree, blockSize, multiSize());
-      BlockBucket bucketMemory = new BlockBucket(bytesToFree, blockSize, memorySize());
-
-      // Scan entire map putting into appropriate buckets
-      for (CachedBlock cachedBlock : map.values()) {
-        switch (cachedBlock.getPriority()) {
-          case SINGLE: {
-            bucketSingle.add(cachedBlock);
-            break;
-          }
-          case MULTI: {
-            bucketMulti.add(cachedBlock);
-            break;
-          }
-          case MEMORY: {
-            bucketMemory.add(cachedBlock);
-            break;
-          }
-        }
-      }
-
-      PriorityQueue<BlockBucket> bucketQueue = new PriorityQueue<>(3);
-
-      bucketQueue.add(bucketSingle);
-      bucketQueue.add(bucketMulti);
-      bucketQueue.add(bucketMemory);
-
-      int remainingBuckets = 3;
-      long bytesFreed = 0;
-
-      BlockBucket bucket;
-      while ((bucket = bucketQueue.poll()) != null) {
-        long overflow = bucket.overflow();
-        if (overflow > 0) {
-          long bucketBytesToFree = Math.min(overflow, (long) Math.ceil((bytesToFree - bytesFreed) / (double) remainingBuckets));
-          bytesFreed += bucket.free(bucketBytesToFree);
-        }
-        remainingBuckets--;
-      }
-
-      float singleMB = ((float) bucketSingle.totalSize()) / ((float) (1024 * 1024));
-      float multiMB = ((float) bucketMulti.totalSize()) / ((float) (1024 * 1024));
-      float memoryMB = ((float) bucketMemory.totalSize()) / ((float) (1024 * 1024));
-
-      log.trace("Block cache LRU eviction completed. Freed {} bytes. Priority Sizes: Single={}MB ({}), Multi={}MB ({}), Memory={}MB ({})", bytesFreed,
-          singleMB, bucketSingle.totalSize(), multiMB, bucketMulti.totalSize(), memoryMB, bucketMemory.totalSize());
-
-    } finally {
-      stats.evict();
-      evictionInProgress = false;
-      evictionLock.unlock();
-    }
-  }
-
-  /**
-   * Used to group blocks into priority buckets. There will be a BlockBucket for each priority (single, multi, memory). Once bucketed, the eviction algorithm
-   * takes the appropriate number of elements out of each according to configuration parameters and their relatives sizes.
-   */
-  private class BlockBucket implements Comparable<BlockBucket> {
-
-    private CachedBlockQueue queue;
-    private long totalSize = 0;
-    private long bucketSize;
-
-    public BlockBucket(long bytesToFree, long blockSize, long bucketSize) {
-      this.bucketSize = bucketSize;
-      queue = new CachedBlockQueue(bytesToFree, blockSize);
-      totalSize = 0;
-    }
-
-    public void add(CachedBlock block) {
-      totalSize += block.heapSize();
-      queue.add(block);
-    }
-
-    public long free(long toFree) {
-      CachedBlock[] blocks = queue.get();
-      long freedBytes = 0;
-      for (int i = 0; i < blocks.length; i++) {
-        freedBytes += evictBlock(blocks[i]);
-        if (freedBytes >= toFree) {
-          return freedBytes;
-        }
-      }
-      return freedBytes;
-    }
-
-    public long overflow() {
-      return totalSize - bucketSize;
-    }
-
-    public long totalSize() {
-      return totalSize;
-    }
-
-    @Override
-    public int compareTo(BlockBucket that) {
-      if (this.overflow() == that.overflow())
-        return 0;
-      return this.overflow() > that.overflow() ? 1 : -1;
-    }
-
-    @Override
-    public int hashCode() {
-      return Objects.hashCode(overflow());
-    }
-
-    @Override
-    public boolean equals(Object that) {
-      if (that instanceof BlockBucket)
-        return compareTo((BlockBucket) that) == 0;
-      return false;
-    }
-  }
-
-  @Override
-  public long getMaxSize() {
-    return this.maxSize;
-  }
-
-  /**
-   * Get the current size of this cache.
-   *
-   * @return current size in bytes
-   */
-  public long getCurrentSize() {
-    return this.size.get();
-  }
-
-  /**
-   * Get the current size of this cache.
-   *
-   * @return current size in bytes
-   */
-  public long getFreeSize() {
-    return getMaxSize() - getCurrentSize();
-  }
-
-  /**
-   * Get the size of this cache (number of cached blocks)
-   *
-   * @return number of cached blocks
-   */
-  public long size() {
-    return this.elements.get();
-  }
-
-  /**
-   * Get the number of eviction runs that have occurred
-   */
-  public long getEvictionCount() {
-    return this.stats.getEvictionCount();
-  }
-
-  /**
-   * Get the number of blocks that have been evicted during the lifetime of this cache.
-   */
-  public long getEvictedCount() {
-    return this.stats.getEvictedCount();
-  }
-
-  /**
-   * Eviction thread. Sits in waiting state until an eviction is triggered when the cache size grows above the acceptable level.
-   *
-   * <p>
-   * Thread is triggered into action by {@link LruBlockCache#runEviction()}
-   */
-  private static class EvictionThread extends Thread {
-    private WeakReference<LruBlockCache> cache;
-    private boolean running = false;
-
-    public EvictionThread(LruBlockCache cache) {
-      super("LruBlockCache.EvictionThread");
-      setDaemon(true);
-      this.cache = new WeakReference<>(cache);
-    }
-
-    public synchronized boolean running() {
-      return running;
-    }
-
-    @Override
-    public void run() {
-      while (true) {
-        synchronized (this) {
-          running = true;
-          try {
-            this.wait();
-          } catch (InterruptedException e) {}
-        }
-        LruBlockCache cache = this.cache.get();
-        if (cache == null)
-          break;
-        cache.evict();
-      }
-    }
-
-    public void evict() {
-      synchronized (this) {
-        this.notify();
-      }
-    }
-  }
-
-  /*
-   * Statistics thread. Periodically prints the cache statistics to the log.
-   */
-  private static class StatisticsThread extends Thread {
-    LruBlockCache lru;
-
-    public StatisticsThread(LruBlockCache lru) {
-      super("LruBlockCache.StatisticsThread");
-      setDaemon(true);
-      this.lru = lru;
-    }
-
-    @Override
-    public void run() {
-      lru.logStats();
-    }
-  }
-
-  public void logStats() {
-    // Log size
-    long totalSize = heapSize();
-    long freeSize = maxSize - totalSize;
-    float sizeMB = ((float) totalSize) / ((float) (1024 * 1024));
-    float freeMB = ((float) freeSize) / ((float) (1024 * 1024));
-    float maxMB = ((float) maxSize) / ((float) (1024 * 1024));
-    log.debug("Cache Stats: Sizes: Total={}MB ({}), Free={}MB ({}), Max={}MB ({}), Counts: Blocks={}, Access={}, Hit={}, Miss={}, Evictions={}, Evicted={},"
-        + "Ratios: Hit Ratio={}%, Miss Ratio={}%, Evicted/Run={}, Duplicate Reads={}", sizeMB, totalSize, freeMB, freeSize, maxMB, maxSize, size(),
-        stats.requestCount(), stats.hitCount(), stats.getMissCount(), stats.getEvictionCount(), stats.getEvictedCount(), stats.getHitRatio() * 100,
-        stats.getMissRatio() * 100, stats.evictedPerEviction(), stats.getDuplicateReads());
-  }
-
-  /**
-   * Get counter statistics for this cache.
-   *
-   * <p>
-   * Includes: total accesses, hits, misses, evicted blocks, and runs of the eviction processes.
-   */
-  public CacheStats getStats() {
-    return this.stats;
-  }
-
-  public static class CacheStats implements BlockCache.Stats {
-    private final AtomicLong accessCount = new AtomicLong(0);
-    private final AtomicLong hitCount = new AtomicLong(0);
-    private final AtomicLong missCount = new AtomicLong(0);
-    private final AtomicLong evictionCount = new AtomicLong(0);
-    private final AtomicLong evictedCount = new AtomicLong(0);
-    private final AtomicLong duplicateReads = new AtomicLong(0);
-
-    public void miss() {
-      missCount.incrementAndGet();
-      accessCount.incrementAndGet();
-    }
-
-    public void hit() {
-      hitCount.incrementAndGet();
-      accessCount.incrementAndGet();
-    }
-
-    public void evict() {
-      evictionCount.incrementAndGet();
-    }
-
-    public void duplicateReads() {
-      duplicateReads.incrementAndGet();
-    }
-
-    public void evicted() {
-      evictedCount.incrementAndGet();
-    }
-
-    @Override
-    public long requestCount() {
-      return accessCount.get();
-    }
-
-    public long getMissCount() {
-      return missCount.get();
-    }
-
-    @Override
-    public long hitCount() {
-      return hitCount.get();
-    }
-
-    public long getEvictionCount() {
-      return evictionCount.get();
-    }
-
-    public long getDuplicateReads() {
-      return duplicateReads.get();
-    }
-
-    public long getEvictedCount() {
-      return evictedCount.get();
-    }
-
-    public double getHitRatio() {
-      return ((float) hitCount() / (float) requestCount());
-    }
-
-    public double getMissRatio() {
-      return ((float) getMissCount() / (float) requestCount());
-    }
-
-    public double evictedPerEviction() {
-      return (float) getEvictedCount() / (float) getEvictionCount();
-    }
-  }
-
-  public final static long CACHE_FIXED_OVERHEAD = ClassSize.align((3 * SizeConstants.SIZEOF_LONG) + (8 * ClassSize.REFERENCE)
-      + (5 * SizeConstants.SIZEOF_FLOAT) + SizeConstants.SIZEOF_BOOLEAN + ClassSize.OBJECT);
-
-  // HeapSize implementation
-  @Override
-  public long heapSize() {
-    return getCurrentSize();
-  }
-
-  public static long calculateOverhead(long maxSize, long blockSize, int concurrency) {
-    return CACHE_FIXED_OVERHEAD + ClassSize.CONCURRENT_HASHMAP + ((int) Math.ceil(maxSize * 1.2 / blockSize) * ClassSize.CONCURRENT_HASHMAP_ENTRY)
-        + (concurrency * ClassSize.CONCURRENT_HASHMAP_SEGMENT);
-  }
-
-  // Simple calculators of sizes given factors and maxSize
-
-  private long acceptableSize() {
-    return (long) Math.floor(this.maxSize * this.acceptableFactor);
-  }
-
-  private long minSize() {
-    return (long) Math.floor(this.maxSize * this.minFactor);
-  }
-
-  private long singleSize() {
-    return (long) Math.floor(this.maxSize * this.singleFactor * this.minFactor);
-  }
-
-  private long multiSize() {
-    return (long) Math.floor(this.maxSize * this.multiFactor * this.minFactor);
-  }
-
-  private long memorySize() {
-    return (long) Math.floor(this.maxSize * this.memoryFactor * this.minFactor);
-  }
-
-  public void shutdown() {
-    this.scheduleThreadPool.shutdown();
-  }
-}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/TinyLfuBlockCache.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/TinyLfuBlockCache.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/TinyLfuBlockCache.java
deleted file mode 100644
index bab52af..0000000
--- a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/TinyLfuBlockCache.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- * 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.accumulo.core.file.blockfile.cache;
-
-import static java.util.Objects.requireNonNull;
-
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
-import com.github.benmanes.caffeine.cache.Policy;
-import com.github.benmanes.caffeine.cache.stats.CacheStats;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-
-/**
- * A block cache that is memory bounded using the W-TinyLFU eviction algorithm. This implementation delegates to a Caffeine cache to provide concurrent O(1)
- * read and write operations.
- * <ul>
- * <li>W-TinyLFU: http://arxiv.org/pdf/1512.00727.pdf</li>
- * <li>Caffeine: https://github.com/ben-manes/caffeine</li>
- * <li>Cache design: http://highscalability.com/blog/2016/1/25/design-of-a-modern-cache.html</li>
- * </ul>
- */
-public final class TinyLfuBlockCache implements BlockCache {
-  private static final Logger log = LoggerFactory.getLogger(TinyLfuBlockCache.class);
-  private static final int STATS_PERIOD_SEC = 60;
-
-  private final Cache<String,Block> cache;
-  private final Policy.Eviction<String,Block> policy;
-  private final ScheduledExecutorService statsExecutor;
-
-  public TinyLfuBlockCache(long maxSize, long blockSize) {
-    cache = Caffeine.newBuilder().initialCapacity((int) Math.ceil(1.2 * maxSize / blockSize)).weigher((String blockName, Block block) -> {
-      int keyWeight = ClassSize.align(blockName.length()) + ClassSize.STRING;
-      return keyWeight + block.weight();
-    }).maximumWeight(maxSize).recordStats().build();
-    policy = cache.policy().eviction().get();
-
-    statsExecutor = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("TinyLfuBlockCacheStatsExecutor").setDaemon(true)
-        .build());
-    statsExecutor.scheduleAtFixedRate(this::logStats, STATS_PERIOD_SEC, STATS_PERIOD_SEC, TimeUnit.SECONDS);
-  }
-
-  @Override
-  public long getMaxSize() {
-    return policy.getMaximum();
-  }
-
-  @Override
-  public CacheEntry getBlock(String blockName) {
-    return cache.getIfPresent(blockName);
-  }
-
-  @Override
-  public CacheEntry cacheBlock(String blockName, byte[] buffer) {
-    return cache.asMap().compute(blockName, (key, block) -> {
-      if (block == null) {
-        return new Block(buffer);
-      }
-      block.buffer = buffer;
-      return block;
-    });
-  }
-
-  @Override
-  public CacheEntry cacheBlock(String blockName, byte[] buffer, /* ignored */boolean inMemory) {
-    return cacheBlock(blockName, buffer);
-  }
-
-  @Override
-  public BlockCache.Stats getStats() {
-    CacheStats stats = cache.stats();
-    return new BlockCache.Stats() {
-      @Override
-      public long hitCount() {
-        return stats.hitCount();
-      }
-
-      @Override
-      public long requestCount() {
-        return stats.requestCount();
-      }
-    };
-  }
-
-  private void logStats() {
-    double maxMB = ((double) policy.getMaximum()) / ((double) (1024 * 1024));
-    double sizeMB = ((double) policy.weightedSize().getAsLong()) / ((double) (1024 * 1024));
-    double freeMB = maxMB - sizeMB;
-    log.debug("Cache Size={}MB, Free={}MB, Max={}MB, Blocks={}", sizeMB, freeMB, maxMB, cache.estimatedSize());
-    log.debug(cache.stats().toString());
-  }
-
-  private static final class Block implements CacheEntry {
-    private volatile byte[] buffer;
-    private volatile Object index;
-
-    Block(byte[] buffer) {
-      this.buffer = requireNonNull(buffer);
-    }
-
-    @Override
-    public byte[] getBuffer() {
-      return buffer;
-    }
-
-    @Override
-    public Object getIndex() {
-      return index;
-    }
-
-    @Override
-    public void setIndex(Object index) {
-      this.index = index;
-    }
-
-    int weight() {
-      return ClassSize.align(buffer.length) + SizeConstants.SIZEOF_LONG + ClassSize.REFERENCE + ClassSize.OBJECT + ClassSize.ARRAY;
-    }
-  }
-}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCache.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCache.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCache.java
new file mode 100644
index 0000000..fa8d824
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCache.java
@@ -0,0 +1,625 @@
+/*
+ * 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.accumulo.core.file.blockfile.cache.lru;
+
+import java.lang.ref.WeakReference;
+import java.util.Objects;
+import java.util.PriorityQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.accumulo.core.file.blockfile.cache.BlockCache;
+import org.apache.accumulo.core.file.blockfile.cache.CacheEntry;
+import org.apache.accumulo.core.file.blockfile.cache.CachedBlock;
+import org.apache.accumulo.core.file.blockfile.cache.CachedBlockQueue;
+import org.apache.accumulo.core.file.blockfile.cache.ClassSize;
+import org.apache.accumulo.core.file.blockfile.cache.HeapSize;
+import org.apache.accumulo.core.file.blockfile.cache.SizeConstants;
+import org.apache.accumulo.core.util.NamingThreadFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A block cache implementation that is memory-aware using {@link HeapSize}, memory-bound using an LRU eviction algorithm, and concurrent: backed by a
+ * {@link ConcurrentHashMap} and with a non-blocking eviction thread giving constant-time {@link #cacheBlock} and {@link #getBlock} operations.
+ *
+ * <p>
+ * Contains three levels of block priority to allow for scan-resistance and in-memory families. A block is added with an inMemory flag if necessary, otherwise a
+ * block becomes a single access priority. Once a blocked is accessed again, it changes to multiple access. This is used to prevent scans from thrashing the
+ * cache, adding a least-frequently-used element to the eviction algorithm.
+ *
+ * <p>
+ * Each priority is given its own chunk of the total cache to ensure fairness during eviction. Each priority will retain close to its maximum size, however, if
+ * any priority is not using its entire chunk the others are able to grow beyond their chunk size.
+ *
+ * <p>
+ * Instantiated at a minimum with the total size and average block size. All sizes are in bytes. The block size is not especially important as this cache is
+ * fully dynamic in its sizing of blocks. It is only used for pre-allocating data structures and in initial heap estimation of the map.
+ *
+ * <p>
+ * The detailed constructor defines the sizes for the three priorities (they should total to the maximum size defined). It also sets the levels that trigger and
+ * control the eviction thread.
+ *
+ * <p>
+ * The acceptable size is the cache size level which triggers the eviction process to start. It evicts enough blocks to get the size below the minimum size
+ * specified.
+ *
+ * <p>
+ * Eviction happens in a separate thread and involves a single full-scan of the map. It determines how many bytes must be freed to reach the minimum size, and
+ * then while scanning determines the fewest least-recently-used blocks necessary from each of the three priorities (would be 3 times bytes to free). It then
+ * uses the priority chunk sizes to evict fairly according to the relative sizes and usage.
+ */
+public class LruBlockCache implements BlockCache, HeapSize {
+
+  private static final Logger log = LoggerFactory.getLogger(LruBlockCache.class);
+
+  /** Statistics thread */
+  static final int statThreadPeriod = 60;
+
+  /** Concurrent map (the cache) */
+  private final ConcurrentHashMap<String,CachedBlock> map;
+
+  /** Eviction lock (locked when eviction in process) */
+  private final ReentrantLock evictionLock = new ReentrantLock(true);
+
+  /** Volatile boolean to track if we are in an eviction process or not */
+  private volatile boolean evictionInProgress = false;
+
+  /** Eviction thread */
+  private final EvictionThread evictionThread;
+
+  /** Statistics thread schedule pool (for heavy debugging, could remove) */
+  private final ScheduledExecutorService scheduleThreadPool = Executors.newScheduledThreadPool(1, new NamingThreadFactory("LRUBlockCacheStats"));
+
+  /** Current size of cache */
+  private final AtomicLong size;
+
+  /** Current number of cached elements */
+  private final AtomicLong elements;
+
+  /** Cache access count (sequential ID) */
+  private final AtomicLong count;
+
+  /** Cache statistics */
+  private final CacheStats stats;
+
+  /** Overhead of the structure itself */
+  private final long overhead;
+
+  private final LruBlockCacheConfiguration conf;
+
+  /**
+   * Default constructor. Specify maximum size and expected average block size (approximation is fine).
+   *
+   * <p>
+   * All other factors will be calculated based on defaults specified in this class.
+   *
+   * @param conf
+   *          block cache configuration
+   */
+  public LruBlockCache(final LruBlockCacheConfiguration conf) {
+    this.conf = conf;
+
+    int mapInitialSize = (int) Math.ceil(1.2 * conf.getMaxSize() / conf.getBlockSize());
+
+    map = new ConcurrentHashMap<>(mapInitialSize, conf.getMapLoadFactor(), conf.getMapConcurrencyLevel());
+    this.stats = new CacheStats();
+    this.count = new AtomicLong(0);
+    this.elements = new AtomicLong(0);
+    this.overhead = calculateOverhead(conf.getMaxSize(), conf.getBlockSize(), conf.getMapConcurrencyLevel());
+    this.size = new AtomicLong(this.overhead);
+
+    if (conf.isUseEvictionThread()) {
+      this.evictionThread = new EvictionThread(this);
+      this.evictionThread.start();
+      while (!this.evictionThread.running()) {
+        try {
+          Thread.sleep(10);
+        } catch (InterruptedException ex) {
+          throw new RuntimeException(ex);
+        }
+      }
+    } else {
+      this.evictionThread = null;
+    }
+    this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this), statThreadPeriod, statThreadPeriod, TimeUnit.SECONDS);
+  }
+
+  public long getOverhead() {
+    return overhead;
+  }
+
+  // BlockCache implementation
+
+  /**
+   * Cache the block with the specified name and buffer.
+   * <p>
+   * It is assumed this will NEVER be called on an already cached block. If that is done, it is assumed that you are reinserting the same exact block due to a
+   * race condition and will update the buffer but not modify the size of the cache.
+   *
+   * @param blockName
+   *          block name
+   * @param buf
+   *          block buffer
+   * @param inMemory
+   *          if block is in-memory
+   */
+  @Override
+  public CacheEntry cacheBlock(String blockName, byte buf[], boolean inMemory) {
+    CachedBlock cb = map.get(blockName);
+    if (cb != null) {
+      stats.duplicateReads();
+      cb.access(count.incrementAndGet());
+    } else {
+      cb = new CachedBlock(blockName, buf, count.incrementAndGet(), inMemory);
+      CachedBlock currCb = map.putIfAbsent(blockName, cb);
+      if (currCb != null) {
+        stats.duplicateReads();
+        cb = currCb;
+        cb.access(count.incrementAndGet());
+      } else {
+        // Actually added block to cache
+        long newSize = size.addAndGet(cb.heapSize());
+        elements.incrementAndGet();
+        if (newSize > acceptableSize() && !evictionInProgress) {
+          runEviction();
+        }
+      }
+    }
+
+    return cb;
+  }
+
+  /**
+   * Cache the block with the specified name and buffer.
+   * <p>
+   * It is assumed this will NEVER be called on an already cached block. If that is done, it is assumed that you are reinserting the same exact block due to a
+   * race condition and will update the buffer but not modify the size of the cache.
+   *
+   * @param blockName
+   *          block name
+   * @param buf
+   *          block buffer
+   */
+  @Override
+  public CacheEntry cacheBlock(String blockName, byte buf[]) {
+    return cacheBlock(blockName, buf, false);
+  }
+
+  /**
+   * Get the buffer of the block with the specified name.
+   *
+   * @param blockName
+   *          block name
+   * @return buffer of specified block name, or null if not in cache
+   */
+  @Override
+  public CachedBlock getBlock(String blockName) {
+    CachedBlock cb = map.get(blockName);
+    if (cb == null) {
+      stats.miss();
+      return null;
+    }
+    stats.hit();
+    cb.access(count.incrementAndGet());
+    return cb;
+  }
+
+  protected long evictBlock(CachedBlock block) {
+    map.remove(block.getName());
+    size.addAndGet(-1 * block.heapSize());
+    elements.decrementAndGet();
+    stats.evicted();
+    return block.heapSize();
+  }
+
+  /**
+   * Multi-threaded call to run the eviction process.
+   */
+  private void runEviction() {
+    if (evictionThread == null) {
+      evict();
+    } else {
+      evictionThread.evict();
+    }
+  }
+
+  /**
+   * Eviction method.
+   */
+  void evict() {
+
+    // Ensure only one eviction at a time
+    if (!evictionLock.tryLock())
+      return;
+
+    try {
+      evictionInProgress = true;
+
+      long bytesToFree = size.get() - minSize();
+
+      log.trace("Block cache LRU eviction started.  Attempting to free {} bytes", bytesToFree);
+
+      if (bytesToFree <= 0)
+        return;
+
+      // Instantiate priority buckets
+      BlockBucket bucketSingle = new BlockBucket(bytesToFree, conf.getBlockSize(), singleSize());
+      BlockBucket bucketMulti = new BlockBucket(bytesToFree, conf.getBlockSize(), multiSize());
+      BlockBucket bucketMemory = new BlockBucket(bytesToFree, conf.getBlockSize(), memorySize());
+
+      // Scan entire map putting into appropriate buckets
+      for (CachedBlock cachedBlock : map.values()) {
+        switch (cachedBlock.getPriority()) {
+          case SINGLE: {
+            bucketSingle.add(cachedBlock);
+            break;
+          }
+          case MULTI: {
+            bucketMulti.add(cachedBlock);
+            break;
+          }
+          case MEMORY: {
+            bucketMemory.add(cachedBlock);
+            break;
+          }
+        }
+      }
+
+      PriorityQueue<BlockBucket> bucketQueue = new PriorityQueue<>(3);
+
+      bucketQueue.add(bucketSingle);
+      bucketQueue.add(bucketMulti);
+      bucketQueue.add(bucketMemory);
+
+      int remainingBuckets = 3;
+      long bytesFreed = 0;
+
+      BlockBucket bucket;
+      while ((bucket = bucketQueue.poll()) != null) {
+        long overflow = bucket.overflow();
+        if (overflow > 0) {
+          long bucketBytesToFree = Math.min(overflow, (long) Math.ceil((bytesToFree - bytesFreed) / (double) remainingBuckets));
+          bytesFreed += bucket.free(bucketBytesToFree);
+        }
+        remainingBuckets--;
+      }
+
+      float singleMB = ((float) bucketSingle.totalSize()) / ((float) (1024 * 1024));
+      float multiMB = ((float) bucketMulti.totalSize()) / ((float) (1024 * 1024));
+      float memoryMB = ((float) bucketMemory.totalSize()) / ((float) (1024 * 1024));
+
+      log.trace("Block cache LRU eviction completed. Freed {} bytes. Priority Sizes: Single={}MB ({}), Multi={}MB ({}), Memory={}MB ({})", bytesFreed,
+          singleMB, bucketSingle.totalSize(), multiMB, bucketMulti.totalSize(), memoryMB, bucketMemory.totalSize());
+
+    } finally {
+      stats.evict();
+      evictionInProgress = false;
+      evictionLock.unlock();
+    }
+  }
+
+  /**
+   * Used to group blocks into priority buckets. There will be a BlockBucket for each priority (single, multi, memory). Once bucketed, the eviction algorithm
+   * takes the appropriate number of elements out of each according to configuration parameters and their relatives sizes.
+   */
+  private class BlockBucket implements Comparable<BlockBucket> {
+
+    private CachedBlockQueue queue;
+    private long totalSize = 0;
+    private long bucketSize;
+
+    public BlockBucket(long bytesToFree, long blockSize, long bucketSize) {
+      this.bucketSize = bucketSize;
+      queue = new CachedBlockQueue(bytesToFree, blockSize);
+      totalSize = 0;
+    }
+
+    public void add(CachedBlock block) {
+      totalSize += block.heapSize();
+      queue.add(block);
+    }
+
+    public long free(long toFree) {
+      CachedBlock[] blocks = queue.get();
+      long freedBytes = 0;
+      for (int i = 0; i < blocks.length; i++) {
+        freedBytes += evictBlock(blocks[i]);
+        if (freedBytes >= toFree) {
+          return freedBytes;
+        }
+      }
+      return freedBytes;
+    }
+
+    public long overflow() {
+      return totalSize - bucketSize;
+    }
+
+    public long totalSize() {
+      return totalSize;
+    }
+
+    @Override
+    public int compareTo(BlockBucket that) {
+      if (this.overflow() == that.overflow())
+        return 0;
+      return this.overflow() > that.overflow() ? 1 : -1;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hashCode(overflow());
+    }
+
+    @Override
+    public boolean equals(Object that) {
+      if (that instanceof BlockBucket)
+        return compareTo((BlockBucket) that) == 0;
+      return false;
+    }
+  }
+
+  @Override
+  public long getMaxHeapSize() {
+    return getMaxSize();
+  }
+
+  @Override
+  public long getMaxSize() {
+    return this.conf.getMaxSize();
+  }
+
+  /**
+   * Get the current size of this cache.
+   *
+   * @return current size in bytes
+   */
+  public long getCurrentSize() {
+    return this.size.get();
+  }
+
+  /**
+   * Get the current size of this cache.
+   *
+   * @return current size in bytes
+   */
+  public long getFreeSize() {
+    return getMaxSize() - getCurrentSize();
+  }
+
+  /**
+   * Get the size of this cache (number of cached blocks)
+   *
+   * @return number of cached blocks
+   */
+  public long size() {
+    return this.elements.get();
+  }
+
+  /**
+   * Get the number of eviction runs that have occurred
+   */
+  public long getEvictionCount() {
+    return this.stats.getEvictionCount();
+  }
+
+  /**
+   * Get the number of blocks that have been evicted during the lifetime of this cache.
+   */
+  public long getEvictedCount() {
+    return this.stats.getEvictedCount();
+  }
+
+  /**
+   * Eviction thread. Sits in waiting state until an eviction is triggered when the cache size grows above the acceptable level.
+   *
+   * <p>
+   * Thread is triggered into action by {@link LruBlockCache#runEviction()}
+   */
+  private static class EvictionThread extends Thread {
+    private WeakReference<LruBlockCache> cache;
+    private boolean running = false;
+
+    public EvictionThread(LruBlockCache cache) {
+      super("LruBlockCache.EvictionThread");
+      setDaemon(true);
+      this.cache = new WeakReference<>(cache);
+    }
+
+    public synchronized boolean running() {
+      return running;
+    }
+
+    @Override
+    public void run() {
+      while (true) {
+        synchronized (this) {
+          running = true;
+          try {
+            this.wait();
+          } catch (InterruptedException e) {}
+        }
+        LruBlockCache cache = this.cache.get();
+        if (cache == null)
+          break;
+        cache.evict();
+      }
+    }
+
+    public void evict() {
+      synchronized (this) {
+        this.notify();
+      }
+    }
+  }
+
+  /*
+   * Statistics thread. Periodically prints the cache statistics to the log.
+   */
+  private static class StatisticsThread extends Thread {
+    LruBlockCache lru;
+
+    public StatisticsThread(LruBlockCache lru) {
+      super("LruBlockCache.StatisticsThread");
+      setDaemon(true);
+      this.lru = lru;
+    }
+
+    @Override
+    public void run() {
+      lru.logStats();
+    }
+  }
+
+  public void logStats() {
+    // Log size
+    long totalSize = heapSize();
+    long freeSize = this.conf.getMaxSize() - totalSize;
+    float sizeMB = ((float) totalSize) / ((float) (1024 * 1024));
+    float freeMB = ((float) freeSize) / ((float) (1024 * 1024));
+    float maxMB = ((float) this.conf.getMaxSize()) / ((float) (1024 * 1024));
+    log.debug("Cache Stats: Sizes: Total={}MB ({}), Free={}MB ({}), Max={}MB ({}), Counts: Blocks={}, Access={}, Hit={}, Miss={}, Evictions={}, Evicted={},"
+        + "Ratios: Hit Ratio={}%, Miss Ratio={}%, Evicted/Run={}, Duplicate Reads={}", sizeMB, totalSize, freeMB, freeSize, maxMB, this.conf.getMaxSize(),
+        size(), stats.requestCount(), stats.hitCount(), stats.getMissCount(), stats.getEvictionCount(), stats.getEvictedCount(), stats.getHitRatio() * 100,
+        stats.getMissRatio() * 100, stats.evictedPerEviction(), stats.getDuplicateReads());
+  }
+
+  /**
+   * Get counter statistics for this cache.
+   *
+   * <p>
+   * Includes: total accesses, hits, misses, evicted blocks, and runs of the eviction processes.
+   */
+  public CacheStats getStats() {
+    return this.stats;
+  }
+
+  public static class CacheStats implements BlockCache.Stats {
+    private final AtomicLong accessCount = new AtomicLong(0);
+    private final AtomicLong hitCount = new AtomicLong(0);
+    private final AtomicLong missCount = new AtomicLong(0);
+    private final AtomicLong evictionCount = new AtomicLong(0);
+    private final AtomicLong evictedCount = new AtomicLong(0);
+    private final AtomicLong duplicateReads = new AtomicLong(0);
+
+    public void miss() {
+      missCount.incrementAndGet();
+      accessCount.incrementAndGet();
+    }
+
+    public void hit() {
+      hitCount.incrementAndGet();
+      accessCount.incrementAndGet();
+    }
+
+    public void evict() {
+      evictionCount.incrementAndGet();
+    }
+
+    public void duplicateReads() {
+      duplicateReads.incrementAndGet();
+    }
+
+    public void evicted() {
+      evictedCount.incrementAndGet();
+    }
+
+    @Override
+    public long requestCount() {
+      return accessCount.get();
+    }
+
+    public long getMissCount() {
+      return missCount.get();
+    }
+
+    @Override
+    public long hitCount() {
+      return hitCount.get();
+    }
+
+    public long getEvictionCount() {
+      return evictionCount.get();
+    }
+
+    public long getDuplicateReads() {
+      return duplicateReads.get();
+    }
+
+    public long getEvictedCount() {
+      return evictedCount.get();
+    }
+
+    public double getHitRatio() {
+      return ((float) hitCount() / (float) requestCount());
+    }
+
+    public double getMissRatio() {
+      return ((float) getMissCount() / (float) requestCount());
+    }
+
+    public double evictedPerEviction() {
+      return (float) getEvictedCount() / (float) getEvictionCount();
+    }
+  }
+
+  public final static long CACHE_FIXED_OVERHEAD = ClassSize.align((3 * SizeConstants.SIZEOF_LONG) + (8 * ClassSize.REFERENCE)
+      + (5 * SizeConstants.SIZEOF_FLOAT) + SizeConstants.SIZEOF_BOOLEAN + ClassSize.OBJECT);
+
+  // HeapSize implementation
+  @Override
+  public long heapSize() {
+    return getCurrentSize();
+  }
+
+  public static long calculateOverhead(long maxSize, long blockSize, int concurrency) {
+    return CACHE_FIXED_OVERHEAD + ClassSize.CONCURRENT_HASHMAP + ((int) Math.ceil(maxSize * 1.2 / blockSize) * ClassSize.CONCURRENT_HASHMAP_ENTRY)
+        + (concurrency * ClassSize.CONCURRENT_HASHMAP_SEGMENT);
+  }
+
+  // Simple calculators of sizes given factors and maxSize
+
+  private long acceptableSize() {
+    return (long) Math.floor(this.conf.getMaxSize() * this.conf.getAcceptableFactor());
+  }
+
+  private long minSize() {
+    return (long) Math.floor(this.conf.getMaxSize() * this.conf.getMinFactor());
+  }
+
+  private long singleSize() {
+    return (long) Math.floor(this.conf.getMaxSize() * this.conf.getSingleFactor() * this.conf.getMinFactor());
+  }
+
+  private long multiSize() {
+    return (long) Math.floor(this.conf.getMaxSize() * this.conf.getMultiFactor() * this.conf.getMinFactor());
+  }
+
+  private long memorySize() {
+    return (long) Math.floor(this.conf.getMaxSize() * this.conf.getMemoryFactor() * this.conf.getMinFactor());
+  }
+
+  public void shutdown() {
+    this.scheduleThreadPool.shutdown();
+  }
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCacheConfiguration.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCacheConfiguration.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCacheConfiguration.java
new file mode 100644
index 0000000..49790cb
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCacheConfiguration.java
@@ -0,0 +1,212 @@
+/*
+ * 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.accumulo.core.file.blockfile.cache.lru;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.file.blockfile.cache.BlockCacheConfiguration;
+import org.apache.accumulo.core.file.blockfile.cache.CacheType;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+
+public final class LruBlockCacheConfiguration extends BlockCacheConfiguration {
+
+  public static final String PROPERTY_PREFIX = "lru";
+
+  /** Default Configuration Parameters */
+
+  /** Backing Concurrent Map Configuration */
+  public static final Float DEFAULT_LOAD_FACTOR = 0.75f;
+  public static final Integer DEFAULT_CONCURRENCY_LEVEL = 16;
+
+  /** Eviction thresholds */
+  public static final Float DEFAULT_MIN_FACTOR = 0.75f;
+  public static final Float DEFAULT_ACCEPTABLE_FACTOR = 0.85f;
+
+  /** Priority buckets */
+  public static final Float DEFAULT_SINGLE_FACTOR = 0.25f;
+  public static final Float DEFAULT_MULTI_FACTOR = 0.50f;
+  public static final Float DEFAULT_MEMORY_FACTOR = 0.25f;
+
+  // property names
+  public static final String ACCEPTABLE_FACTOR_PROPERTY = "acceptable.factor";
+  public static final String MIN_FACTOR_PROPERTY = "min.factor";
+  public static final String SINGLE_FACTOR_PROPERTY = "single.factor";
+  public static final String MULTI_FACTOR_PROPERTY = "multi.factor";
+  public static final String MEMORY_FACTOR_PROPERTY = "memory.factor";
+  public static final String MAP_LOAD_PROPERTY = "map.load";
+  public static final String MAP_CONCURRENCY_PROPERTY = "map.concurrency";
+  public static final String EVICTION_THREAD_PROPERTY = "eviction.thread";
+
+  /** Acceptable size of cache (no evictions if size < acceptable) */
+  private final float acceptableFactor;
+
+  /** Minimum threshold of cache (when evicting, evict until size < min) */
+  private final float minFactor;
+
+  /** Single access bucket size */
+  private final float singleFactor;
+
+  /** Multiple access bucket size */
+  private final float multiFactor;
+
+  /** In-memory bucket size */
+  private final float memoryFactor;
+
+  /** LruBlockCache cache = new LruBlockCache **/
+  private final float mapLoadFactor;
+
+  /** LruBlockCache cache = new LruBlockCache **/
+  private final int mapConcurrencyLevel;
+
+  private final boolean useEvictionThread;
+
+  public LruBlockCacheConfiguration(AccumuloConfiguration conf, CacheType type) {
+    super(conf, type, PROPERTY_PREFIX);
+
+    this.acceptableFactor = get(ACCEPTABLE_FACTOR_PROPERTY).map(Float::valueOf).filter(f -> f > 0).orElse(DEFAULT_ACCEPTABLE_FACTOR);
+    this.minFactor = get(MIN_FACTOR_PROPERTY).map(Float::valueOf).filter(f -> f > 0).orElse(DEFAULT_MIN_FACTOR);
+    this.singleFactor = get(SINGLE_FACTOR_PROPERTY).map(Float::valueOf).filter(f -> f > 0).orElse(DEFAULT_SINGLE_FACTOR);
+    this.multiFactor = get(MULTI_FACTOR_PROPERTY).map(Float::valueOf).filter(f -> f > 0).orElse(DEFAULT_MULTI_FACTOR);
+    this.memoryFactor = get(MEMORY_FACTOR_PROPERTY).map(Float::valueOf).filter(f -> f > 0).orElse(DEFAULT_MEMORY_FACTOR);
+    this.mapLoadFactor = get(MAP_LOAD_PROPERTY).map(Float::valueOf).filter(f -> f > 0).orElse(DEFAULT_LOAD_FACTOR);
+    this.mapConcurrencyLevel = get(MAP_CONCURRENCY_PROPERTY).map(Integer::valueOf).filter(i -> i > 0).orElse(DEFAULT_CONCURRENCY_LEVEL);
+    this.useEvictionThread = get(EVICTION_THREAD_PROPERTY).map(Boolean::valueOf).orElse(true);
+
+    if (this.getSingleFactor() + this.getMultiFactor() + this.getMemoryFactor() != 1) {
+      throw new IllegalArgumentException("Single, multi, and memory factors " + " should total 1.0");
+    }
+    if (this.getMinFactor() >= this.getAcceptableFactor()) {
+      throw new IllegalArgumentException("minFactor must be smaller than acceptableFactor");
+    }
+    if (this.getMinFactor() >= 1.0f || this.getAcceptableFactor() >= 1.0f) {
+      throw new IllegalArgumentException("all factors must be < 1");
+    }
+  }
+
+  public float getAcceptableFactor() {
+    return acceptableFactor;
+  }
+
+  public float getMinFactor() {
+    return minFactor;
+  }
+
+  public float getSingleFactor() {
+    return singleFactor;
+  }
+
+  public float getMultiFactor() {
+    return multiFactor;
+  }
+
+  public float getMemoryFactor() {
+    return memoryFactor;
+  }
+
+  public float getMapLoadFactor() {
+    return mapLoadFactor;
+  }
+
+  public int getMapConcurrencyLevel() {
+    return mapConcurrencyLevel;
+  }
+
+  public boolean isUseEvictionThread() {
+    return useEvictionThread;
+  }
+
+  public static class Builder {
+    private Map<String,String> props = new HashMap<>();
+    private String prefix;
+
+    private Builder(String prefix) {
+      this.prefix = prefix;
+    }
+
+    private void set(String prop, float val) {
+      props.put(prefix + prop, Float.toString(val));
+    }
+
+    public Builder acceptableFactor(float af) {
+      Preconditions.checkArgument(af > 0);
+      set(ACCEPTABLE_FACTOR_PROPERTY, af);
+      return this;
+    }
+
+    public Builder minFactor(float mf) {
+      Preconditions.checkArgument(mf > 0);
+      set(MIN_FACTOR_PROPERTY, mf);
+      return this;
+    }
+
+    public Builder singleFactor(float sf) {
+      Preconditions.checkArgument(sf > 0);
+      set(SINGLE_FACTOR_PROPERTY, sf);
+      return this;
+    }
+
+    public Builder multiFactor(float mf) {
+      Preconditions.checkArgument(mf > 0);
+      set(MULTI_FACTOR_PROPERTY, mf);
+      return this;
+    }
+
+    public Builder memoryFactor(float mf) {
+      Preconditions.checkArgument(mf > 0);
+      set(MEMORY_FACTOR_PROPERTY, mf);
+      return this;
+    }
+
+    public Builder mapLoadFactor(float mlf) {
+      Preconditions.checkArgument(mlf > 0);
+      set(MAP_LOAD_PROPERTY, mlf);
+      return this;
+    }
+
+    public Builder mapConcurrencyLevel(int mcl) {
+      Preconditions.checkArgument(mcl > 0);
+      props.put(prefix + MAP_CONCURRENCY_PROPERTY, mcl + "");
+      return this;
+    }
+
+    public Builder useEvictionThread(boolean uet) {
+      props.put(prefix + EVICTION_THREAD_PROPERTY, uet + "");
+      return this;
+    }
+
+    public Map<String,String> buildMap() {
+      return ImmutableMap.copyOf(props);
+    }
+  }
+
+  public static Builder builder(CacheType ct) {
+    return new Builder(getPrefix(ct, PROPERTY_PREFIX));
+  }
+
+  @Override
+  public String toString() {
+    return super.toString() + ", acceptableFactor: " + this.getAcceptableFactor() + ", minFactor: " + this.getMinFactor() + ", singleFactor: "
+        + this.getSingleFactor() + ", multiFactor: " + this.getMultiFactor() + ", memoryFactor: " + this.getMemoryFactor() + ", mapLoadFactor: "
+        + this.getMapLoadFactor() + ", mapConcurrencyLevel: " + this.getMapConcurrencyLevel() + ", useEvictionThread: " + this.isUseEvictionThread();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCacheManager.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCacheManager.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCacheManager.java
new file mode 100644
index 0000000..8a1e430
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/lru/LruBlockCacheManager.java
@@ -0,0 +1,49 @@
+/*
+ * 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.accumulo.core.file.blockfile.cache.lru;
+
+import org.apache.accumulo.core.conf.AccumuloConfiguration;
+import org.apache.accumulo.core.file.blockfile.cache.BlockCache;
+import org.apache.accumulo.core.file.blockfile.cache.BlockCacheManager;
+import org.apache.accumulo.core.file.blockfile.cache.CacheType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LruBlockCacheManager extends BlockCacheManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(LruBlockCacheManager.class);
+
+  @Override
+  protected BlockCache createCache(AccumuloConfiguration conf, CacheType type) {
+    LruBlockCacheConfiguration cc = new LruBlockCacheConfiguration(conf, type);
+    LOG.info("Creating {} cache with configuration {}", type, cc);
+    return new LruBlockCache(cc);
+  }
+
+  @Override
+  public void stop() {
+    for (CacheType type : CacheType.values()) {
+      LruBlockCache cache = ((LruBlockCache) this.getBlockCache(type));
+      if (null != cache) {
+        cache.shutdown();
+      }
+    }
+    super.stop();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/452732c8/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/tinylfu/TinyLfuBlockCache.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/tinylfu/TinyLfuBlockCache.java b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/tinylfu/TinyLfuBlockCache.java
new file mode 100644
index 0000000..db4e789
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/tinylfu/TinyLfuBlockCache.java
@@ -0,0 +1,150 @@
+/*
+ * 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.accumulo.core.file.blockfile.cache.tinylfu;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.accumulo.core.file.blockfile.cache.BlockCache;
+import org.apache.accumulo.core.file.blockfile.cache.CacheEntry;
+import org.apache.accumulo.core.file.blockfile.cache.ClassSize;
+import org.apache.accumulo.core.file.blockfile.cache.SizeConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Policy;
+import com.github.benmanes.caffeine.cache.stats.CacheStats;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+/**
+ * A block cache that is memory bounded using the W-TinyLFU eviction algorithm. This implementation delegates to a Caffeine cache to provide concurrent O(1)
+ * read and write operations.
+ * <ul>
+ * <li>W-TinyLFU: http://arxiv.org/pdf/1512.00727.pdf</li>
+ * <li>Caffeine: https://github.com/ben-manes/caffeine</li>
+ * <li>Cache design: http://highscalability.com/blog/2016/1/25/design-of-a-modern-cache.html</li>
+ * </ul>
+ */
+public final class TinyLfuBlockCache implements BlockCache {
+  private static final Logger log = LoggerFactory.getLogger(TinyLfuBlockCache.class);
+  private static final int STATS_PERIOD_SEC = 60;
+
+  private Cache<String,Block> cache;
+  private Policy.Eviction<String,Block> policy;
+  private ScheduledExecutorService statsExecutor;
+
+  public TinyLfuBlockCache(TinyLfuBlockCacheConfiguration conf) {
+    cache = Caffeine.newBuilder().initialCapacity((int) Math.ceil(1.2 * conf.getMaxSize() / conf.getBlockSize())).weigher((String blockName, Block block) -> {
+      int keyWeight = ClassSize.align(blockName.length()) + ClassSize.STRING;
+      return keyWeight + block.weight();
+    }).maximumWeight(conf.getMaxSize()).recordStats().build();
+    policy = cache.policy().eviction().get();
+    statsExecutor = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("TinyLfuBlockCacheStatsExecutor").setDaemon(true)
+        .build());
+    statsExecutor.scheduleAtFixedRate(this::logStats, STATS_PERIOD_SEC, STATS_PERIOD_SEC, TimeUnit.SECONDS);
+
+  }
+
+  @Override
+  public long getMaxHeapSize() {
+    return getMaxSize();
+  }
+
+  @Override
+  public long getMaxSize() {
+    return policy.getMaximum();
+  }
+
+  @Override
+  public CacheEntry getBlock(String blockName) {
+    return cache.getIfPresent(blockName);
+  }
+
+  @Override
+  public CacheEntry cacheBlock(String blockName, byte[] buffer) {
+    return cache.asMap().compute(blockName, (key, block) -> {
+      if (block == null) {
+        return new Block(buffer);
+      }
+      block.buffer = buffer;
+      return block;
+    });
+  }
+
+  @Override
+  public CacheEntry cacheBlock(String blockName, byte[] buffer, /* ignored */boolean inMemory) {
+    return cacheBlock(blockName, buffer);
+  }
+
+  @Override
+  public BlockCache.Stats getStats() {
+    CacheStats stats = cache.stats();
+    return new BlockCache.Stats() {
+      @Override
+      public long hitCount() {
+        return stats.hitCount();
+      }
+
+      @Override
+      public long requestCount() {
+        return stats.requestCount();
+      }
+    };
+  }
+
+  private void logStats() {
+    double maxMB = ((double) policy.getMaximum()) / ((double) (1024 * 1024));
+    double sizeMB = ((double) policy.weightedSize().getAsLong()) / ((double) (1024 * 1024));
+    double freeMB = maxMB - sizeMB;
+    log.debug("Cache Size={}MB, Free={}MB, Max={}MB, Blocks={}", sizeMB, freeMB, maxMB, cache.estimatedSize());
+    log.debug(cache.stats().toString());
+  }
+
+  private static final class Block implements CacheEntry {
+    private volatile byte[] buffer;
+    private volatile Object index;
+
+    Block(byte[] buffer) {
+      this.buffer = requireNonNull(buffer);
+    }
+
+    @Override
+    public byte[] getBuffer() {
+      return buffer;
+    }
+
+    @Override
+    public Object getIndex() {
+      return index;
+    }
+
+    @Override
+    public void setIndex(Object index) {
+      this.index = index;
+    }
+
+    int weight() {
+      return ClassSize.align(buffer.length) + SizeConstants.SIZEOF_LONG + ClassSize.REFERENCE + ClassSize.OBJECT + ClassSize.ARRAY;
+    }
+  }
+}


Mime
View raw message