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