View Javadoc

1   /**
2    * Copyright The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing, software
15  
16   * distributed under the License is distributed on an "AS IS" BASIS,
17   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
18   * See the License for the specific language governing permissions and
19   * limitations under the License.
20   */
21  package org.apache.hadoop.hbase.io.hfile.bucket;
22  
23  import java.io.File;
24  import java.io.FileInputStream;
25  import java.io.FileNotFoundException;
26  import java.io.FileOutputStream;
27  import java.io.IOException;
28  import java.io.ObjectInputStream;
29  import java.io.ObjectOutputStream;
30  import java.io.Serializable;
31  import java.nio.ByteBuffer;
32  import java.util.ArrayList;
33  import java.util.Comparator;
34  import java.util.HashSet;
35  import java.util.Iterator;
36  import java.util.List;
37  import java.util.Map;
38  import java.util.NavigableSet;
39  import java.util.PriorityQueue;
40  import java.util.Set;
41  import java.util.concurrent.ArrayBlockingQueue;
42  import java.util.concurrent.BlockingQueue;
43  import java.util.concurrent.ConcurrentHashMap;
44  import java.util.concurrent.ConcurrentMap;
45  import java.util.concurrent.ConcurrentSkipListSet;
46  import java.util.concurrent.Executors;
47  import java.util.concurrent.ScheduledExecutorService;
48  import java.util.concurrent.TimeUnit;
49  import java.util.concurrent.atomic.AtomicLong;
50  import java.util.concurrent.locks.Lock;
51  import java.util.concurrent.locks.ReentrantLock;
52  import java.util.concurrent.locks.ReentrantReadWriteLock;
53  
54  import org.apache.commons.logging.Log;
55  import org.apache.commons.logging.LogFactory;
56  import org.apache.hadoop.hbase.classification.InterfaceAudience;
57  import org.apache.hadoop.hbase.io.HeapSize;
58  import org.apache.hadoop.hbase.io.hfile.BlockCache;
59  import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
60  import org.apache.hadoop.hbase.io.hfile.BlockCacheUtil;
61  import org.apache.hadoop.hbase.io.hfile.BlockPriority;
62  import org.apache.hadoop.hbase.io.hfile.BlockType;
63  import org.apache.hadoop.hbase.io.hfile.CacheStats;
64  import org.apache.hadoop.hbase.io.hfile.Cacheable;
65  import org.apache.hadoop.hbase.io.hfile.CacheableDeserializer;
66  import org.apache.hadoop.hbase.io.hfile.CacheableDeserializerIdManager;
67  import org.apache.hadoop.hbase.io.hfile.CachedBlock;
68  import org.apache.hadoop.hbase.io.hfile.HFileBlock;
69  import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
70  import org.apache.hadoop.hbase.util.HasThread;
71  import org.apache.hadoop.hbase.util.IdReadWriteLock;
72  import org.apache.hadoop.util.StringUtils;
73  
74  import com.google.common.annotations.VisibleForTesting;
75  import com.google.common.util.concurrent.ThreadFactoryBuilder;
76  
77  /**
78   * BucketCache uses {@link BucketAllocator} to allocate/free blocks, and uses
79   * BucketCache#ramCache and BucketCache#backingMap in order to
80   * determine if a given element is in the cache. The bucket cache can use on-heap or
81   * off-heap memory {@link ByteBufferIOEngine} or in a file {@link FileIOEngine} to
82   * store/read the block data.
83   *
84   * <p>Eviction is via a similar algorithm as used in
85   * {@link org.apache.hadoop.hbase.io.hfile.LruBlockCache}
86   *
87   * <p>BucketCache can be used as mainly a block cache (see
88   * {@link org.apache.hadoop.hbase.io.hfile.CombinedBlockCache}), combined with
89   * combined with LruBlockCache to decrease CMS GC and heap fragmentation.
90   *
91   * <p>It also can be used as a secondary cache (e.g. using a file on ssd/fusionio to store
92   * blocks) to enlarge cache space via
93   * {@link org.apache.hadoop.hbase.io.hfile.LruBlockCache#setVictimCache}
94   */
95  @InterfaceAudience.Private
96  public class BucketCache implements BlockCache, HeapSize {
97    private static final Log LOG = LogFactory.getLog(BucketCache.class);
98  
99    /** Priority buckets */
100   private static final float DEFAULT_SINGLE_FACTOR = 0.25f;
101   private static final float DEFAULT_MULTI_FACTOR = 0.50f;
102   private static final float DEFAULT_MEMORY_FACTOR = 0.25f;
103   private static final float DEFAULT_EXTRA_FREE_FACTOR = 0.10f;
104 
105   private static final float DEFAULT_ACCEPT_FACTOR = 0.95f;
106   private static final float DEFAULT_MIN_FACTOR = 0.85f;
107 
108   // Number of blocks to clear for each of the bucket size that is full
109   private static final int DEFAULT_FREE_ENTIRE_BLOCK_FACTOR = 2;
110 
111   /** Statistics thread */
112   private static final int statThreadPeriod = 5 * 60;
113 
114   final static int DEFAULT_WRITER_THREADS = 3;
115   final static int DEFAULT_WRITER_QUEUE_ITEMS = 64;
116 
117   // Store/read block data
118   final IOEngine ioEngine;
119 
120   // Store the block in this map before writing it to cache
121   @VisibleForTesting
122   final ConcurrentMap<BlockCacheKey, RAMQueueEntry> ramCache;
123   // In this map, store the block's meta data like offset, length
124   @VisibleForTesting
125   ConcurrentMap<BlockCacheKey, BucketEntry> backingMap;
126 
127   /**
128    * Flag if the cache is enabled or not... We shut it off if there are IO
129    * errors for some time, so that Bucket IO exceptions/errors don't bring down
130    * the HBase server.
131    */
132   private volatile boolean cacheEnabled;
133 
134   /**
135    * A list of writer queues.  We have a queue per {@link WriterThread} we have running.
136    * In other words, the work adding blocks to the BucketCache is divided up amongst the
137    * running WriterThreads.  Its done by taking hash of the cache key modulo queue count.
138    * WriterThread when it runs takes whatever has been recently added and 'drains' the entries
139    * to the BucketCache.  It then updates the ramCache and backingMap accordingly.
140    */
141   @VisibleForTesting
142   final ArrayList<BlockingQueue<RAMQueueEntry>> writerQueues =
143       new ArrayList<BlockingQueue<RAMQueueEntry>>();
144   @VisibleForTesting
145   final WriterThread[] writerThreads;
146 
147   /** Volatile boolean to track if free space is in process or not */
148   private volatile boolean freeInProgress = false;
149   private final Lock freeSpaceLock = new ReentrantLock();
150 
151   private UniqueIndexMap<Integer> deserialiserMap = new UniqueIndexMap<Integer>();
152 
153   private final AtomicLong realCacheSize = new AtomicLong(0);
154   private final AtomicLong heapSize = new AtomicLong(0);
155   /** Current number of cached elements */
156   private final AtomicLong blockNumber = new AtomicLong(0);
157 
158   /** Cache access count (sequential ID) */
159   private final AtomicLong accessCount = new AtomicLong(0);
160 
161   private static final int DEFAULT_CACHE_WAIT_TIME = 50;
162   // Used in test now. If the flag is false and the cache speed is very fast,
163   // bucket cache will skip some blocks when caching. If the flag is true, we
164   // will wait blocks flushed to IOEngine for some time when caching
165   boolean wait_when_cache = false;
166 
167   private final BucketCacheStats cacheStats = new BucketCacheStats();
168 
169   private final String persistencePath;
170   private final long cacheCapacity;
171   /** Approximate block size */
172   private final long blockSize;
173 
174   /** Duration of IO errors tolerated before we disable cache, 1 min as default */
175   private final int ioErrorsTolerationDuration;
176   // 1 min
177   public static final int DEFAULT_ERROR_TOLERATION_DURATION = 60 * 1000;
178 
179   // Start time of first IO error when reading or writing IO Engine, it will be
180   // reset after a successful read/write.
181   private volatile long ioErrorStartTime = -1;
182 
183   /**
184    * A ReentrantReadWriteLock to lock on a particular block identified by offset.
185    * The purpose of this is to avoid freeing the block which is being read.
186    */
187   @VisibleForTesting
188   final IdReadWriteLock offsetLock = new IdReadWriteLock();
189 
190   private final NavigableSet<BlockCacheKey> blocksByHFile =
191       new ConcurrentSkipListSet<BlockCacheKey>(new Comparator<BlockCacheKey>() {
192         @Override
193         public int compare(BlockCacheKey a, BlockCacheKey b) {
194           int nameComparison = a.getHfileName().compareTo(b.getHfileName());
195           if (nameComparison != 0) {
196             return nameComparison;
197           }
198 
199           if (a.getOffset() == b.getOffset()) {
200             return 0;
201           } else if (a.getOffset() < b.getOffset()) {
202             return -1;
203           }
204           return 1;
205         }
206       });
207 
208   /** Statistics thread schedule pool (for heavy debugging, could remove) */
209   private final ScheduledExecutorService scheduleThreadPool = Executors.newScheduledThreadPool(1,
210     new ThreadFactoryBuilder().setNameFormat("BucketCacheStatsExecutor").setDaemon(true).build());
211 
212   // Allocate or free space for the block
213   private BucketAllocator bucketAllocator;
214 
215   public BucketCache(String ioEngineName, long capacity, int blockSize, int[] bucketSizes,
216       int writerThreadNum, int writerQLen, String persistencePath) throws FileNotFoundException,
217       IOException {
218     this(ioEngineName, capacity, blockSize, bucketSizes, writerThreadNum, writerQLen,
219       persistencePath, DEFAULT_ERROR_TOLERATION_DURATION);
220   }
221 
222   public BucketCache(String ioEngineName, long capacity, int blockSize, int[] bucketSizes,
223       int writerThreadNum, int writerQLen, String persistencePath, int ioErrorsTolerationDuration)
224       throws FileNotFoundException, IOException {
225     this.ioEngine = getIOEngineFromName(ioEngineName, capacity);
226     this.writerThreads = new WriterThread[writerThreadNum];
227     long blockNumCapacity = capacity / blockSize;
228     if (blockNumCapacity >= Integer.MAX_VALUE) {
229       // Enough for about 32TB of cache!
230       throw new IllegalArgumentException("Cache capacity is too large, only support 32TB now");
231     }
232 
233     this.cacheCapacity = capacity;
234     this.persistencePath = persistencePath;
235     this.blockSize = blockSize;
236     this.ioErrorsTolerationDuration = ioErrorsTolerationDuration;
237 
238     bucketAllocator = new BucketAllocator(capacity, bucketSizes);
239     for (int i = 0; i < writerThreads.length; ++i) {
240       writerQueues.add(new ArrayBlockingQueue<RAMQueueEntry>(writerQLen));
241     }
242 
243     assert writerQueues.size() == writerThreads.length;
244     this.ramCache = new ConcurrentHashMap<BlockCacheKey, RAMQueueEntry>();
245 
246     this.backingMap = new ConcurrentHashMap<BlockCacheKey, BucketEntry>((int) blockNumCapacity);
247 
248     if (ioEngine.isPersistent() && persistencePath != null) {
249       try {
250         retrieveFromFile(bucketSizes);
251       } catch (IOException ioex) {
252         LOG.error("Can't restore from file because of", ioex);
253       } catch (ClassNotFoundException cnfe) {
254         LOG.error("Can't restore from file in rebuild because can't deserialise",cnfe);
255         throw new RuntimeException(cnfe);
256       }
257     }
258     final String threadName = Thread.currentThread().getName();
259     this.cacheEnabled = true;
260     for (int i = 0; i < writerThreads.length; ++i) {
261       writerThreads[i] = new WriterThread(writerQueues.get(i));
262       writerThreads[i].setName(threadName + "-BucketCacheWriter-" + i);
263       writerThreads[i].setDaemon(true);
264     }
265     startWriterThreads();
266 
267     // Run the statistics thread periodically to print the cache statistics log
268     // TODO: Add means of turning this off.  Bit obnoxious running thread just to make a log
269     // every five minutes.
270     this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
271         statThreadPeriod, statThreadPeriod, TimeUnit.SECONDS);
272     LOG.info("Started bucket cache; ioengine=" + ioEngineName +
273         ", capacity=" + StringUtils.byteDesc(capacity) +
274       ", blockSize=" + StringUtils.byteDesc(blockSize) + ", writerThreadNum=" +
275         writerThreadNum + ", writerQLen=" + writerQLen + ", persistencePath=" +
276       persistencePath + ", bucketAllocator=" + this.bucketAllocator.getClass().getName());
277   }
278 
279   /**
280    * Called by the constructor to start the writer threads. Used by tests that need to override
281    * starting the threads.
282    */
283   @VisibleForTesting
284   protected void startWriterThreads() {
285     for (WriterThread thread : writerThreads) {
286       thread.start();
287     }
288   }
289 
290   @VisibleForTesting
291   boolean isCacheEnabled() {
292     return this.cacheEnabled;
293   }
294 
295   public long getMaxSize() {
296     return this.cacheCapacity;
297   }
298 
299   public String getIoEngine() {
300     return ioEngine.toString();
301   }
302 
303   /**
304    * Get the IOEngine from the IO engine name
305    * @param ioEngineName
306    * @param capacity
307    * @return the IOEngine
308    * @throws IOException
309    */
310   private IOEngine getIOEngineFromName(String ioEngineName, long capacity)
311       throws IOException {
312     if (ioEngineName.startsWith("file:") || ioEngineName.startsWith("files:")) {
313       // In order to make the usage simple, we only need the prefix 'files:' in
314       // document whether one or multiple file(s), but also support 'file:' for
315       // the compatibility
316       String[] filePaths =
317           ioEngineName.substring(ioEngineName.indexOf(":") + 1).split(FileIOEngine.FILE_DELIMITER);
318       return new FileIOEngine(capacity, filePaths);
319     } else if (ioEngineName.startsWith("offheap"))
320       return new ByteBufferIOEngine(capacity, true);
321     else if (ioEngineName.startsWith("heap"))
322       return new ByteBufferIOEngine(capacity, false);
323     else
324       throw new IllegalArgumentException(
325           "Don't understand io engine name for cache - prefix with file:, heap or offheap");
326   }
327 
328   /**
329    * Cache the block with the specified name and buffer.
330    * @param cacheKey block's cache key
331    * @param buf block buffer
332    */
333   @Override
334   public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
335     cacheBlock(cacheKey, buf, false, false);
336   }
337 
338   /**
339    * Cache the block with the specified name and buffer.
340    * @param cacheKey block's cache key
341    * @param cachedItem block buffer
342    * @param inMemory if block is in-memory
343    * @param cacheDataInL1
344    */
345   @Override
346   public void cacheBlock(BlockCacheKey cacheKey, Cacheable cachedItem, boolean inMemory,
347       final boolean cacheDataInL1) {
348     cacheBlockWithWait(cacheKey, cachedItem, inMemory, wait_when_cache);
349   }
350 
351   /**
352    * Cache the block to ramCache
353    * @param cacheKey block's cache key
354    * @param cachedItem block buffer
355    * @param inMemory if block is in-memory
356    * @param wait if true, blocking wait when queue is full
357    */
358   public void cacheBlockWithWait(BlockCacheKey cacheKey, Cacheable cachedItem, boolean inMemory,
359       boolean wait) {
360     if (LOG.isTraceEnabled()) LOG.trace("Caching key=" + cacheKey + ", item=" + cachedItem);
361     if (!cacheEnabled) {
362       return;
363     }
364 
365     if (backingMap.containsKey(cacheKey)) {
366       return;
367     }
368 
369     /*
370      * Stuff the entry into the RAM cache so it can get drained to the persistent store
371      */
372     RAMQueueEntry re =
373         new RAMQueueEntry(cacheKey, cachedItem, accessCount.incrementAndGet(), inMemory);
374     if (ramCache.putIfAbsent(cacheKey, re) != null) {
375       return;
376     }
377     int queueNum = (cacheKey.hashCode() & 0x7FFFFFFF) % writerQueues.size();
378     BlockingQueue<RAMQueueEntry> bq = writerQueues.get(queueNum);
379     boolean successfulAddition = false;
380     if (wait) {
381       try {
382         successfulAddition = bq.offer(re, DEFAULT_CACHE_WAIT_TIME, TimeUnit.MILLISECONDS);
383       } catch (InterruptedException e) {
384         Thread.currentThread().interrupt();
385       }
386     } else {
387       successfulAddition = bq.offer(re);
388     }
389     if (!successfulAddition) {
390       ramCache.remove(cacheKey);
391       cacheStats.failInsert();
392     } else {
393       this.blockNumber.incrementAndGet();
394       this.heapSize.addAndGet(cachedItem.heapSize());
395       blocksByHFile.add(cacheKey);
396     }
397   }
398 
399   /**
400    * Get the buffer of the block with the specified key.
401    * @param key block's cache key
402    * @param caching true if the caller caches blocks on cache misses
403    * @param repeat Whether this is a repeat lookup for the same block
404    * @param updateCacheMetrics Whether we should update cache metrics or not
405    * @return buffer of specified cache key, or null if not in cache
406    */
407   @Override
408   public Cacheable getBlock(BlockCacheKey key, boolean caching, boolean repeat,
409       boolean updateCacheMetrics) {
410     if (!cacheEnabled) {
411       return null;
412     }
413     RAMQueueEntry re = ramCache.get(key);
414     if (re != null) {
415       if (updateCacheMetrics) {
416         cacheStats.hit(caching, key.isPrimary(), key.getBlockType());
417       }
418       re.access(accessCount.incrementAndGet());
419       return re.getData();
420     }
421     BucketEntry bucketEntry = backingMap.get(key);
422     if (bucketEntry != null) {
423       long start = System.nanoTime();
424       ReentrantReadWriteLock lock = offsetLock.getLock(bucketEntry.offset());
425       try {
426         lock.readLock().lock();
427         // We can not read here even if backingMap does contain the given key because its offset
428         // maybe changed. If we lock BlockCacheKey instead of offset, then we can only check
429         // existence here.
430         if (bucketEntry.equals(backingMap.get(key))) {
431           int len = bucketEntry.getLength();
432           if (LOG.isTraceEnabled()) {
433             LOG.trace("Read offset=" + bucketEntry.offset() + ", len=" + len);
434           }
435           ByteBuffer bb = ByteBuffer.allocate(len);
436           int lenRead = ioEngine.read(bb, bucketEntry.offset());
437           if (lenRead != len) {
438             throw new RuntimeException("Only " + lenRead + " bytes read, " + len + " expected");
439           }
440           CacheableDeserializer<Cacheable> deserializer =
441             bucketEntry.deserializerReference(this.deserialiserMap);
442           Cacheable cachedBlock = deserializer.deserialize(bb, true);
443           long timeTaken = System.nanoTime() - start;
444           if (updateCacheMetrics) {
445             cacheStats.hit(caching, key.isPrimary(), key.getBlockType());
446             cacheStats.ioHit(timeTaken);
447           }
448           bucketEntry.access(accessCount.incrementAndGet());
449           if (this.ioErrorStartTime > 0) {
450             ioErrorStartTime = -1;
451           }
452           return cachedBlock;
453         }
454       } catch (IOException ioex) {
455         LOG.error("Failed reading block " + key + " from bucket cache", ioex);
456         checkIOErrorIsTolerated();
457       } finally {
458         lock.readLock().unlock();
459       }
460     }
461     if (!repeat && updateCacheMetrics) {
462       cacheStats.miss(caching, key.isPrimary(), key.getBlockType());
463     }
464     return null;
465   }
466 
467   @VisibleForTesting
468   void blockEvicted(BlockCacheKey cacheKey, BucketEntry bucketEntry, boolean decrementBlockNumber) {
469     bucketAllocator.freeBlock(bucketEntry.offset());
470     realCacheSize.addAndGet(-1 * bucketEntry.getLength());
471     blocksByHFile.remove(cacheKey);
472     if (decrementBlockNumber) {
473       this.blockNumber.decrementAndGet();
474     }
475   }
476 
477   @Override
478   public boolean evictBlock(BlockCacheKey cacheKey) {
479     if (!cacheEnabled) {
480       return false;
481     }
482     RAMQueueEntry removedBlock = ramCache.remove(cacheKey);
483     if (removedBlock != null) {
484       this.blockNumber.decrementAndGet();
485       this.heapSize.addAndGet(-1 * removedBlock.getData().heapSize());
486     }
487     BucketEntry bucketEntry = backingMap.get(cacheKey);
488     if (bucketEntry == null) {
489       if (removedBlock != null) {
490         cacheStats.evicted(0, cacheKey.isPrimary());
491         return true;
492       } else {
493         return false;
494       }
495     }
496     ReentrantReadWriteLock lock = offsetLock.getLock(bucketEntry.offset());
497     try {
498       lock.writeLock().lock();
499       if (backingMap.remove(cacheKey, bucketEntry)) {
500         blockEvicted(cacheKey, bucketEntry, removedBlock == null);
501       } else {
502         return false;
503       }
504     } finally {
505       lock.writeLock().unlock();
506     }
507     cacheStats.evicted(bucketEntry.getCachedTime(), cacheKey.isPrimary());
508     return true;
509   }
510 
511   /*
512    * Statistics thread.  Periodically output cache statistics to the log.
513    */
514   private static class StatisticsThread extends Thread {
515     private final BucketCache bucketCache;
516 
517     public StatisticsThread(BucketCache bucketCache) {
518       super("BucketCacheStatsThread");
519       setDaemon(true);
520       this.bucketCache = bucketCache;
521     }
522 
523     @Override
524     public void run() {
525       bucketCache.logStats();
526     }
527   }
528 
529   public void logStats() {
530     long totalSize = bucketAllocator.getTotalSize();
531     long usedSize = bucketAllocator.getUsedSize();
532     long freeSize = totalSize - usedSize;
533     long cacheSize = getRealCacheSize();
534     LOG.info("failedBlockAdditions=" + cacheStats.getFailedInserts() + ", " +
535         "totalSize=" + StringUtils.byteDesc(totalSize) + ", " +
536         "freeSize=" + StringUtils.byteDesc(freeSize) + ", " +
537         "usedSize=" + StringUtils.byteDesc(usedSize) +", " +
538         "cacheSize=" + StringUtils.byteDesc(cacheSize) +", " +
539         "accesses=" + cacheStats.getRequestCount() + ", " +
540         "hits=" + cacheStats.getHitCount() + ", " +
541         "IOhitsPerSecond=" + cacheStats.getIOHitsPerSecond() + ", " +
542         "IOTimePerHit=" + String.format("%.2f", cacheStats.getIOTimePerHit())+ ", " +
543         "hitRatio=" + (cacheStats.getHitCount() == 0 ? "0," :
544           (StringUtils.formatPercent(cacheStats.getHitRatio(), 2)+ ", ")) +
545         "cachingAccesses=" + cacheStats.getRequestCachingCount() + ", " +
546         "cachingHits=" + cacheStats.getHitCachingCount() + ", " +
547         "cachingHitsRatio=" +(cacheStats.getHitCachingCount() == 0 ? "0," :
548           (StringUtils.formatPercent(cacheStats.getHitCachingRatio(), 2)+ ", ")) +
549         "evictions=" + cacheStats.getEvictionCount() + ", " +
550         "evicted=" + cacheStats.getEvictedCount() + ", " +
551         "evictedPerRun=" + cacheStats.evictedPerEviction());
552     cacheStats.reset();
553   }
554 
555   public long getRealCacheSize() {
556     return this.realCacheSize.get();
557   }
558 
559   private long acceptableSize() {
560     return (long) Math.floor(bucketAllocator.getTotalSize() * DEFAULT_ACCEPT_FACTOR);
561   }
562 
563   private long singleSize() {
564     return (long) Math.floor(bucketAllocator.getTotalSize()
565         * DEFAULT_SINGLE_FACTOR * DEFAULT_MIN_FACTOR);
566   }
567 
568   private long multiSize() {
569     return (long) Math.floor(bucketAllocator.getTotalSize() * DEFAULT_MULTI_FACTOR
570         * DEFAULT_MIN_FACTOR);
571   }
572 
573   private long memorySize() {
574     return (long) Math.floor(bucketAllocator.getTotalSize() * DEFAULT_MEMORY_FACTOR
575         * DEFAULT_MIN_FACTOR);
576   }
577 
578   /**
579    * Return the count of bucketSizeinfos still needf ree space
580    */
581   private int bucketSizesAboveThresholdCount(float minFactor) {
582     BucketAllocator.IndexStatistics[] stats = bucketAllocator.getIndexStatistics();
583     int fullCount = 0;
584     for (int i = 0; i < stats.length; i++) {
585       long freeGoal = (long) Math.floor(stats[i].totalCount() * (1 - minFactor));
586       freeGoal = Math.max(freeGoal, 1);
587       if (stats[i].freeCount() < freeGoal) {
588         fullCount++;
589       }
590     }
591     return fullCount;
592   }
593 
594   /**
595    * This method will find the buckets that are minimally occupied
596    * and are not reference counted and will free them completely
597    * without any constraint on the access times of the elements,
598    * and as a process will completely free at most the number of buckets
599    * passed, sometimes it might not due to changing refCounts
600    *
601    * @param completelyFreeBucketsNeeded number of buckets to free
602    **/
603   private void freeEntireBuckets(int completelyFreeBucketsNeeded) {
604     if (completelyFreeBucketsNeeded != 0) {
605       // First we will build a set where the offsets are reference counted, usually
606       // this set is small around O(Handler Count) unless something else is wrong
607       Set<Integer> inUseBuckets = new HashSet<Integer>();
608       for (BucketEntry entry : backingMap.values()) {
609         inUseBuckets.add(bucketAllocator.getBucketIndex(entry.offset()));
610       }
611 
612       Set<Integer> candidateBuckets = bucketAllocator.getLeastFilledBuckets(
613           inUseBuckets, completelyFreeBucketsNeeded);
614       for (Map.Entry<BlockCacheKey, BucketEntry> entry : backingMap.entrySet()) {
615         if (candidateBuckets.contains(bucketAllocator
616             .getBucketIndex(entry.getValue().offset()))) {
617           evictBlock(entry.getKey());
618         }
619       }
620     }
621   }
622 
623   /**
624    * Free the space if the used size reaches acceptableSize() or one size block
625    * couldn't be allocated. When freeing the space, we use the LRU algorithm and
626    * ensure there must be some blocks evicted
627    * @param why Why we are being called
628    */
629   private void freeSpace(final String why) {
630     // Ensure only one freeSpace progress at a time
631     if (!freeSpaceLock.tryLock()) {
632       return;
633     }
634     try {
635       freeInProgress = true;
636       long bytesToFreeWithoutExtra = 0;
637       // Calculate free byte for each bucketSizeinfo
638       StringBuffer msgBuffer = LOG.isDebugEnabled()? new StringBuffer(): null;
639       BucketAllocator.IndexStatistics[] stats = bucketAllocator.getIndexStatistics();
640       long[] bytesToFreeForBucket = new long[stats.length];
641       for (int i = 0; i < stats.length; i++) {
642         bytesToFreeForBucket[i] = 0;
643         long freeGoal = (long) Math.floor(stats[i].totalCount() * (1 - DEFAULT_MIN_FACTOR));
644         freeGoal = Math.max(freeGoal, 1);
645         if (stats[i].freeCount() < freeGoal) {
646           bytesToFreeForBucket[i] = stats[i].itemSize() * (freeGoal - stats[i].freeCount());
647           bytesToFreeWithoutExtra += bytesToFreeForBucket[i];
648           if (msgBuffer != null) {
649             msgBuffer.append("Free for bucketSize(" + stats[i].itemSize() + ")="
650               + StringUtils.byteDesc(bytesToFreeForBucket[i]) + ", ");
651           }
652         }
653       }
654       if (msgBuffer != null) {
655         msgBuffer.append("Free for total=" + StringUtils.byteDesc(bytesToFreeWithoutExtra) + ", ");
656       }
657 
658       if (bytesToFreeWithoutExtra <= 0) {
659         return;
660       }
661       long currentSize = bucketAllocator.getUsedSize();
662       long totalSize = bucketAllocator.getTotalSize();
663       if (LOG.isDebugEnabled() && msgBuffer != null) {
664         LOG.debug("Free started because \"" + why + "\"; " + msgBuffer.toString() +
665           " of current used=" + StringUtils.byteDesc(currentSize) + ", actual cacheSize=" +
666           StringUtils.byteDesc(realCacheSize.get()) + ", total=" + StringUtils.byteDesc(totalSize));
667       }
668 
669       long bytesToFreeWithExtra = (long) Math.floor(bytesToFreeWithoutExtra
670           * (1 + DEFAULT_EXTRA_FREE_FACTOR));
671 
672       // Instantiate priority buckets
673       BucketEntryGroup bucketSingle = new BucketEntryGroup(bytesToFreeWithExtra,
674           blockSize, singleSize());
675       BucketEntryGroup bucketMulti = new BucketEntryGroup(bytesToFreeWithExtra,
676           blockSize, multiSize());
677       BucketEntryGroup bucketMemory = new BucketEntryGroup(bytesToFreeWithExtra,
678           blockSize, memorySize());
679 
680       // Scan entire map putting bucket entry into appropriate bucket entry
681       // group
682       for (Map.Entry<BlockCacheKey, BucketEntry> bucketEntryWithKey : backingMap.entrySet()) {
683         switch (bucketEntryWithKey.getValue().getPriority()) {
684           case SINGLE: {
685             bucketSingle.add(bucketEntryWithKey);
686             break;
687           }
688           case MULTI: {
689             bucketMulti.add(bucketEntryWithKey);
690             break;
691           }
692           case MEMORY: {
693             bucketMemory.add(bucketEntryWithKey);
694             break;
695           }
696         }
697       }
698 
699       PriorityQueue<BucketEntryGroup> bucketQueue = new PriorityQueue<BucketEntryGroup>(3);
700 
701       bucketQueue.add(bucketSingle);
702       bucketQueue.add(bucketMulti);
703       bucketQueue.add(bucketMemory);
704 
705       int remainingBuckets = 3;
706       long bytesFreed = 0;
707 
708       BucketEntryGroup bucketGroup;
709       while ((bucketGroup = bucketQueue.poll()) != null) {
710         long overflow = bucketGroup.overflow();
711         if (overflow > 0) {
712           long bucketBytesToFree = Math.min(overflow,
713               (bytesToFreeWithoutExtra - bytesFreed) / remainingBuckets);
714           bytesFreed += bucketGroup.free(bucketBytesToFree);
715         }
716         remainingBuckets--;
717       }
718 
719       // Check and free if there are buckets that still need freeing of space
720       if (bucketSizesAboveThresholdCount(DEFAULT_MIN_FACTOR) > 0) {
721         bucketQueue.clear();
722         remainingBuckets = 3;
723 
724         bucketQueue.add(bucketSingle);
725         bucketQueue.add(bucketMulti);
726         bucketQueue.add(bucketMemory);
727 
728         while ((bucketGroup = bucketQueue.poll()) != null) {
729           long bucketBytesToFree = (bytesToFreeWithExtra - bytesFreed) / remainingBuckets;
730           bytesFreed += bucketGroup.free(bucketBytesToFree);
731           remainingBuckets--;
732         }
733       }
734 
735       // Even after the above free we might still need freeing because of the
736       // De-fragmentation of the buckets (also called Slab Calcification problem), i.e
737       // there might be some buckets where the occupancy is very sparse and thus are not
738       // yielding the free for the other bucket sizes, the fix for this to evict some
739       // of the buckets, we do this by evicting the buckets that are least fulled
740       freeEntireBuckets(DEFAULT_FREE_ENTIRE_BLOCK_FACTOR *
741           bucketSizesAboveThresholdCount(1.0f));
742 
743       if (LOG.isDebugEnabled()) {
744         long single = bucketSingle.totalSize();
745         long multi = bucketMulti.totalSize();
746         long memory = bucketMemory.totalSize();
747         if (LOG.isDebugEnabled()) {
748           LOG.debug("Bucket cache free space completed; " + "freed="
749             + StringUtils.byteDesc(bytesFreed) + ", " + "total="
750             + StringUtils.byteDesc(totalSize) + ", " + "single="
751             + StringUtils.byteDesc(single) + ", " + "multi="
752             + StringUtils.byteDesc(multi) + ", " + "memory="
753             + StringUtils.byteDesc(memory));
754         }
755       }
756 
757     } catch (Throwable t) {
758       LOG.warn("Failed freeing space", t);
759     } finally {
760       cacheStats.evict();
761       freeInProgress = false;
762       freeSpaceLock.unlock();
763     }
764   }
765 
766   // This handles flushing the RAM cache to IOEngine.
767   @VisibleForTesting
768   class WriterThread extends HasThread {
769     private final BlockingQueue<RAMQueueEntry> inputQueue;
770     private volatile boolean writerEnabled = true;
771 
772     WriterThread(BlockingQueue<RAMQueueEntry> queue) {
773       super("BucketCacheWriterThread");
774       this.inputQueue = queue;
775     }
776 
777     // Used for test
778     @VisibleForTesting
779     void disableWriter() {
780       this.writerEnabled = false;
781     }
782 
783     public void run() {
784       List<RAMQueueEntry> entries = new ArrayList<RAMQueueEntry>();
785       try {
786         while (cacheEnabled && writerEnabled) {
787           try {
788             try {
789               // Blocks
790               entries = getRAMQueueEntries(inputQueue, entries);
791             } catch (InterruptedException ie) {
792               if (!cacheEnabled) break;
793             }
794             doDrain(entries);
795           } catch (Exception ioe) {
796             LOG.error("WriterThread encountered error", ioe);
797           }
798         }
799       } catch (Throwable t) {
800         LOG.warn("Failed doing drain", t);
801       }
802       LOG.info(this.getName() + " exiting, cacheEnabled=" + cacheEnabled);
803     }
804 
805     /**
806      * Flush the entries in ramCache to IOEngine and add bucket entry to backingMap.
807      * Process all that are passed in even if failure being sure to remove from ramCache else we'll
808      * never undo the references and we'll OOME.
809      * @param entries Presumes list passed in here will be processed by this invocation only. No
810      * interference expected.
811      * @throws InterruptedException
812      */
813     @VisibleForTesting
814     void doDrain(final List<RAMQueueEntry> entries) throws InterruptedException {
815       if (entries.isEmpty()) {
816         return;
817       }
818       // This method is a little hard to follow. We run through the passed in entries and for each
819       // successful add, we add a non-null BucketEntry to the below bucketEntries.  Later we must
820       // do cleanup making sure we've cleared ramCache of all entries regardless of whether we
821       // successfully added the item to the bucketcache; if we don't do the cleanup, we'll OOME by
822       // filling ramCache.  We do the clean up by again running through the passed in entries
823       // doing extra work when we find a non-null bucketEntries corresponding entry.
824       final int size = entries.size();
825       BucketEntry[] bucketEntries = new BucketEntry[size];
826       // Index updated inside loop if success or if we can't succeed. We retry if cache is full
827       // when we go to add an entry by going around the loop again without upping the index.
828       int index = 0;
829       while (cacheEnabled && index < size) {
830         RAMQueueEntry re = null;
831         try {
832           re = entries.get(index);
833           if (re == null) {
834             LOG.warn("Couldn't get entry or changed on us; who else is messing with it?");
835             index++;
836             continue;
837           }
838           BucketEntry bucketEntry =
839             re.writeToCache(ioEngine, bucketAllocator, deserialiserMap, realCacheSize);
840           // Successfully added.  Up index and add bucketEntry. Clear io exceptions.
841           bucketEntries[index] = bucketEntry;
842           if (ioErrorStartTime > 0) {
843             ioErrorStartTime = -1;
844           }
845           index++;
846         } catch (BucketAllocatorException fle) {
847           LOG.warn("Failed allocation for " + (re == null ? "" : re.getKey()) + "; " + fle);
848           // Presume can't add. Too big? Move index on. Entry will be cleared from ramCache below.
849           bucketEntries[index] = null;
850           index++;
851         } catch (CacheFullException cfe) {
852           // Cache full when we tried to add. Try freeing space and then retrying (don't up index)
853           if (!freeInProgress) {
854             freeSpace("Full!");
855           } else {
856             Thread.sleep(50);
857           }
858         } catch (IOException ioex) {
859           // Hopefully transient. Retry. checkIOErrorIsTolerated disables cache if problem.
860           LOG.error("Failed writing to bucket cache", ioex);
861           checkIOErrorIsTolerated();
862         }
863       }
864 
865       // Make sure data pages are written on media before we update maps.
866       try {
867         ioEngine.sync();
868       } catch (IOException ioex) {
869         LOG.error("Failed syncing IO engine", ioex);
870         checkIOErrorIsTolerated();
871         // Since we failed sync, free the blocks in bucket allocator
872         for (int i = 0; i < entries.size(); ++i) {
873           if (bucketEntries[i] != null) {
874             bucketAllocator.freeBlock(bucketEntries[i].offset());
875             bucketEntries[i] = null;
876           }
877         }
878       }
879 
880       // Now add to backingMap if successfully added to bucket cache.  Remove from ramCache if
881       // success or error.
882       for (int i = 0; i < size; ++i) {
883         BlockCacheKey key = entries.get(i).getKey();
884         // Only add if non-null entry.
885         if (bucketEntries[i] != null) {
886           backingMap.put(key, bucketEntries[i]);
887         }
888         // Always remove from ramCache even if we failed adding it to the block cache above.
889         RAMQueueEntry ramCacheEntry = ramCache.remove(key);
890         if (ramCacheEntry != null) {
891           heapSize.addAndGet(-1 * entries.get(i).getData().heapSize());
892         } else if (bucketEntries[i] != null){
893           // Block should have already been evicted. Remove it and free space.
894           ReentrantReadWriteLock lock = offsetLock.getLock(bucketEntries[i].offset());
895           try {
896             lock.writeLock().lock();
897             if (backingMap.remove(key, bucketEntries[i])) {
898               blockEvicted(key, bucketEntries[i], false);
899             }
900           } finally {
901             lock.writeLock().unlock();
902           }
903         }
904       }
905 
906       long used = bucketAllocator.getUsedSize();
907       if (used > acceptableSize()) {
908         freeSpace("Used=" + used + " > acceptable=" + acceptableSize());
909       }
910       return;
911     }
912   }
913 
914   /**
915    * Blocks until elements available in <code>q</code> then tries to grab as many as possible
916    * before returning.
917    * @param recepticle Where to stash the elements taken from queue. We clear before we use it
918    * just in case.
919    * @param q The queue to take from.
920    * @return <code>receptical laden with elements taken from the queue or empty if none found.
921    */
922   @VisibleForTesting
923   static List<RAMQueueEntry> getRAMQueueEntries(final BlockingQueue<RAMQueueEntry> q,
924       final List<RAMQueueEntry> receptical)
925   throws InterruptedException {
926     // Clear sets all entries to null and sets size to 0. We retain allocations. Presume it
927     // ok even if list grew to accommodate thousands.
928     receptical.clear();
929     receptical.add(q.take());
930     q.drainTo(receptical);
931     return receptical;
932   }
933 
934   private void persistToFile() throws IOException {
935     assert !cacheEnabled;
936     FileOutputStream fos = null;
937     ObjectOutputStream oos = null;
938     try {
939       if (!ioEngine.isPersistent()) {
940         throw new IOException("Attempt to persist non-persistent cache mappings!");
941       }
942       fos = new FileOutputStream(persistencePath, false);
943       oos = new ObjectOutputStream(fos);
944       oos.writeLong(cacheCapacity);
945       oos.writeUTF(ioEngine.getClass().getName());
946       oos.writeUTF(backingMap.getClass().getName());
947       oos.writeObject(deserialiserMap);
948       oos.writeObject(backingMap);
949     } finally {
950       if (oos != null) oos.close();
951       if (fos != null) fos.close();
952     }
953   }
954 
955   @SuppressWarnings("unchecked")
956   private void retrieveFromFile(int[] bucketSizes) throws IOException, BucketAllocatorException,
957       ClassNotFoundException {
958     File persistenceFile = new File(persistencePath);
959     if (!persistenceFile.exists()) {
960       return;
961     }
962     assert !cacheEnabled;
963     FileInputStream fis = null;
964     ObjectInputStream ois = null;
965     try {
966       if (!ioEngine.isPersistent())
967         throw new IOException(
968             "Attempt to restore non-persistent cache mappings!");
969       fis = new FileInputStream(persistencePath);
970       ois = new ObjectInputStream(fis);
971       long capacitySize = ois.readLong();
972       if (capacitySize != cacheCapacity)
973         throw new IOException("Mismatched cache capacity:"
974             + StringUtils.byteDesc(capacitySize) + ", expected: "
975             + StringUtils.byteDesc(cacheCapacity));
976       String ioclass = ois.readUTF();
977       String mapclass = ois.readUTF();
978       if (!ioEngine.getClass().getName().equals(ioclass))
979         throw new IOException("Class name for IO engine mismatch: " + ioclass
980             + ", expected:" + ioEngine.getClass().getName());
981       if (!backingMap.getClass().getName().equals(mapclass))
982         throw new IOException("Class name for cache map mismatch: " + mapclass
983             + ", expected:" + backingMap.getClass().getName());
984       UniqueIndexMap<Integer> deserMap = (UniqueIndexMap<Integer>) ois
985           .readObject();
986       ConcurrentHashMap<BlockCacheKey, BucketEntry> backingMapFromFile =
987           (ConcurrentHashMap<BlockCacheKey, BucketEntry>) ois.readObject();
988       BucketAllocator allocator = new BucketAllocator(cacheCapacity, bucketSizes,
989         backingMapFromFile, realCacheSize);
990       bucketAllocator = allocator;
991       deserialiserMap = deserMap;
992       backingMap = backingMapFromFile;
993     } finally {
994       if (ois != null) ois.close();
995       if (fis != null) fis.close();
996       if (!persistenceFile.delete()) {
997         throw new IOException("Failed deleting persistence file "
998             + persistenceFile.getAbsolutePath());
999       }
1000     }
1001   }
1002 
1003   /**
1004    * Check whether we tolerate IO error this time. If the duration of IOEngine
1005    * throwing errors exceeds ioErrorsDurationTimeTolerated, we will disable the
1006    * cache
1007    */
1008   private void checkIOErrorIsTolerated() {
1009     long now = EnvironmentEdgeManager.currentTime();
1010     if (this.ioErrorStartTime > 0) {
1011       if (cacheEnabled && (now - ioErrorStartTime) > this.ioErrorsTolerationDuration) {
1012         LOG.error("IO errors duration time has exceeded " + ioErrorsTolerationDuration +
1013           "ms, disabling cache, please check your IOEngine");
1014         disableCache();
1015       }
1016     } else {
1017       this.ioErrorStartTime = now;
1018     }
1019   }
1020 
1021   /**
1022    * Used to shut down the cache -or- turn it off in the case of something broken.
1023    */
1024   private void disableCache() {
1025     if (!cacheEnabled) return;
1026     cacheEnabled = false;
1027     ioEngine.shutdown();
1028     this.scheduleThreadPool.shutdown();
1029     for (int i = 0; i < writerThreads.length; ++i) writerThreads[i].interrupt();
1030     this.ramCache.clear();
1031     if (!ioEngine.isPersistent() || persistencePath == null) {
1032       // If persistent ioengine and a path, we will serialize out the backingMap.
1033       this.backingMap.clear();
1034     }
1035   }
1036 
1037   private void join() throws InterruptedException {
1038     for (int i = 0; i < writerThreads.length; ++i)
1039       writerThreads[i].join();
1040   }
1041 
1042   @Override
1043   public void shutdown() {
1044     disableCache();
1045     LOG.info("Shutdown bucket cache: IO persistent=" + ioEngine.isPersistent()
1046         + "; path to write=" + persistencePath);
1047     if (ioEngine.isPersistent() && persistencePath != null) {
1048       try {
1049         join();
1050         persistToFile();
1051       } catch (IOException ex) {
1052         LOG.error("Unable to persist data on exit: " + ex.toString(), ex);
1053       } catch (InterruptedException e) {
1054         LOG.warn("Failed to persist data on exit", e);
1055       }
1056     }
1057   }
1058 
1059   @Override
1060   public CacheStats getStats() {
1061     return cacheStats;
1062   }
1063 
1064   public BucketAllocator getAllocator() {
1065     return this.bucketAllocator;
1066   }
1067 
1068   @Override
1069   public long heapSize() {
1070     return this.heapSize.get();
1071   }
1072 
1073   @Override
1074   public long size() {
1075     return this.realCacheSize.get();
1076   }
1077 
1078   @Override
1079   public long getFreeSize() {
1080     return this.bucketAllocator.getFreeSize();
1081   }
1082 
1083   @Override
1084   public long getBlockCount() {
1085     return this.blockNumber.get();
1086   }
1087 
1088   @Override
1089   public long getCurrentSize() {
1090     return this.bucketAllocator.getUsedSize();
1091   }
1092 
1093   /**
1094    * Evicts all blocks for a specific HFile.
1095    * <p>
1096    * This is used for evict-on-close to remove all blocks of a specific HFile.
1097    *
1098    * @return the number of blocks evicted
1099    */
1100   @Override
1101   public int evictBlocksByHfileName(String hfileName) {
1102     Set<BlockCacheKey> keySet = blocksByHFile.subSet(
1103         new BlockCacheKey(hfileName, Long.MIN_VALUE), true,
1104         new BlockCacheKey(hfileName, Long.MAX_VALUE), true);
1105 
1106     int numEvicted = 0;
1107     for (BlockCacheKey key : keySet) {
1108       if (evictBlock(key)) {
1109           ++numEvicted;
1110       }
1111     }
1112 
1113     return numEvicted;
1114   }
1115 
1116   /**
1117    * Item in cache. We expect this to be where most memory goes. Java uses 8
1118    * bytes just for object headers; after this, we want to use as little as
1119    * possible - so we only use 8 bytes, but in order to do so we end up messing
1120    * around with all this Java casting stuff. Offset stored as 5 bytes that make
1121    * up the long. Doubt we'll see devices this big for ages. Offsets are divided
1122    * by 256. So 5 bytes gives us 256TB or so.
1123    */
1124   static class BucketEntry implements Serializable {
1125     private static final long serialVersionUID = -6741504807982257534L;
1126 
1127     // access counter comparator, descending order
1128     static final Comparator<BucketEntry> COMPARATOR = new Comparator<BucketCache.BucketEntry>() {
1129 
1130       @Override
1131       public int compare(BucketEntry o1, BucketEntry o2) {
1132         return Long.compare(o2.accessCounter, o1.accessCounter);
1133       }
1134     };
1135 
1136     private int offsetBase;
1137     private int length;
1138     private byte offset1;
1139     byte deserialiserIndex;
1140     private volatile long accessCounter;
1141     private BlockPriority priority;
1142     /**
1143      * Time this block was cached.  Presumes we are created just before we are added to the cache.
1144      */
1145     private final long cachedTime = System.nanoTime();
1146 
1147     BucketEntry(long offset, int length, long accessCounter, boolean inMemory) {
1148       setOffset(offset);
1149       this.length = length;
1150       this.accessCounter = accessCounter;
1151       if (inMemory) {
1152         this.priority = BlockPriority.MEMORY;
1153       } else {
1154         this.priority = BlockPriority.SINGLE;
1155       }
1156     }
1157 
1158     long offset() { // Java has no unsigned numbers
1159       long o = ((long) offsetBase) & 0xFFFFFFFFL; //This needs the L cast otherwise it will be sign extended as a negative number.
1160       o += (((long) (offset1)) & 0xFF) << 32; //The 0xFF here does not need the L cast because it is treated as a positive int.
1161       return o << 8;
1162     }
1163 
1164     private void setOffset(long value) {
1165       assert (value & 0xFF) == 0;
1166       value >>= 8;
1167       offsetBase = (int) value;
1168       offset1 = (byte) (value >> 32);
1169     }
1170 
1171     public int getLength() {
1172       return length;
1173     }
1174 
1175     protected CacheableDeserializer<Cacheable> deserializerReference(
1176         UniqueIndexMap<Integer> deserialiserMap) {
1177       return CacheableDeserializerIdManager.getDeserializer(deserialiserMap
1178           .unmap(deserialiserIndex));
1179     }
1180 
1181     protected void setDeserialiserReference(
1182         CacheableDeserializer<Cacheable> deserializer,
1183         UniqueIndexMap<Integer> deserialiserMap) {
1184       this.deserialiserIndex = ((byte) deserialiserMap.map(deserializer
1185           .getDeserialiserIdentifier()));
1186     }
1187 
1188     /**
1189      * Block has been accessed. Update its local access counter.
1190      */
1191     public void access(long accessCounter) {
1192       this.accessCounter = accessCounter;
1193       if (this.priority == BlockPriority.SINGLE) {
1194         this.priority = BlockPriority.MULTI;
1195       }
1196     }
1197 
1198     public BlockPriority getPriority() {
1199       return this.priority;
1200     }
1201 
1202     public long getCachedTime() {
1203       return cachedTime;
1204     }
1205   }
1206 
1207   /**
1208    * Used to group bucket entries into priority buckets. There will be a
1209    * BucketEntryGroup for each priority (single, multi, memory). Once bucketed,
1210    * the eviction algorithm takes the appropriate number of elements out of each
1211    * according to configuration parameters and their relative sizes.
1212    */
1213   private class BucketEntryGroup implements Comparable<BucketEntryGroup> {
1214 
1215     private CachedEntryQueue queue;
1216     private long totalSize = 0;
1217     private long bucketSize;
1218 
1219     public BucketEntryGroup(long bytesToFree, long blockSize, long bucketSize) {
1220       this.bucketSize = bucketSize;
1221       queue = new CachedEntryQueue(bytesToFree, blockSize);
1222       totalSize = 0;
1223     }
1224 
1225     public void add(Map.Entry<BlockCacheKey, BucketEntry> block) {
1226       totalSize += block.getValue().getLength();
1227       queue.add(block);
1228     }
1229 
1230     public long free(long toFree) {
1231       Map.Entry<BlockCacheKey, BucketEntry> entry;
1232       long freedBytes = 0;
1233       while ((entry = queue.pollLast()) != null) {
1234         evictBlock(entry.getKey());
1235         freedBytes += entry.getValue().getLength();
1236         if (freedBytes >= toFree) {
1237           return freedBytes;
1238         }
1239       }
1240       return freedBytes;
1241     }
1242 
1243     public long overflow() {
1244       return totalSize - bucketSize;
1245     }
1246 
1247     public long totalSize() {
1248       return totalSize;
1249     }
1250 
1251     @Override
1252     public int compareTo(BucketEntryGroup that) {
1253       return Long.compare(this.overflow(), that.overflow());
1254     }
1255 
1256     @Override
1257     public boolean equals(Object that) {
1258       return this == that;
1259     }
1260 
1261   }
1262 
1263   /**
1264    * Block Entry stored in the memory with key,data and so on
1265    */
1266   @VisibleForTesting
1267   static class RAMQueueEntry {
1268     private BlockCacheKey key;
1269     private Cacheable data;
1270     private long accessCounter;
1271     private boolean inMemory;
1272 
1273     public RAMQueueEntry(BlockCacheKey bck, Cacheable data, long accessCounter,
1274         boolean inMemory) {
1275       this.key = bck;
1276       this.data = data;
1277       this.accessCounter = accessCounter;
1278       this.inMemory = inMemory;
1279     }
1280 
1281     public Cacheable getData() {
1282       return data;
1283     }
1284 
1285     public BlockCacheKey getKey() {
1286       return key;
1287     }
1288 
1289     public void access(long accessCounter) {
1290       this.accessCounter = accessCounter;
1291     }
1292 
1293     public BucketEntry writeToCache(final IOEngine ioEngine,
1294         final BucketAllocator bucketAllocator,
1295         final UniqueIndexMap<Integer> deserialiserMap,
1296         final AtomicLong realCacheSize) throws CacheFullException, IOException,
1297         BucketAllocatorException {
1298       int len = data.getSerializedLength();
1299       // This cacheable thing can't be serialized
1300       if (len == 0) return null;
1301       long offset = bucketAllocator.allocateBlock(len);
1302       BucketEntry bucketEntry = new BucketEntry(offset, len, accessCounter, inMemory);
1303       bucketEntry.setDeserialiserReference(data.getDeserializer(), deserialiserMap);
1304       try {
1305         if (data instanceof HFileBlock) {
1306           // If an instance of HFileBlock, save on some allocations.
1307           HFileBlock block = (HFileBlock)data;
1308           ByteBuffer sliceBuf = block.getBufferReadOnly();
1309           ByteBuffer metadata = block.getMetaData();
1310           if (LOG.isTraceEnabled()) {
1311             LOG.trace("Write offset=" + offset + ", len=" + len);
1312           }
1313           ioEngine.write(sliceBuf, offset);
1314           ioEngine.write(metadata, offset + len - metadata.limit());
1315         } else {
1316           ByteBuffer bb = ByteBuffer.allocate(len);
1317           data.serialize(bb);
1318           ioEngine.write(bb, offset);
1319         }
1320       } catch (IOException ioe) {
1321         // free it in bucket allocator
1322         bucketAllocator.freeBlock(offset);
1323         throw ioe;
1324       }
1325 
1326       realCacheSize.addAndGet(len);
1327       return bucketEntry;
1328     }
1329   }
1330 
1331   /**
1332    * Only used in test
1333    * @throws InterruptedException
1334    */
1335   void stopWriterThreads() throws InterruptedException {
1336     for (WriterThread writerThread : writerThreads) {
1337       writerThread.disableWriter();
1338       writerThread.interrupt();
1339       writerThread.join();
1340     }
1341   }
1342 
1343   @Override
1344   public Iterator<CachedBlock> iterator() {
1345     // Don't bother with ramcache since stuff is in here only a little while.
1346     final Iterator<Map.Entry<BlockCacheKey, BucketEntry>> i =
1347         this.backingMap.entrySet().iterator();
1348     return new Iterator<CachedBlock>() {
1349       private final long now = System.nanoTime();
1350 
1351       @Override
1352       public boolean hasNext() {
1353         return i.hasNext();
1354       }
1355 
1356       @Override
1357       public CachedBlock next() {
1358         final Map.Entry<BlockCacheKey, BucketEntry> e = i.next();
1359         return new CachedBlock() {
1360           @Override
1361           public String toString() {
1362             return BlockCacheUtil.toString(this, now);
1363           }
1364 
1365           @Override
1366           public BlockPriority getBlockPriority() {
1367             return e.getValue().getPriority();
1368           }
1369 
1370           @Override
1371           public BlockType getBlockType() {
1372             // Not held by BucketEntry.  Could add it if wanted on BucketEntry creation.
1373             return null;
1374           }
1375 
1376           @Override
1377           public long getOffset() {
1378             return e.getKey().getOffset();
1379           }
1380 
1381           @Override
1382           public long getSize() {
1383             return e.getValue().getLength();
1384           }
1385 
1386           @Override
1387           public long getCachedTime() {
1388             return e.getValue().getCachedTime();
1389           }
1390 
1391           @Override
1392           public String getFilename() {
1393             return e.getKey().getHfileName();
1394           }
1395 
1396           @Override
1397           public int compareTo(CachedBlock other) {
1398             int diff = this.getFilename().compareTo(other.getFilename());
1399             if (diff != 0) return diff;
1400 
1401             diff = Long.compare(this.getOffset(), other.getOffset());
1402             if (diff != 0) return diff;
1403             if (other.getCachedTime() < 0 || this.getCachedTime() < 0) {
1404               throw new IllegalStateException("" + this.getCachedTime() + ", " +
1405                 other.getCachedTime());
1406             }
1407             return Long.compare(other.getCachedTime(), this.getCachedTime());
1408           }
1409 
1410           @Override
1411           public int hashCode() {
1412             return e.getKey().hashCode();
1413           }
1414 
1415           @Override
1416           public boolean equals(Object obj) {
1417             if (obj instanceof CachedBlock) {
1418               CachedBlock cb = (CachedBlock)obj;
1419               return compareTo(cb) == 0;
1420             } else {
1421               return false;
1422             }
1423           }
1424         };
1425       }
1426 
1427       @Override
1428       public void remove() {
1429         throw new UnsupportedOperationException();
1430       }
1431     };
1432   }
1433 
1434   @Override
1435   public BlockCache[] getBlockCaches() {
1436     return null;
1437   }
1438 }