View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.io.hfile;
20  
21  import java.lang.ref.WeakReference;
22  import java.nio.ByteBuffer;
23  import java.util.EnumMap;
24  import java.util.Iterator;
25  import java.util.List;
26  import java.util.Map;
27  import java.util.PriorityQueue;
28  import java.util.SortedSet;
29  import java.util.TreeSet;
30  import java.util.concurrent.ConcurrentHashMap;
31  import java.util.concurrent.Executors;
32  import java.util.concurrent.ScheduledExecutorService;
33  import java.util.concurrent.TimeUnit;
34  import java.util.concurrent.atomic.AtomicLong;
35  import java.util.concurrent.locks.ReentrantLock;
36  
37  import com.google.common.base.Objects;
38  import org.apache.commons.logging.Log;
39  import org.apache.commons.logging.LogFactory;
40  import org.apache.hadoop.hbase.classification.InterfaceAudience;
41  import org.apache.hadoop.conf.Configuration;
42  import org.apache.hadoop.hbase.io.HeapSize;
43  import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
44  import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache;
45  import org.apache.hadoop.hbase.util.Bytes;
46  import org.apache.hadoop.hbase.util.ClassSize;
47  import org.apache.hadoop.hbase.util.HasThread;
48  import org.apache.hadoop.util.StringUtils;
49  import org.codehaus.jackson.annotate.JsonIgnoreProperties;
50  
51  import com.google.common.annotations.VisibleForTesting;
52  import com.google.common.util.concurrent.ThreadFactoryBuilder;
53  
54  /**
55   * A block cache implementation that is memory-aware using {@link HeapSize},
56   * memory-bound using an LRU eviction algorithm, and concurrent: backed by a
57   * {@link ConcurrentHashMap} and with a non-blocking eviction thread giving
58   * constant-time {@link #cacheBlock} and {@link #getBlock} operations.<p>
59   *
60   * Contains three levels of block priority to allow for
61   * scan-resistance and in-memory families 
62   * {@link org.apache.hadoop.hbase.HColumnDescriptor#setInMemory(boolean)} (An
63   * in-memory column family is a column family that should be served from memory if possible):
64   * single-access, multiple-accesses, and in-memory priority.
65   * A block is added with an in-memory priority flag if
66   * {@link org.apache.hadoop.hbase.HColumnDescriptor#isInMemory()}, 
67   * otherwise a block becomes a single access
68   * priority the first time it is read into this block cache.  If a block is accessed again while
69   * in cache, it is marked as a multiple access priority block.  This delineation of blocks is used
70   * to prevent scans from thrashing the cache adding a least-frequently-used
71   * element to the eviction algorithm.<p>
72   *
73   * Each priority is given its own chunk of the total cache to ensure
74   * fairness during eviction.  Each priority will retain close to its maximum
75   * size, however, if any priority is not using its entire chunk the others
76   * are able to grow beyond their chunk size.<p>
77   *
78   * Instantiated at a minimum with the total size and average block size.
79   * All sizes are in bytes.  The block size is not especially important as this
80   * cache is fully dynamic in its sizing of blocks.  It is only used for
81   * pre-allocating data structures and in initial heap estimation of the map.<p>
82   *
83   * The detailed constructor defines the sizes for the three priorities (they
84   * should total to the <code>maximum size</code> defined).  It also sets the levels that
85   * trigger and control the eviction thread.<p>
86   *
87   * The <code>acceptable size</code> is the cache size level which triggers the eviction
88   * process to start.  It evicts enough blocks to get the size below the
89   * minimum size specified.<p>
90   *
91   * Eviction happens in a separate thread and involves a single full-scan
92   * of the map.  It determines how many bytes must be freed to reach the minimum
93   * size, and then while scanning determines the fewest least-recently-used
94   * blocks necessary from each of the three priorities (would be 3 times bytes
95   * to free).  It then uses the priority chunk sizes to evict fairly according
96   * to the relative sizes and usage.
97   */
98  @InterfaceAudience.Private
99  @JsonIgnoreProperties({"encodingCountsForTest"})
100 public class LruBlockCache implements ResizableBlockCache, HeapSize {
101 
102   private static final Log LOG = LogFactory.getLog(LruBlockCache.class);
103 
104   /**
105    * Percentage of total size that eviction will evict until; e.g. if set to .8, then we will keep
106    * evicting during an eviction run till the cache size is down to 80% of the total.
107    */
108   static final String LRU_MIN_FACTOR_CONFIG_NAME = "hbase.lru.blockcache.min.factor";
109 
110   /**
111    * Acceptable size of cache (no evictions if size < acceptable)
112    */
113   static final String LRU_ACCEPTABLE_FACTOR_CONFIG_NAME = "hbase.lru.blockcache.acceptable.factor";
114 
115   /**
116    * Hard capacity limit of cache, will reject any put if size > this * acceptable
117    */
118   static final String LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME = "hbase.lru.blockcache.hard.capacity.limit.factor";
119   static final String LRU_SINGLE_PERCENTAGE_CONFIG_NAME = "hbase.lru.blockcache.single.percentage";
120   static final String LRU_MULTI_PERCENTAGE_CONFIG_NAME = "hbase.lru.blockcache.multi.percentage";
121   static final String LRU_MEMORY_PERCENTAGE_CONFIG_NAME = "hbase.lru.blockcache.memory.percentage";
122 
123   /**
124    * Configuration key to force data-block always (except in-memory are too much)
125    * cached in memory for in-memory hfile, unlike inMemory, which is a column-family
126    * configuration, inMemoryForceMode is a cluster-wide configuration
127    */
128   static final String LRU_IN_MEMORY_FORCE_MODE_CONFIG_NAME = "hbase.lru.rs.inmemoryforcemode";
129 
130   /** Default Configuration Parameters*/
131 
132   /** Backing Concurrent Map Configuration */
133   static final float DEFAULT_LOAD_FACTOR = 0.75f;
134   static final int DEFAULT_CONCURRENCY_LEVEL = 16;
135 
136   /** Eviction thresholds */
137   static final float DEFAULT_MIN_FACTOR = 0.95f;
138   static final float DEFAULT_ACCEPTABLE_FACTOR = 0.99f;
139 
140   /** Priority buckets */
141   static final float DEFAULT_SINGLE_FACTOR = 0.25f;
142   static final float DEFAULT_MULTI_FACTOR = 0.50f;
143   static final float DEFAULT_MEMORY_FACTOR = 0.25f;
144 
145   /** default hard capacity limit */
146   static final float DEFAULT_HARD_CAPACITY_LIMIT_FACTOR = 1.2f;
147 
148   static final boolean DEFAULT_IN_MEMORY_FORCE_MODE = false;
149 
150   /** Statistics thread */
151   static final int statThreadPeriod = 60 * 5;
152   private static final String LRU_MAX_BLOCK_SIZE = "hbase.lru.max.block.size";
153   private static final long DEFAULT_MAX_BLOCK_SIZE = 16L * 1024L * 1024L;
154 
155   /** Concurrent map (the cache) */
156   private final Map<BlockCacheKey,LruCachedBlock> map;
157 
158   /** Eviction lock (locked when eviction in process) */
159   private final ReentrantLock evictionLock = new ReentrantLock(true);
160   private final long maxBlockSize;
161 
162   /** Volatile boolean to track if we are in an eviction process or not */
163   private volatile boolean evictionInProgress = false;
164 
165   /** Eviction thread */
166   private final EvictionThread evictionThread;
167 
168   /** Statistics thread schedule pool (for heavy debugging, could remove) */
169   private final ScheduledExecutorService scheduleThreadPool = Executors.newScheduledThreadPool(1,
170     new ThreadFactoryBuilder().setNameFormat("LruBlockCacheStatsExecutor").setDaemon(true).build());
171 
172   /** Current size of cache */
173   private final AtomicLong size;
174 
175   /** Current number of cached elements */
176   private final AtomicLong elements;
177 
178   /** Cache access count (sequential ID) */
179   private final AtomicLong count;
180 
181   /** hard capacity limit */
182   private float hardCapacityLimitFactor;
183 
184   /** Cache statistics */
185   private final CacheStats stats;
186 
187   /** Maximum allowable size of cache (block put if size > max, evict) */
188   private long maxSize;
189 
190   /** Approximate block size */
191   private long blockSize;
192 
193   /** Acceptable size of cache (no evictions if size < acceptable) */
194   private float acceptableFactor;
195 
196   /** Minimum threshold of cache (when evicting, evict until size < min) */
197   private float minFactor;
198 
199   /** Single access bucket size */
200   private float singleFactor;
201 
202   /** Multiple access bucket size */
203   private float multiFactor;
204 
205   /** In-memory bucket size */
206   private float memoryFactor;
207 
208   /** Overhead of the structure itself */
209   private long overhead;
210 
211   /** Whether in-memory hfile's data block has higher priority when evicting */
212   private boolean forceInMemory;
213 
214   /** Where to send victims (blocks evicted/missing from the cache) */
215   private BlockCache victimHandler = null;
216 
217   /**
218    * Default constructor.  Specify maximum size and expected average block
219    * size (approximation is fine).
220    *
221    * <p>All other factors will be calculated based on defaults specified in
222    * this class.
223    * @param maxSize maximum size of cache, in bytes
224    * @param blockSize approximate size of each block, in bytes
225    */
226   public LruBlockCache(long maxSize, long blockSize) {
227     this(maxSize, blockSize, true);
228   }
229 
230   /**
231    * Constructor used for testing.  Allows disabling of the eviction thread.
232    */
233   public LruBlockCache(long maxSize, long blockSize, boolean evictionThread) {
234     this(maxSize, blockSize, evictionThread,
235         (int)Math.ceil(1.2*maxSize/blockSize),
236         DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL,
237         DEFAULT_MIN_FACTOR, DEFAULT_ACCEPTABLE_FACTOR,
238         DEFAULT_SINGLE_FACTOR,
239         DEFAULT_MULTI_FACTOR,
240         DEFAULT_MEMORY_FACTOR,
241         DEFAULT_HARD_CAPACITY_LIMIT_FACTOR,
242         false,
243         DEFAULT_MAX_BLOCK_SIZE
244         );
245   }
246 
247   public LruBlockCache(long maxSize, long blockSize, boolean evictionThread, Configuration conf) {
248     this(maxSize, blockSize, evictionThread,
249         (int)Math.ceil(1.2*maxSize/blockSize),
250         DEFAULT_LOAD_FACTOR,
251         DEFAULT_CONCURRENCY_LEVEL,
252         conf.getFloat(LRU_MIN_FACTOR_CONFIG_NAME, DEFAULT_MIN_FACTOR),
253         conf.getFloat(LRU_ACCEPTABLE_FACTOR_CONFIG_NAME, DEFAULT_ACCEPTABLE_FACTOR),
254         conf.getFloat(LRU_SINGLE_PERCENTAGE_CONFIG_NAME, DEFAULT_SINGLE_FACTOR),
255         conf.getFloat(LRU_MULTI_PERCENTAGE_CONFIG_NAME, DEFAULT_MULTI_FACTOR),
256         conf.getFloat(LRU_MEMORY_PERCENTAGE_CONFIG_NAME, DEFAULT_MEMORY_FACTOR),
257         conf.getFloat(LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME, DEFAULT_HARD_CAPACITY_LIMIT_FACTOR),
258         conf.getBoolean(LRU_IN_MEMORY_FORCE_MODE_CONFIG_NAME, DEFAULT_IN_MEMORY_FORCE_MODE),
259         conf.getLong(LRU_MAX_BLOCK_SIZE, DEFAULT_MAX_BLOCK_SIZE)
260         );
261   }
262 
263   public LruBlockCache(long maxSize, long blockSize, Configuration conf) {
264     this(maxSize, blockSize, true, conf);
265   }
266 
267   /**
268    * Configurable constructor.  Use this constructor if not using defaults.
269    * @param maxSize maximum size of this cache, in bytes
270    * @param blockSize expected average size of blocks, in bytes
271    * @param evictionThread whether to run evictions in a bg thread or not
272    * @param mapInitialSize initial size of backing ConcurrentHashMap
273    * @param mapLoadFactor initial load factor of backing ConcurrentHashMap
274    * @param mapConcurrencyLevel initial concurrency factor for backing CHM
275    * @param minFactor percentage of total size that eviction will evict until
276    * @param acceptableFactor percentage of total size that triggers eviction
277    * @param singleFactor percentage of total size for single-access blocks
278    * @param multiFactor percentage of total size for multiple-access blocks
279    * @param memoryFactor percentage of total size for in-memory blocks
280    */
281   public LruBlockCache(long maxSize, long blockSize, boolean evictionThread,
282       int mapInitialSize, float mapLoadFactor, int mapConcurrencyLevel,
283       float minFactor, float acceptableFactor, float singleFactor,
284       float multiFactor, float memoryFactor, float hardLimitFactor,
285       boolean forceInMemory, long maxBlockSize) {
286     this.maxBlockSize = maxBlockSize;
287     if(singleFactor + multiFactor + memoryFactor != 1 ||
288         singleFactor < 0 || multiFactor < 0 || memoryFactor < 0) {
289       throw new IllegalArgumentException("Single, multi, and memory factors " +
290           " should be non-negative and total 1.0");
291     }
292     if(minFactor >= acceptableFactor) {
293       throw new IllegalArgumentException("minFactor must be smaller than acceptableFactor");
294     }
295     if(minFactor >= 1.0f || acceptableFactor >= 1.0f) {
296       throw new IllegalArgumentException("all factors must be < 1");
297     }
298     this.maxSize = maxSize;
299     this.blockSize = blockSize;
300     this.forceInMemory = forceInMemory;
301     map = new ConcurrentHashMap<BlockCacheKey,LruCachedBlock>(mapInitialSize,
302         mapLoadFactor, mapConcurrencyLevel);
303     this.minFactor = minFactor;
304     this.acceptableFactor = acceptableFactor;
305     this.singleFactor = singleFactor;
306     this.multiFactor = multiFactor;
307     this.memoryFactor = memoryFactor;
308     this.stats = new CacheStats(this.getClass().getSimpleName());
309     this.count = new AtomicLong(0);
310     this.elements = new AtomicLong(0);
311     this.overhead = calculateOverhead(maxSize, blockSize, mapConcurrencyLevel);
312     this.size = new AtomicLong(this.overhead);
313     this.hardCapacityLimitFactor = hardLimitFactor;
314     if(evictionThread) {
315       this.evictionThread = new EvictionThread(this);
316       this.evictionThread.start(); // FindBugs SC_START_IN_CTOR
317     } else {
318       this.evictionThread = null;
319     }
320     // TODO: Add means of turning this off.  Bit obnoxious running thread just to make a log
321     // every five minutes.
322     this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
323         statThreadPeriod, statThreadPeriod, TimeUnit.SECONDS);
324   }
325 
326   @Override
327   public void setMaxSize(long maxSize) {
328     this.maxSize = maxSize;
329     if(this.size.get() > acceptableSize() && !evictionInProgress) {
330       runEviction();
331     }
332   }
333 
334   // BlockCache implementation
335 
336   /**
337    * Cache the block with the specified name and buffer.
338    * <p>
339    * It is assumed this will NOT be called on an already cached block. In rare cases (HBASE-8547)
340    * this can happen, for which we compare the buffer contents.
341    * @param cacheKey block's cache key
342    * @param buf block buffer
343    * @param inMemory if block is in-memory
344    * @param cacheDataInL1
345    */
346   @Override
347   public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
348       final boolean cacheDataInL1) {
349 
350     if (buf.heapSize() > maxBlockSize) {
351       // If there are a lot of blocks that are too
352       // big this can make the logs way too noisy.
353       // So we log 2%
354       if (stats.failInsert() % 50 == 0) {
355         LOG.warn("Trying to cache too large a block "
356             + cacheKey.getHfileName() + " @ "
357             + cacheKey.getOffset()
358             + " is " + buf.heapSize()
359             + " which is larger than " + maxBlockSize);
360       }
361       return;
362     }
363 
364     LruCachedBlock cb = map.get(cacheKey);
365     if (cb != null) {
366       // compare the contents, if they are not equal, we are in big trouble
367       if (compare(buf, cb.getBuffer()) != 0) {
368         throw new RuntimeException("Cached block contents differ, which should not have happened."
369           + "cacheKey:" + cacheKey);
370       }
371       String msg = "Cached an already cached block: " + cacheKey + " cb:" + cb.getCacheKey();
372       msg += ". This is harmless and can happen in rare cases (see HBASE-8547)";
373       LOG.warn(msg);
374       return;
375     }
376     long currentSize = size.get();
377     long currentAcceptableSize = acceptableSize();
378     long hardLimitSize = (long) (hardCapacityLimitFactor * currentAcceptableSize);
379     if (currentSize >= hardLimitSize) {
380       stats.failInsert();
381       if (LOG.isTraceEnabled()) {
382         LOG.trace("LruBlockCache current size " + StringUtils.byteDesc(currentSize)
383           + " has exceeded acceptable size " + StringUtils.byteDesc(currentAcceptableSize) + "  too many."
384           + " the hard limit size is " + StringUtils.byteDesc(hardLimitSize) + ", failed to put cacheKey:"
385           + cacheKey + " into LruBlockCache.");
386       }
387       if (!evictionInProgress) {
388         runEviction();
389       }
390       return;
391     }
392     cb = new LruCachedBlock(cacheKey, buf, count.incrementAndGet(), inMemory);
393     long newSize = updateSizeMetrics(cb, false);
394     map.put(cacheKey, cb);
395     long val = elements.incrementAndGet();
396     if (LOG.isTraceEnabled()) {
397       long size = map.size();
398       assertCounterSanity(size, val);
399     }
400     if (newSize > currentAcceptableSize && !evictionInProgress) {
401       runEviction();
402     }
403   }
404 
405   /**
406    * Sanity-checking for parity between actual block cache content and metrics.
407    * Intended only for use with TRACE level logging and -ea JVM.
408    */
409   private static void assertCounterSanity(long mapSize, long counterVal) {
410     if (counterVal < 0) {
411       LOG.trace("counterVal overflow. Assertions unreliable. counterVal=" + counterVal +
412         ", mapSize=" + mapSize);
413       return;
414     }
415     if (mapSize < Integer.MAX_VALUE) {
416       double pct_diff = Math.abs((((double) counterVal) / ((double) mapSize)) - 1.);
417       if (pct_diff > 0.05) {
418         LOG.trace("delta between reported and actual size > 5%. counterVal=" + counterVal +
419           ", mapSize=" + mapSize);
420       }
421     }
422   }
423 
424   private int compare(Cacheable left, Cacheable right) {
425     ByteBuffer l = ByteBuffer.allocate(left.getSerializedLength());
426     left.serialize(l);
427     ByteBuffer r = ByteBuffer.allocate(right.getSerializedLength());
428     right.serialize(r);
429     return Bytes.compareTo(l.array(), l.arrayOffset(), l.limit(),
430       r.array(), r.arrayOffset(), r.limit());
431   }
432 
433   /**
434    * Cache the block with the specified name and buffer.
435    * <p>
436    * @param cacheKey block's cache key
437    * @param buf block buffer
438    */
439   public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
440     cacheBlock(cacheKey, buf, false, false);
441   }
442 
443   /**
444    * Helper function that updates the local size counter and also updates any
445    * per-cf or per-blocktype metrics it can discern from given
446    * {@link LruCachedBlock}
447    *
448    * @param cb
449    * @param evict
450    */
451   protected long updateSizeMetrics(LruCachedBlock cb, boolean evict) {
452     long heapsize = cb.heapSize();
453     if (evict) {
454       heapsize *= -1;
455     }
456     return size.addAndGet(heapsize);
457   }
458 
459   /**
460    * Get the buffer of the block with the specified name.
461    * @param cacheKey block's cache key
462    * @param caching true if the caller caches blocks on cache misses
463    * @param repeat Whether this is a repeat lookup for the same block
464    *        (used to avoid double counting cache misses when doing double-check locking)
465    * @param updateCacheMetrics Whether to update cache metrics or not
466    * @return buffer of specified cache key, or null if not in cache
467    */
468   @Override
469   public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching, boolean repeat,
470       boolean updateCacheMetrics) {
471     LruCachedBlock cb = map.get(cacheKey);
472     if (cb == null) {
473       if (!repeat && updateCacheMetrics) {
474         stats.miss(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
475       }
476       // If there is another block cache then try and read there.
477       // However if this is a retry ( second time in double checked locking )
478       // And it's already a miss then the l2 will also be a miss.
479       if (victimHandler != null && !repeat) {
480         Cacheable result = victimHandler.getBlock(cacheKey, caching, repeat, updateCacheMetrics);
481 
482         // Promote this to L1.
483         if (result != null && caching) {
484           cacheBlock(cacheKey, result, /* inMemory = */ false, /* cacheData = */ true);
485         }
486         return result;
487       }
488       return null;
489     }
490     if (updateCacheMetrics) stats.hit(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
491     cb.access(count.incrementAndGet());
492     return cb.getBuffer();
493   }
494 
495   /**
496    * Whether the cache contains block with specified cacheKey
497    * @param cacheKey
498    * @return true if contains the block
499    */
500   public boolean containsBlock(BlockCacheKey cacheKey) {
501     return map.containsKey(cacheKey);
502   }
503 
504   @Override
505   public boolean evictBlock(BlockCacheKey cacheKey) {
506     LruCachedBlock cb = map.get(cacheKey);
507     if (cb == null) return false;
508     evictBlock(cb, false);
509     return true;
510   }
511 
512   /**
513    * Evicts all blocks for a specific HFile. This is an
514    * expensive operation implemented as a linear-time search through all blocks
515    * in the cache. Ideally this should be a search in a log-access-time map.
516    *
517    * <p>
518    * This is used for evict-on-close to remove all blocks of a specific HFile.
519    *
520    * @return the number of blocks evicted
521    */
522   @Override
523   public int evictBlocksByHfileName(String hfileName) {
524     int numEvicted = 0;
525     for (BlockCacheKey key : map.keySet()) {
526       if (key.getHfileName().equals(hfileName)) {
527         if (evictBlock(key))
528           ++numEvicted;
529       }
530     }
531     if (victimHandler != null) {
532       numEvicted += victimHandler.evictBlocksByHfileName(hfileName);
533     }
534     return numEvicted;
535   }
536 
537   /**
538    * Evict the block, and it will be cached by the victim handler if exists &amp;&amp;
539    * block may be read again later
540    * @param block
541    * @param evictedByEvictionProcess true if the given block is evicted by
542    *          EvictionThread
543    * @return the heap size of evicted block
544    */
545   protected long evictBlock(LruCachedBlock block, boolean evictedByEvictionProcess) {
546     map.remove(block.getCacheKey());
547     updateSizeMetrics(block, true);
548     long val = elements.decrementAndGet();
549     if (LOG.isTraceEnabled()) {
550       long size = map.size();
551       assertCounterSanity(size, val);
552     }
553     stats.evicted(block.getCachedTime(), block.getCacheKey().isPrimary());
554     if (evictedByEvictionProcess && victimHandler != null) {
555       if (victimHandler instanceof BucketCache) {
556         boolean wait = getCurrentSize() < acceptableSize();
557         boolean inMemory = block.getPriority() == BlockPriority.MEMORY;
558         ((BucketCache)victimHandler).cacheBlockWithWait(block.getCacheKey(), block.getBuffer(),
559             inMemory, wait);
560       } else {
561         victimHandler.cacheBlock(block.getCacheKey(), block.getBuffer());
562       }
563     }
564     return block.heapSize();
565   }
566 
567   /**
568    * Multi-threaded call to run the eviction process.
569    */
570   private void runEviction() {
571     if(evictionThread == null) {
572       evict();
573     } else {
574       evictionThread.evict();
575     }
576   }
577 
578   /**
579    * Eviction method.
580    */
581   void evict() {
582 
583     // Ensure only one eviction at a time
584     if(!evictionLock.tryLock()) return;
585 
586     try {
587       evictionInProgress = true;
588       long currentSize = this.size.get();
589       long bytesToFree = currentSize - minSize();
590 
591       if (LOG.isTraceEnabled()) {
592         LOG.trace("Block cache LRU eviction started; Attempting to free " +
593           StringUtils.byteDesc(bytesToFree) + " of total=" +
594           StringUtils.byteDesc(currentSize));
595       }
596 
597       if(bytesToFree <= 0) return;
598 
599       // Instantiate priority buckets
600       BlockBucket bucketSingle = new BlockBucket("single", bytesToFree, blockSize,
601           singleSize());
602       BlockBucket bucketMulti = new BlockBucket("multi", bytesToFree, blockSize,
603           multiSize());
604       BlockBucket bucketMemory = new BlockBucket("memory", bytesToFree, blockSize,
605           memorySize());
606 
607       // Scan entire map putting into appropriate buckets
608       for(LruCachedBlock cachedBlock : map.values()) {
609         switch(cachedBlock.getPriority()) {
610           case SINGLE: {
611             bucketSingle.add(cachedBlock);
612             break;
613           }
614           case MULTI: {
615             bucketMulti.add(cachedBlock);
616             break;
617           }
618           case MEMORY: {
619             bucketMemory.add(cachedBlock);
620             break;
621           }
622         }
623       }
624 
625       long bytesFreed = 0;
626       if (forceInMemory || memoryFactor > 0.999f) {
627         long s = bucketSingle.totalSize();
628         long m = bucketMulti.totalSize();
629         if (bytesToFree > (s + m)) {
630           // this means we need to evict blocks in memory bucket to make room,
631           // so the single and multi buckets will be emptied
632           bytesFreed = bucketSingle.free(s);
633           bytesFreed += bucketMulti.free(m);
634           if (LOG.isTraceEnabled()) {
635             LOG.trace("freed " + StringUtils.byteDesc(bytesFreed) +
636               " from single and multi buckets");
637           }
638           bytesFreed += bucketMemory.free(bytesToFree - bytesFreed);
639           if (LOG.isTraceEnabled()) {
640             LOG.trace("freed " + StringUtils.byteDesc(bytesFreed) +
641               " total from all three buckets ");
642           }
643         } else {
644           // this means no need to evict block in memory bucket,
645           // and we try best to make the ratio between single-bucket and
646           // multi-bucket is 1:2
647           long bytesRemain = s + m - bytesToFree;
648           if (3 * s <= bytesRemain) {
649             // single-bucket is small enough that no eviction happens for it
650             // hence all eviction goes from multi-bucket
651             bytesFreed = bucketMulti.free(bytesToFree);
652           } else if (3 * m <= 2 * bytesRemain) {
653             // multi-bucket is small enough that no eviction happens for it
654             // hence all eviction goes from single-bucket
655             bytesFreed = bucketSingle.free(bytesToFree);
656           } else {
657             // both buckets need to evict some blocks
658             bytesFreed = bucketSingle.free(s - bytesRemain / 3);
659             if (bytesFreed < bytesToFree) {
660               bytesFreed += bucketMulti.free(bytesToFree - bytesFreed);
661             }
662           }
663         }
664       } else {
665         PriorityQueue<BlockBucket> bucketQueue =
666           new PriorityQueue<BlockBucket>(3);
667 
668         bucketQueue.add(bucketSingle);
669         bucketQueue.add(bucketMulti);
670         bucketQueue.add(bucketMemory);
671 
672         int remainingBuckets = 3;
673 
674         BlockBucket bucket;
675         while((bucket = bucketQueue.poll()) != null) {
676           long overflow = bucket.overflow();
677           if(overflow > 0) {
678             long bucketBytesToFree = Math.min(overflow,
679                 (bytesToFree - bytesFreed) / remainingBuckets);
680             bytesFreed += bucket.free(bucketBytesToFree);
681           }
682           remainingBuckets--;
683         }
684       }
685 
686       if (LOG.isTraceEnabled()) {
687         long single = bucketSingle.totalSize();
688         long multi = bucketMulti.totalSize();
689         long memory = bucketMemory.totalSize();
690         LOG.trace("Block cache LRU eviction completed; " +
691           "freed=" + StringUtils.byteDesc(bytesFreed) + ", " +
692           "total=" + StringUtils.byteDesc(this.size.get()) + ", " +
693           "single=" + StringUtils.byteDesc(single) + ", " +
694           "multi=" + StringUtils.byteDesc(multi) + ", " +
695           "memory=" + StringUtils.byteDesc(memory));
696       }
697     } finally {
698       stats.evict();
699       evictionInProgress = false;
700       evictionLock.unlock();
701     }
702   }
703 
704   @Override
705   public String toString() {
706     return Objects.toStringHelper(this)
707       .add("blockCount", getBlockCount())
708       .add("currentSize", getCurrentSize())
709       .add("freeSize", getFreeSize())
710       .add("maxSize", getMaxSize())
711       .add("heapSize", heapSize())
712       .add("minSize", minSize())
713       .add("minFactor", minFactor)
714       .add("multiSize", multiSize())
715       .add("multiFactor", multiFactor)
716       .add("singleSize", singleSize())
717       .add("singleFactor", singleFactor)
718       .toString();
719   }
720 
721   /**
722    * Used to group blocks into priority buckets.  There will be a BlockBucket
723    * for each priority (single, multi, memory).  Once bucketed, the eviction
724    * algorithm takes the appropriate number of elements out of each according
725    * to configuration parameters and their relatives sizes.
726    */
727   private class BlockBucket implements Comparable<BlockBucket> {
728 
729     private final String name;
730     private LruCachedBlockQueue queue;
731     private long totalSize = 0;
732     private long bucketSize;
733 
734     public BlockBucket(String name, long bytesToFree, long blockSize, long bucketSize) {
735       this.name = name;
736       this.bucketSize = bucketSize;
737       queue = new LruCachedBlockQueue(bytesToFree, blockSize);
738       totalSize = 0;
739     }
740 
741     public void add(LruCachedBlock block) {
742       totalSize += block.heapSize();
743       queue.add(block);
744     }
745 
746     public long free(long toFree) {
747       if (LOG.isTraceEnabled()) {
748         LOG.trace("freeing " + StringUtils.byteDesc(toFree) + " from " + this);
749       }
750       LruCachedBlock cb;
751       long freedBytes = 0;
752       while ((cb = queue.pollLast()) != null) {
753         freedBytes += evictBlock(cb, true);
754         if (freedBytes >= toFree) {
755           return freedBytes;
756         }
757       }
758       if (LOG.isTraceEnabled()) {
759         LOG.trace("freed " + StringUtils.byteDesc(freedBytes) + " from " + this);
760       }
761       return freedBytes;
762     }
763 
764     public long overflow() {
765       return totalSize - bucketSize;
766     }
767 
768     public long totalSize() {
769       return totalSize;
770     }
771 
772     public int compareTo(BlockBucket that) {
773       return Long.compare(this.overflow(), that.overflow());
774     }
775 
776     @Override
777     public boolean equals(Object that) {
778       if (that == null || !(that instanceof BlockBucket)){
779         return false;
780       }
781       return compareTo((BlockBucket)that) == 0;
782     }
783 
784     @Override
785     public int hashCode() {
786       return Objects.hashCode(name, bucketSize, queue, totalSize);
787     }
788 
789     @Override
790     public String toString() {
791       return Objects.toStringHelper(this)
792         .add("name", name)
793         .add("totalSize", StringUtils.byteDesc(totalSize))
794         .add("bucketSize", StringUtils.byteDesc(bucketSize))
795         .toString();
796     }
797   }
798 
799   /**
800    * Get the maximum size of this cache.
801    * @return max size in bytes
802    */
803   public long getMaxSize() {
804     return this.maxSize;
805   }
806 
807   @Override
808   public long getCurrentSize() {
809     return this.size.get();
810   }
811 
812   @Override
813   public long getFreeSize() {
814     return getMaxSize() - getCurrentSize();
815   }
816 
817   @Override
818   public long size() {
819     return getMaxSize();
820   }
821 
822   @Override
823   public long getBlockCount() {
824     return this.elements.get();
825   }
826 
827   EvictionThread getEvictionThread() {
828     return this.evictionThread;
829   }
830 
831   /*
832    * Eviction thread.  Sits in waiting state until an eviction is triggered
833    * when the cache size grows above the acceptable level.<p>
834    *
835    * Thread is triggered into action by {@link LruBlockCache#runEviction()}
836    */
837   static class EvictionThread extends HasThread {
838     private WeakReference<LruBlockCache> cache;
839     private volatile boolean go = true;
840     // flag set after enter the run method, used for test
841     private boolean enteringRun = false;
842 
843     public EvictionThread(LruBlockCache cache) {
844       super(Thread.currentThread().getName() + ".LruBlockCache.EvictionThread");
845       setDaemon(true);
846       this.cache = new WeakReference<LruBlockCache>(cache);
847     }
848 
849     @Override
850     public void run() {
851       enteringRun = true;
852       while (this.go) {
853         synchronized(this) {
854           try {
855             this.wait(1000 * 10/*Don't wait for ever*/);
856           } catch(InterruptedException e) {
857             LOG.warn("Interrupted eviction thread ", e);
858             Thread.currentThread().interrupt();
859           }
860         }
861         LruBlockCache cache = this.cache.get();
862         if (cache == null) break;
863         cache.evict();
864       }
865     }
866 
867     @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NN_NAKED_NOTIFY",
868         justification="This is what we want")
869     public void evict() {
870       synchronized(this) {
871         this.notifyAll();
872       }
873     }
874 
875     synchronized void shutdown() {
876       this.go = false;
877       this.notifyAll();
878     }
879 
880     /**
881      * Used for the test.
882      */
883     boolean isEnteringRun() {
884       return this.enteringRun;
885     }
886   }
887 
888   /*
889    * Statistics thread.  Periodically prints the cache statistics to the log.
890    */
891   static class StatisticsThread extends Thread {
892     private final LruBlockCache lru;
893 
894     public StatisticsThread(LruBlockCache lru) {
895       super("LruBlockCacheStats");
896       setDaemon(true);
897       this.lru = lru;
898     }
899 
900     @Override
901     public void run() {
902       lru.logStats();
903     }
904   }
905 
906   public void logStats() {
907     // Log size
908     long totalSize = heapSize();
909     long freeSize = maxSize - totalSize;
910     LruBlockCache.LOG.info("totalSize=" + StringUtils.byteDesc(totalSize) + ", " +
911         "freeSize=" + StringUtils.byteDesc(freeSize) + ", " +
912         "max=" + StringUtils.byteDesc(this.maxSize) + ", " +
913         "blockCount=" + getBlockCount() + ", " +
914         "accesses=" + stats.getRequestCount() + ", " +
915         "hits=" + stats.getHitCount() + ", " +
916         "hitRatio=" + (stats.getHitCount() == 0 ?
917           "0" : (StringUtils.formatPercent(stats.getHitRatio(), 2)+ ", ")) + ", " +
918         "cachingAccesses=" + stats.getRequestCachingCount() + ", " +
919         "cachingHits=" + stats.getHitCachingCount() + ", " +
920         "cachingHitsRatio=" + (stats.getHitCachingCount() == 0 ?
921           "0,": (StringUtils.formatPercent(stats.getHitCachingRatio(), 2) + ", ")) +
922         "evictions=" + stats.getEvictionCount() + ", " +
923         "evicted=" + stats.getEvictedCount() + ", " +
924         "evictedPerRun=" + stats.evictedPerEviction());
925   }
926 
927   /**
928    * Get counter statistics for this cache.
929    *
930    * <p>Includes: total accesses, hits, misses, evicted blocks, and runs
931    * of the eviction processes.
932    */
933   public CacheStats getStats() {
934     return this.stats;
935   }
936 
937   public final static long CACHE_FIXED_OVERHEAD = ClassSize.align(
938       (4 * Bytes.SIZEOF_LONG) + (9 * ClassSize.REFERENCE) +
939       (6 * Bytes.SIZEOF_FLOAT) + (2 * Bytes.SIZEOF_BOOLEAN)
940       + ClassSize.OBJECT);
941 
942   @Override
943   public long heapSize() {
944     return getCurrentSize();
945   }
946 
947   public static long calculateOverhead(long maxSize, long blockSize, int concurrency){
948     // FindBugs ICAST_INTEGER_MULTIPLY_CAST_TO_LONG
949     return CACHE_FIXED_OVERHEAD + ClassSize.CONCURRENT_HASHMAP +
950         ((long)Math.ceil(maxSize*1.2/blockSize)
951             * ClassSize.CONCURRENT_HASHMAP_ENTRY) +
952         ((long)concurrency * ClassSize.CONCURRENT_HASHMAP_SEGMENT);
953   }
954 
955   @Override
956   public Iterator<CachedBlock> iterator() {
957     final Iterator<LruCachedBlock> iterator = map.values().iterator();
958 
959     return new Iterator<CachedBlock>() {
960       private final long now = System.nanoTime();
961 
962       @Override
963       public boolean hasNext() {
964         return iterator.hasNext();
965       }
966 
967       @Override
968       public CachedBlock next() {
969         final LruCachedBlock b = iterator.next();
970         return new CachedBlock() {
971           @Override
972           public String toString() {
973             return BlockCacheUtil.toString(this, now);
974           }
975 
976           @Override
977           public BlockPriority getBlockPriority() {
978             return b.getPriority();
979           }
980 
981           @Override
982           public BlockType getBlockType() {
983             return b.getBuffer().getBlockType();
984           }
985 
986           @Override
987           public long getOffset() {
988             return b.getCacheKey().getOffset();
989           }
990 
991           @Override
992           public long getSize() {
993             return b.getBuffer().heapSize();
994           }
995 
996           @Override
997           public long getCachedTime() {
998             return b.getCachedTime();
999           }
1000 
1001           @Override
1002           public String getFilename() {
1003             return b.getCacheKey().getHfileName();
1004           }
1005 
1006           @Override
1007           public int compareTo(CachedBlock other) {
1008             int diff = this.getFilename().compareTo(other.getFilename());
1009             if (diff != 0) return diff;
1010             diff = Long.compare(this.getOffset(), other.getOffset());
1011             if (diff != 0) return diff;
1012             if (other.getCachedTime() < 0 || this.getCachedTime() < 0) {
1013               throw new IllegalStateException("" + this.getCachedTime() + ", " +
1014                 other.getCachedTime());
1015             }
1016             return Long.compare(other.getCachedTime(), this.getCachedTime());
1017           }
1018 
1019           @Override
1020           public int hashCode() {
1021             return b.hashCode();
1022           }
1023 
1024           @Override
1025           public boolean equals(Object obj) {
1026             if (obj instanceof CachedBlock) {
1027               CachedBlock cb = (CachedBlock)obj;
1028               return compareTo(cb) == 0;
1029             } else {
1030               return false;
1031             }
1032           }
1033         };
1034       }
1035 
1036       @Override
1037       public void remove() {
1038         throw new UnsupportedOperationException();
1039       }
1040     };
1041   }
1042 
1043   // Simple calculators of sizes given factors and maxSize
1044 
1045   long acceptableSize() {
1046     return (long)Math.floor(this.maxSize * this.acceptableFactor);
1047   }
1048   private long minSize() {
1049     return (long)Math.floor(this.maxSize * this.minFactor);
1050   }
1051   private long singleSize() {
1052     return (long)Math.floor(this.maxSize * this.singleFactor * this.minFactor);
1053   }
1054   private long multiSize() {
1055     return (long)Math.floor(this.maxSize * this.multiFactor * this.minFactor);
1056   }
1057   private long memorySize() {
1058     return (long)Math.floor(this.maxSize * this.memoryFactor * this.minFactor);
1059   }
1060 
1061   public void shutdown() {
1062     if (victimHandler != null)
1063       victimHandler.shutdown();
1064     this.scheduleThreadPool.shutdown();
1065     for (int i = 0; i < 10; i++) {
1066       if (!this.scheduleThreadPool.isShutdown()) {
1067         try {
1068           Thread.sleep(10);
1069         } catch (InterruptedException e) {
1070           LOG.warn("Interrupted while sleeping");
1071           Thread.currentThread().interrupt();
1072           break;
1073         }
1074       }
1075     }
1076 
1077     if (!this.scheduleThreadPool.isShutdown()) {
1078       List<Runnable> runnables = this.scheduleThreadPool.shutdownNow();
1079       LOG.debug("Still running " + runnables);
1080     }
1081     this.evictionThread.shutdown();
1082   }
1083 
1084   /** Clears the cache. Used in tests. */
1085   @VisibleForTesting
1086   public void clearCache() {
1087     this.map.clear();
1088     this.elements.set(0);
1089   }
1090 
1091   /**
1092    * Used in testing. May be very inefficient.
1093    * @return the set of cached file names
1094    */
1095   @VisibleForTesting
1096   SortedSet<String> getCachedFileNamesForTest() {
1097     SortedSet<String> fileNames = new TreeSet<String>();
1098     for (BlockCacheKey cacheKey : map.keySet()) {
1099       fileNames.add(cacheKey.getHfileName());
1100     }
1101     return fileNames;
1102   }
1103 
1104   @VisibleForTesting
1105   Map<BlockType, Integer> getBlockTypeCountsForTest() {
1106     Map<BlockType, Integer> counts =
1107         new EnumMap<BlockType, Integer>(BlockType.class);
1108     for (LruCachedBlock cb : map.values()) {
1109       BlockType blockType = ((Cacheable)cb.getBuffer()).getBlockType();
1110       Integer count = counts.get(blockType);
1111       counts.put(blockType, (count == null ? 0 : count) + 1);
1112     }
1113     return counts;
1114   }
1115 
1116   @VisibleForTesting
1117   public Map<DataBlockEncoding, Integer> getEncodingCountsForTest() {
1118     Map<DataBlockEncoding, Integer> counts =
1119         new EnumMap<DataBlockEncoding, Integer>(DataBlockEncoding.class);
1120     for (LruCachedBlock block : map.values()) {
1121       DataBlockEncoding encoding =
1122               ((HFileBlock) block.getBuffer()).getDataBlockEncoding();
1123       Integer count = counts.get(encoding);
1124       counts.put(encoding, (count == null ? 0 : count) + 1);
1125     }
1126     return counts;
1127   }
1128 
1129   public void setVictimCache(BlockCache handler) {
1130     assert victimHandler == null;
1131     victimHandler = handler;
1132   }
1133 
1134   @VisibleForTesting
1135   Map<BlockCacheKey, LruCachedBlock> getMapForTests() {
1136     return map;
1137   }
1138 
1139   BlockCache getVictimHandler() {
1140     return this.victimHandler;
1141   }
1142 
1143   @Override
1144   public BlockCache[] getBlockCaches() {
1145     return null;
1146   }
1147 }