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