accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From els...@apache.org
Subject accumulo git commit: ACCUMULO-4177 TinyLFU-based block cache
Date Fri, 16 Sep 2016 20:50:57 GMT
Repository: accumulo
Updated Branches:
  refs/heads/master c9391894e -> 9d94d30e1


ACCUMULO-4177 TinyLFU-based block cache

Signed-off-by: Josh Elser <elserj@apache.org>


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

Branch: refs/heads/master
Commit: 9d94d30e16a5768fd6441ef603ea7afe2e7d37f6
Parents: c939189
Author: Ben Manes <ben.manes@gmail.com>
Authored: Fri May 6 19:35:41 2016 -0700
Committer: Josh Elser <elserj@apache.org>
Committed: Fri Sep 16 16:26:48 2016 -0400

----------------------------------------------------------------------
 assemble/pom.xml                                |   4 +
 assemble/src/main/assemblies/component.xml      |   1 +
 core/pom.xml                                    |   4 +
 .../core/client/rfile/RFileScanner.java         |  15 ++
 .../org/apache/accumulo/core/conf/Property.java |   1 +
 .../core/file/blockfile/cache/BlockCache.java   |  25 ++++
 .../file/blockfile/cache/LruBlockCache.java     |  14 +-
 .../file/blockfile/cache/TinyLfuBlockCache.java | 141 +++++++++++++++++++
 pom.xml                                         |   5 +
 .../apache/accumulo/tserver/TabletServer.java   |   8 +-
 .../tserver/TabletServerResourceManager.java    |  22 ++-
 11 files changed, 224 insertions(+), 16 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/assemble/pom.xml
----------------------------------------------------------------------
diff --git a/assemble/pom.xml b/assemble/pom.xml
index d0f14a9..1f2a899 100644
--- a/assemble/pom.xml
+++ b/assemble/pom.xml
@@ -33,6 +33,10 @@
       <optional>true</optional>
     </dependency>
     <dependency>
+      <groupId>com.github.ben-manes.caffeine</groupId>
+      <artifactId>caffeine</artifactId>
+    </dependency>
+    <dependency>
       <groupId>com.google.code.gson</groupId>
       <artifactId>gson</artifactId>
       <optional>true</optional>

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/assemble/src/main/assemblies/component.xml
----------------------------------------------------------------------
diff --git a/assemble/src/main/assemblies/component.xml b/assemble/src/main/assemblies/component.xml
index 6fc6656..2156e40 100644
--- a/assemble/src/main/assemblies/component.xml
+++ b/assemble/src/main/assemblies/component.xml
@@ -31,6 +31,7 @@
         version listing for packaged artifacts -->
         <include>${groupId}:${artifactId}-*</include>
         <include>com.beust:jcommander</include>
+        <include>com.github.ben-manes.caffeine:caffeine</include>
         <include>com.google.code.gson:gson</include>
         <include>com.google.guava:guava</include>
         <include>com.google.protobuf:protobuf-java</include>

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/core/pom.xml
----------------------------------------------------------------------
diff --git a/core/pom.xml b/core/pom.xml
index 44afddb..2e858b8 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -31,6 +31,10 @@
       <artifactId>jcommander</artifactId>
     </dependency>
     <dependency>
+      <groupId>com.github.ben-manes.caffeine</groupId>
+      <artifactId>caffeine</artifactId>
+    </dependency>
+    <dependency>
       <groupId>com.google.auto.service</groupId>
       <artifactId>auto-service</artifactId>
       <optional>true</optional>

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/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 4dfba68..186471d 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
@@ -112,6 +112,21 @@ class RFileScanner extends ScannerOptions implements Scanner {
     public long getMaxSize() {
       return Integer.MAX_VALUE;
     }
+
+    @Override
+    public Stats getStats() {
+      return new BlockCache.Stats() {
+        @Override
+        public long hitCount() {
+          return 0L;
+        }
+
+        @Override
+        public long requestCount() {
+          return 0L;
+        }
+      };
+    }
   }
 
   RFileScanner(Opts opts) {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/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 ede1c6f..860e57b 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
@@ -243,6 +243,7 @@ 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.MEMORY, "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_DATACACHE_SIZE("tserver.cache.data.size", "128M", PropertyType.MEMORY, "Specifies
the size of the cache for file data blocks."),
   TSERV_INDEXCACHE_SIZE("tserver.cache.index.size", "512M", PropertyType.MEMORY, "Specifies
the size of the cache for file indices."),
   TSERV_PORTSEARCH("tserver.port.search", "false", PropertyType.BOOLEAN, "if the ports above
are in use, search higher ports until one is available"),

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/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 094782d..82f8b1e 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
@@ -58,4 +58,29 @@ public interface BlockCache {
    * @return max size in bytes
    */
   long getMaxSize();
+
+  /**
+   * Get the statistics of this cache.
+   *
+   * @return statistics
+   */
+  Stats getStats();
+
+  /** Cache statistics. */
+  interface Stats {
+
+    /**
+     * Returns the number of lookups that have returned a cached value.
+     *
+     * @return the number of lookups that have returned a cached value
+     */
+    long hitCount();
+
+    /**
+     * Returns the number of times the lookup methods have returned either a cached or uncached
value.
+     *
+     * @return the number of lookups
+     */
+    long requestCount();
+  }
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/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
index 1beaccb..cbdaca5 100644
--- 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
@@ -578,7 +578,7 @@ public class LruBlockCache implements BlockCache, HeapSize {
     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.getRequestCount(), stats.getHitCount(), stats.getMissCount(), stats.getEvictionCount(),
stats.getEvictedCount(), stats.getHitRatio() * 100,
+        stats.requestCount(), stats.hitCount(), stats.getMissCount(), stats.getEvictionCount(),
stats.getEvictedCount(), stats.getHitRatio() * 100,
         stats.getMissRatio() * 100, stats.evictedPerEviction(), stats.getDuplicateReads());
   }
 
@@ -592,7 +592,7 @@ public class LruBlockCache implements BlockCache, HeapSize {
     return this.stats;
   }
 
-  public static class CacheStats {
+  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);
@@ -622,7 +622,8 @@ public class LruBlockCache implements BlockCache, HeapSize {
       evictedCount.incrementAndGet();
     }
 
-    public long getRequestCount() {
+    @Override
+    public long requestCount() {
       return accessCount.get();
     }
 
@@ -630,7 +631,8 @@ public class LruBlockCache implements BlockCache, HeapSize {
       return missCount.get();
     }
 
-    public long getHitCount() {
+    @Override
+    public long hitCount() {
       return hitCount.get();
     }
 
@@ -647,11 +649,11 @@ public class LruBlockCache implements BlockCache, HeapSize {
     }
 
     public double getHitRatio() {
-      return ((float) getHitCount() / (float) getRequestCount());
+      return ((float) hitCount() / (float) requestCount());
     }
 
     public double getMissRatio() {
-      return ((float) getMissCount() / (float) getRequestCount());
+      return ((float) getMissCount() / (float) requestCount());
     }
 
     public double evictedPerEviction() {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/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
new file mode 100644
index 0000000..bab52af
--- /dev/null
+++ b/core/src/main/java/org/apache/accumulo/core/file/blockfile/cache/TinyLfuBlockCache.java
@@ -0,0 +1,141 @@
+/*
+ * 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/9d94d30e/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 54e4a72..fbe9070 100644
--- a/pom.xml
+++ b/pom.xml
@@ -164,6 +164,11 @@
         <version>1.48</version>
       </dependency>
       <dependency>
+        <groupId>com.github.ben-manes.caffeine</groupId>
+        <artifactId>caffeine</artifactId>
+        <version>2.3.3</version>
+      </dependency>
+      <dependency>
         <groupId>com.google.auto.service</groupId>
         <artifactId>auto-service</artifactId>
         <version>1.0-rc2</version>

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
index 7751681..450c1c6 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java
@@ -2893,10 +2893,10 @@ public class TabletServer extends AccumuloServerContext implements
Runnable {
     result.name = getClientAddressString();
     result.holdTime = resourceManager.holdTime();
     result.lookups = seekCount.get();
-    result.indexCacheHits = resourceManager.getIndexCache().getStats().getHitCount();
-    result.indexCacheRequest = resourceManager.getIndexCache().getStats().getRequestCount();
-    result.dataCacheHits = resourceManager.getDataCache().getStats().getHitCount();
-    result.dataCacheRequest = resourceManager.getDataCache().getStats().getRequestCount();
+    result.indexCacheHits = resourceManager.getIndexCache().getStats().hitCount();
+    result.indexCacheRequest = resourceManager.getIndexCache().getStats().requestCount();
+    result.dataCacheHits = resourceManager.getDataCache().getStats().hitCount();
+    result.dataCacheRequest = resourceManager.getDataCache().getStats().requestCount();
     result.logSorts = logSorter.getLogSorts();
     result.flushs = flushCounter.get();
     result.syncs = syncCounter.get();

http://git-wip-us.apache.org/repos/asf/accumulo/blob/9d94d30e/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java
----------------------------------------------------------------------
diff --git a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java
b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java
index 089bd12..3cd7bfa 100644
--- a/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java
+++ b/server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServerResourceManager.java
@@ -38,7 +38,9 @@ import java.util.concurrent.atomic.AtomicLong;
 import org.apache.accumulo.core.conf.AccumuloConfiguration;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.impl.KeyExtent;
+import org.apache.accumulo.core.file.blockfile.cache.BlockCache;
 import org.apache.accumulo.core.file.blockfile.cache.LruBlockCache;
+import org.apache.accumulo.core.file.blockfile.cache.TinyLfuBlockCache;
 import org.apache.accumulo.core.metadata.schema.DataFileValue;
 import org.apache.accumulo.core.util.Daemon;
 import org.apache.accumulo.core.util.NamingThreadFactory;
@@ -96,8 +98,8 @@ public class TabletServerResourceManager {
 
   private final MemoryManagementFramework memMgmt;
 
-  private final LruBlockCache _dCache;
-  private final LruBlockCache _iCache;
+  private final BlockCache _dCache;
+  private final BlockCache _iCache;
   private final TabletServer tserver;
   private final ServerConfigurationFactory conf;
 
@@ -163,8 +165,16 @@ public class TabletServerResourceManager {
     long iCacheSize = acuConf.getMemoryInBytes(Property.TSERV_INDEXCACHE_SIZE);
     long totalQueueSize = acuConf.getMemoryInBytes(Property.TSERV_TOTAL_MUTATION_QUEUE_MAX);
 
-    _iCache = new LruBlockCache(iCacheSize, blockSize);
-    _dCache = new LruBlockCache(dCacheSize, blockSize);
+    String policy = acuConf.get(Property.TSERV_CACHE_POLICY);
+    if (policy.equalsIgnoreCase("LRU")) {
+      _iCache = new LruBlockCache(iCacheSize, blockSize);
+      _dCache = new LruBlockCache(dCacheSize, blockSize);
+    } else if (policy.equalsIgnoreCase("TinyLFU")) {
+      _iCache = new TinyLfuBlockCache(iCacheSize, blockSize);
+      _dCache = new TinyLfuBlockCache(dCacheSize, blockSize);
+    } else {
+      throw new IllegalArgumentException("Unknown Block cache policy " + policy);
+    }
 
     Runtime runtime = Runtime.getRuntime();
     if (usingNativeMap) {
@@ -742,11 +752,11 @@ public class TabletServerResourceManager {
     }
   }
 
-  public LruBlockCache getIndexCache() {
+  public BlockCache getIndexCache() {
     return _iCache;
   }
 
-  public LruBlockCache getDataCache() {
+  public BlockCache getDataCache() {
     return _dCache;
   }
 


Mime
View raw message