1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.io.hfile;
20
21 import java.lang.ref.WeakReference;
22 import java.nio.ByteBuffer;
23 import java.util.EnumMap;
24 import java.util.Iterator;
25 import java.util.List;
26 import java.util.Map;
27 import java.util.PriorityQueue;
28 import java.util.SortedSet;
29 import java.util.TreeSet;
30 import java.util.concurrent.ConcurrentHashMap;
31 import java.util.concurrent.Executors;
32 import java.util.concurrent.ScheduledExecutorService;
33 import java.util.concurrent.TimeUnit;
34 import java.util.concurrent.atomic.AtomicLong;
35 import java.util.concurrent.locks.ReentrantLock;
36
37 import com.google.common.base.Objects;
38 import org.apache.commons.logging.Log;
39 import org.apache.commons.logging.LogFactory;
40 import org.apache.hadoop.hbase.classification.InterfaceAudience;
41 import org.apache.hadoop.conf.Configuration;
42 import org.apache.hadoop.hbase.io.HeapSize;
43 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
44 import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache;
45 import org.apache.hadoop.hbase.util.Bytes;
46 import org.apache.hadoop.hbase.util.ClassSize;
47 import org.apache.hadoop.hbase.util.HasThread;
48 import org.apache.hadoop.util.StringUtils;
49 import org.codehaus.jackson.annotate.JsonIgnoreProperties;
50
51 import com.google.common.annotations.VisibleForTesting;
52 import com.google.common.util.concurrent.ThreadFactoryBuilder;
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98 @InterfaceAudience.Private
99 @JsonIgnoreProperties({"encodingCountsForTest"})
100 public class LruBlockCache implements ResizableBlockCache, HeapSize {
101
102 private static final Log LOG = LogFactory.getLog(LruBlockCache.class);
103
104
105
106
107
108 static final String LRU_MIN_FACTOR_CONFIG_NAME = "hbase.lru.blockcache.min.factor";
109
110
111
112
113 static final String LRU_ACCEPTABLE_FACTOR_CONFIG_NAME = "hbase.lru.blockcache.acceptable.factor";
114
115
116
117
118 static final String LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME = "hbase.lru.blockcache.hard.capacity.limit.factor";
119 static final String LRU_SINGLE_PERCENTAGE_CONFIG_NAME = "hbase.lru.blockcache.single.percentage";
120 static final String LRU_MULTI_PERCENTAGE_CONFIG_NAME = "hbase.lru.blockcache.multi.percentage";
121 static final String LRU_MEMORY_PERCENTAGE_CONFIG_NAME = "hbase.lru.blockcache.memory.percentage";
122
123
124
125
126
127
128 static final String LRU_IN_MEMORY_FORCE_MODE_CONFIG_NAME = "hbase.lru.rs.inmemoryforcemode";
129
130
131
132
133 static final float DEFAULT_LOAD_FACTOR = 0.75f;
134 static final int DEFAULT_CONCURRENCY_LEVEL = 16;
135
136
137 static final float DEFAULT_MIN_FACTOR = 0.95f;
138 static final float DEFAULT_ACCEPTABLE_FACTOR = 0.99f;
139
140
141 static final float DEFAULT_SINGLE_FACTOR = 0.25f;
142 static final float DEFAULT_MULTI_FACTOR = 0.50f;
143 static final float DEFAULT_MEMORY_FACTOR = 0.25f;
144
145
146 static final float DEFAULT_HARD_CAPACITY_LIMIT_FACTOR = 1.2f;
147
148 static final boolean DEFAULT_IN_MEMORY_FORCE_MODE = false;
149
150
151 static final int statThreadPeriod = 60 * 5;
152 private static final String LRU_MAX_BLOCK_SIZE = "hbase.lru.max.block.size";
153 private static final long DEFAULT_MAX_BLOCK_SIZE = 16L * 1024L * 1024L;
154
155
156 private final Map<BlockCacheKey,LruCachedBlock> map;
157
158
159 private final ReentrantLock evictionLock = new ReentrantLock(true);
160 private final long maxBlockSize;
161
162
163 private volatile boolean evictionInProgress = false;
164
165
166 private final EvictionThread evictionThread;
167
168
169 private final ScheduledExecutorService scheduleThreadPool = Executors.newScheduledThreadPool(1,
170 new ThreadFactoryBuilder().setNameFormat("LruBlockCacheStatsExecutor").setDaemon(true).build());
171
172
173 private final AtomicLong size;
174
175
176 private final AtomicLong elements;
177
178
179 private final AtomicLong count;
180
181
182 private float hardCapacityLimitFactor;
183
184
185 private final CacheStats stats;
186
187
188 private long maxSize;
189
190
191 private long blockSize;
192
193
194 private float acceptableFactor;
195
196
197 private float minFactor;
198
199
200 private float singleFactor;
201
202
203 private float multiFactor;
204
205
206 private float memoryFactor;
207
208
209 private long overhead;
210
211
212 private boolean forceInMemory;
213
214
215 private BlockCache victimHandler = null;
216
217
218
219
220
221
222
223
224
225
226 public LruBlockCache(long maxSize, long blockSize) {
227 this(maxSize, blockSize, true);
228 }
229
230
231
232
233 public LruBlockCache(long maxSize, long blockSize, boolean evictionThread) {
234 this(maxSize, blockSize, evictionThread,
235 (int)Math.ceil(1.2*maxSize/blockSize),
236 DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL,
237 DEFAULT_MIN_FACTOR, DEFAULT_ACCEPTABLE_FACTOR,
238 DEFAULT_SINGLE_FACTOR,
239 DEFAULT_MULTI_FACTOR,
240 DEFAULT_MEMORY_FACTOR,
241 DEFAULT_HARD_CAPACITY_LIMIT_FACTOR,
242 false,
243 DEFAULT_MAX_BLOCK_SIZE
244 );
245 }
246
247 public LruBlockCache(long maxSize, long blockSize, boolean evictionThread, Configuration conf) {
248 this(maxSize, blockSize, evictionThread,
249 (int)Math.ceil(1.2*maxSize/blockSize),
250 DEFAULT_LOAD_FACTOR,
251 DEFAULT_CONCURRENCY_LEVEL,
252 conf.getFloat(LRU_MIN_FACTOR_CONFIG_NAME, DEFAULT_MIN_FACTOR),
253 conf.getFloat(LRU_ACCEPTABLE_FACTOR_CONFIG_NAME, DEFAULT_ACCEPTABLE_FACTOR),
254 conf.getFloat(LRU_SINGLE_PERCENTAGE_CONFIG_NAME, DEFAULT_SINGLE_FACTOR),
255 conf.getFloat(LRU_MULTI_PERCENTAGE_CONFIG_NAME, DEFAULT_MULTI_FACTOR),
256 conf.getFloat(LRU_MEMORY_PERCENTAGE_CONFIG_NAME, DEFAULT_MEMORY_FACTOR),
257 conf.getFloat(LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME, DEFAULT_HARD_CAPACITY_LIMIT_FACTOR),
258 conf.getBoolean(LRU_IN_MEMORY_FORCE_MODE_CONFIG_NAME, DEFAULT_IN_MEMORY_FORCE_MODE),
259 conf.getLong(LRU_MAX_BLOCK_SIZE, DEFAULT_MAX_BLOCK_SIZE)
260 );
261 }
262
263 public LruBlockCache(long maxSize, long blockSize, Configuration conf) {
264 this(maxSize, blockSize, true, conf);
265 }
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281 public LruBlockCache(long maxSize, long blockSize, boolean evictionThread,
282 int mapInitialSize, float mapLoadFactor, int mapConcurrencyLevel,
283 float minFactor, float acceptableFactor, float singleFactor,
284 float multiFactor, float memoryFactor, float hardLimitFactor,
285 boolean forceInMemory, long maxBlockSize) {
286 this.maxBlockSize = maxBlockSize;
287 if(singleFactor + multiFactor + memoryFactor != 1 ||
288 singleFactor < 0 || multiFactor < 0 || memoryFactor < 0) {
289 throw new IllegalArgumentException("Single, multi, and memory factors " +
290 " should be non-negative and total 1.0");
291 }
292 if(minFactor >= acceptableFactor) {
293 throw new IllegalArgumentException("minFactor must be smaller than acceptableFactor");
294 }
295 if(minFactor >= 1.0f || acceptableFactor >= 1.0f) {
296 throw new IllegalArgumentException("all factors must be < 1");
297 }
298 this.maxSize = maxSize;
299 this.blockSize = blockSize;
300 this.forceInMemory = forceInMemory;
301 map = new ConcurrentHashMap<BlockCacheKey,LruCachedBlock>(mapInitialSize,
302 mapLoadFactor, mapConcurrencyLevel);
303 this.minFactor = minFactor;
304 this.acceptableFactor = acceptableFactor;
305 this.singleFactor = singleFactor;
306 this.multiFactor = multiFactor;
307 this.memoryFactor = memoryFactor;
308 this.stats = new CacheStats(this.getClass().getSimpleName());
309 this.count = new AtomicLong(0);
310 this.elements = new AtomicLong(0);
311 this.overhead = calculateOverhead(maxSize, blockSize, mapConcurrencyLevel);
312 this.size = new AtomicLong(this.overhead);
313 this.hardCapacityLimitFactor = hardLimitFactor;
314 if(evictionThread) {
315 this.evictionThread = new EvictionThread(this);
316 this.evictionThread.start();
317 } else {
318 this.evictionThread = null;
319 }
320
321
322 this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
323 statThreadPeriod, statThreadPeriod, TimeUnit.SECONDS);
324 }
325
326 @Override
327 public void setMaxSize(long maxSize) {
328 this.maxSize = maxSize;
329 if(this.size.get() > acceptableSize() && !evictionInProgress) {
330 runEviction();
331 }
332 }
333
334
335
336
337
338
339
340
341
342
343
344
345
346 @Override
347 public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
348 final boolean cacheDataInL1) {
349
350 if (buf.heapSize() > maxBlockSize) {
351
352
353
354 if (stats.failInsert() % 50 == 0) {
355 LOG.warn("Trying to cache too large a block "
356 + cacheKey.getHfileName() + " @ "
357 + cacheKey.getOffset()
358 + " is " + buf.heapSize()
359 + " which is larger than " + maxBlockSize);
360 }
361 return;
362 }
363
364 LruCachedBlock cb = map.get(cacheKey);
365 if (cb != null) {
366
367 if (compare(buf, cb.getBuffer()) != 0) {
368 throw new RuntimeException("Cached block contents differ, which should not have happened."
369 + "cacheKey:" + cacheKey);
370 }
371 String msg = "Cached an already cached block: " + cacheKey + " cb:" + cb.getCacheKey();
372 msg += ". This is harmless and can happen in rare cases (see HBASE-8547)";
373 LOG.debug(msg);
374 return;
375 }
376 long currentSize = size.get();
377 long currentAcceptableSize = acceptableSize();
378 long hardLimitSize = (long) (hardCapacityLimitFactor * currentAcceptableSize);
379 if (currentSize >= hardLimitSize) {
380 stats.failInsert();
381 if (LOG.isTraceEnabled()) {
382 LOG.trace("LruBlockCache current size " + StringUtils.byteDesc(currentSize)
383 + " has exceeded acceptable size " + StringUtils.byteDesc(currentAcceptableSize) + " too many."
384 + " the hard limit size is " + StringUtils.byteDesc(hardLimitSize) + ", failed to put cacheKey:"
385 + cacheKey + " into LruBlockCache.");
386 }
387 if (!evictionInProgress) {
388 runEviction();
389 }
390 return;
391 }
392 cb = new LruCachedBlock(cacheKey, buf, count.incrementAndGet(), inMemory);
393 long newSize = updateSizeMetrics(cb, false);
394 map.put(cacheKey, cb);
395 long val = elements.incrementAndGet();
396 if (LOG.isTraceEnabled()) {
397 long size = map.size();
398 assertCounterSanity(size, val);
399 }
400 if (newSize > currentAcceptableSize && !evictionInProgress) {
401 runEviction();
402 }
403 }
404
405
406
407
408
409 private static void assertCounterSanity(long mapSize, long counterVal) {
410 if (counterVal < 0) {
411 LOG.trace("counterVal overflow. Assertions unreliable. counterVal=" + counterVal +
412 ", mapSize=" + mapSize);
413 return;
414 }
415 if (mapSize < Integer.MAX_VALUE) {
416 double pct_diff = Math.abs((((double) counterVal) / ((double) mapSize)) - 1.);
417 if (pct_diff > 0.05) {
418 LOG.trace("delta between reported and actual size > 5%. counterVal=" + counterVal +
419 ", mapSize=" + mapSize);
420 }
421 }
422 }
423
424 private int compare(Cacheable left, Cacheable right) {
425 ByteBuffer l = ByteBuffer.allocate(left.getSerializedLength());
426 left.serialize(l);
427 ByteBuffer r = ByteBuffer.allocate(right.getSerializedLength());
428 right.serialize(r);
429 return Bytes.compareTo(l.array(), l.arrayOffset(), l.limit(),
430 r.array(), r.arrayOffset(), r.limit());
431 }
432
433
434
435
436
437
438
439 public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
440 cacheBlock(cacheKey, buf, false, false);
441 }
442
443
444
445
446
447
448
449
450
451 protected long updateSizeMetrics(LruCachedBlock cb, boolean evict) {
452 long heapsize = cb.heapSize();
453 if (evict) {
454 heapsize *= -1;
455 }
456 return size.addAndGet(heapsize);
457 }
458
459
460
461
462
463
464
465
466
467
468 @Override
469 public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching, boolean repeat,
470 boolean updateCacheMetrics) {
471 LruCachedBlock cb = map.get(cacheKey);
472 if (cb == null) {
473 if (!repeat && updateCacheMetrics) {
474 stats.miss(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
475 }
476
477
478
479 if (victimHandler != null && !repeat) {
480 Cacheable result = victimHandler.getBlock(cacheKey, caching, repeat, updateCacheMetrics);
481
482
483 if (result != null && caching) {
484 cacheBlock(cacheKey, result,
485 }
486 return result;
487 }
488 return null;
489 }
490 if (updateCacheMetrics) stats.hit(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
491 cb.access(count.incrementAndGet());
492 return cb.getBuffer();
493 }
494
495
496
497
498
499
500 public boolean containsBlock(BlockCacheKey cacheKey) {
501 return map.containsKey(cacheKey);
502 }
503
504 @Override
505 public boolean evictBlock(BlockCacheKey cacheKey) {
506 LruCachedBlock cb = map.get(cacheKey);
507 if (cb == null) return false;
508 evictBlock(cb, false);
509 return true;
510 }
511
512
513
514
515
516
517
518
519
520
521
522 @Override
523 public int evictBlocksByHfileName(String hfileName) {
524 int numEvicted = 0;
525 for (BlockCacheKey key : map.keySet()) {
526 if (key.getHfileName().equals(hfileName)) {
527 if (evictBlock(key))
528 ++numEvicted;
529 }
530 }
531 if (victimHandler != null) {
532 numEvicted += victimHandler.evictBlocksByHfileName(hfileName);
533 }
534 return numEvicted;
535 }
536
537
538
539
540
541
542
543
544
545 protected long evictBlock(LruCachedBlock block, boolean evictedByEvictionProcess) {
546 map.remove(block.getCacheKey());
547 updateSizeMetrics(block, true);
548 long val = elements.decrementAndGet();
549 if (LOG.isTraceEnabled()) {
550 long size = map.size();
551 assertCounterSanity(size, val);
552 }
553 stats.evicted(block.getCachedTime(), block.getCacheKey().isPrimary());
554 if (evictedByEvictionProcess && victimHandler != null) {
555 if (victimHandler instanceof BucketCache) {
556 boolean wait = getCurrentSize() < acceptableSize();
557 boolean inMemory = block.getPriority() == BlockPriority.MEMORY;
558 ((BucketCache)victimHandler).cacheBlockWithWait(block.getCacheKey(), block.getBuffer(),
559 inMemory, wait);
560 } else {
561 victimHandler.cacheBlock(block.getCacheKey(), block.getBuffer());
562 }
563 }
564 return block.heapSize();
565 }
566
567
568
569
570 private void runEviction() {
571 if(evictionThread == null) {
572 evict();
573 } else {
574 evictionThread.evict();
575 }
576 }
577
578
579
580
581 void evict() {
582
583
584 if(!evictionLock.tryLock()) return;
585
586 try {
587 evictionInProgress = true;
588 long currentSize = this.size.get();
589 long bytesToFree = currentSize - minSize();
590
591 if (LOG.isTraceEnabled()) {
592 LOG.trace("Block cache LRU eviction started; Attempting to free " +
593 StringUtils.byteDesc(bytesToFree) + " of total=" +
594 StringUtils.byteDesc(currentSize));
595 }
596
597 if(bytesToFree <= 0) return;
598
599
600 BlockBucket bucketSingle = new BlockBucket("single", bytesToFree, blockSize,
601 singleSize());
602 BlockBucket bucketMulti = new BlockBucket("multi", bytesToFree, blockSize,
603 multiSize());
604 BlockBucket bucketMemory = new BlockBucket("memory", bytesToFree, blockSize,
605 memorySize());
606
607
608 for(LruCachedBlock cachedBlock : map.values()) {
609 switch(cachedBlock.getPriority()) {
610 case SINGLE: {
611 bucketSingle.add(cachedBlock);
612 break;
613 }
614 case MULTI: {
615 bucketMulti.add(cachedBlock);
616 break;
617 }
618 case MEMORY: {
619 bucketMemory.add(cachedBlock);
620 break;
621 }
622 }
623 }
624
625 long bytesFreed = 0;
626 if (forceInMemory || memoryFactor > 0.999f) {
627 long s = bucketSingle.totalSize();
628 long m = bucketMulti.totalSize();
629 if (bytesToFree > (s + m)) {
630
631
632 bytesFreed = bucketSingle.free(s);
633 bytesFreed += bucketMulti.free(m);
634 if (LOG.isTraceEnabled()) {
635 LOG.trace("freed " + StringUtils.byteDesc(bytesFreed) +
636 " from single and multi buckets");
637 }
638 bytesFreed += bucketMemory.free(bytesToFree - bytesFreed);
639 if (LOG.isTraceEnabled()) {
640 LOG.trace("freed " + StringUtils.byteDesc(bytesFreed) +
641 " total from all three buckets ");
642 }
643 } else {
644
645
646
647 long bytesRemain = s + m - bytesToFree;
648 if (3 * s <= bytesRemain) {
649
650
651 bytesFreed = bucketMulti.free(bytesToFree);
652 } else if (3 * m <= 2 * bytesRemain) {
653
654
655 bytesFreed = bucketSingle.free(bytesToFree);
656 } else {
657
658 bytesFreed = bucketSingle.free(s - bytesRemain / 3);
659 if (bytesFreed < bytesToFree) {
660 bytesFreed += bucketMulti.free(bytesToFree - bytesFreed);
661 }
662 }
663 }
664 } else {
665 PriorityQueue<BlockBucket> bucketQueue =
666 new PriorityQueue<BlockBucket>(3);
667
668 bucketQueue.add(bucketSingle);
669 bucketQueue.add(bucketMulti);
670 bucketQueue.add(bucketMemory);
671
672 int remainingBuckets = 3;
673
674 BlockBucket bucket;
675 while((bucket = bucketQueue.poll()) != null) {
676 long overflow = bucket.overflow();
677 if(overflow > 0) {
678 long bucketBytesToFree = Math.min(overflow,
679 (bytesToFree - bytesFreed) / remainingBuckets);
680 bytesFreed += bucket.free(bucketBytesToFree);
681 }
682 remainingBuckets--;
683 }
684 }
685
686 if (LOG.isTraceEnabled()) {
687 long single = bucketSingle.totalSize();
688 long multi = bucketMulti.totalSize();
689 long memory = bucketMemory.totalSize();
690 LOG.trace("Block cache LRU eviction completed; " +
691 "freed=" + StringUtils.byteDesc(bytesFreed) + ", " +
692 "total=" + StringUtils.byteDesc(this.size.get()) + ", " +
693 "single=" + StringUtils.byteDesc(single) + ", " +
694 "multi=" + StringUtils.byteDesc(multi) + ", " +
695 "memory=" + StringUtils.byteDesc(memory));
696 }
697 } finally {
698 stats.evict();
699 evictionInProgress = false;
700 evictionLock.unlock();
701 }
702 }
703
704 @Override
705 public String toString() {
706 return Objects.toStringHelper(this)
707 .add("blockCount", getBlockCount())
708 .add("currentSize", getCurrentSize())
709 .add("freeSize", getFreeSize())
710 .add("maxSize", getMaxSize())
711 .add("heapSize", heapSize())
712 .add("minSize", minSize())
713 .add("minFactor", minFactor)
714 .add("multiSize", multiSize())
715 .add("multiFactor", multiFactor)
716 .add("singleSize", singleSize())
717 .add("singleFactor", singleFactor)
718 .toString();
719 }
720
721
722
723
724
725
726
727 private class BlockBucket implements Comparable<BlockBucket> {
728
729 private final String name;
730 private LruCachedBlockQueue queue;
731 private long totalSize = 0;
732 private long bucketSize;
733
734 public BlockBucket(String name, long bytesToFree, long blockSize, long bucketSize) {
735 this.name = name;
736 this.bucketSize = bucketSize;
737 queue = new LruCachedBlockQueue(bytesToFree, blockSize);
738 totalSize = 0;
739 }
740
741 public void add(LruCachedBlock block) {
742 totalSize += block.heapSize();
743 queue.add(block);
744 }
745
746 public long free(long toFree) {
747 if (LOG.isTraceEnabled()) {
748 LOG.trace("freeing " + StringUtils.byteDesc(toFree) + " from " + this);
749 }
750 LruCachedBlock cb;
751 long freedBytes = 0;
752 while ((cb = queue.pollLast()) != null) {
753 freedBytes += evictBlock(cb, true);
754 if (freedBytes >= toFree) {
755 return freedBytes;
756 }
757 }
758 if (LOG.isTraceEnabled()) {
759 LOG.trace("freed " + StringUtils.byteDesc(freedBytes) + " from " + this);
760 }
761 return freedBytes;
762 }
763
764 public long overflow() {
765 return totalSize - bucketSize;
766 }
767
768 public long totalSize() {
769 return totalSize;
770 }
771
772 public int compareTo(BlockBucket that) {
773 return Long.compare(this.overflow(), that.overflow());
774 }
775
776 @Override
777 public boolean equals(Object that) {
778 if (that == null || !(that instanceof BlockBucket)){
779 return false;
780 }
781 return compareTo((BlockBucket)that) == 0;
782 }
783
784 @Override
785 public int hashCode() {
786 return Objects.hashCode(name, bucketSize, queue, totalSize);
787 }
788
789 @Override
790 public String toString() {
791 return Objects.toStringHelper(this)
792 .add("name", name)
793 .add("totalSize", StringUtils.byteDesc(totalSize))
794 .add("bucketSize", StringUtils.byteDesc(bucketSize))
795 .toString();
796 }
797 }
798
799
800
801
802
803 public long getMaxSize() {
804 return this.maxSize;
805 }
806
807 @Override
808 public long getCurrentSize() {
809 return this.size.get();
810 }
811
812 @Override
813 public long getFreeSize() {
814 return getMaxSize() - getCurrentSize();
815 }
816
817 @Override
818 public long size() {
819 return getMaxSize();
820 }
821
822 @Override
823 public long getBlockCount() {
824 return this.elements.get();
825 }
826
827 EvictionThread getEvictionThread() {
828 return this.evictionThread;
829 }
830
831
832
833
834
835
836
837 static class EvictionThread extends HasThread {
838 private WeakReference<LruBlockCache> cache;
839 private volatile boolean go = true;
840
841 private boolean enteringRun = false;
842
843 public EvictionThread(LruBlockCache cache) {
844 super(Thread.currentThread().getName() + ".LruBlockCache.EvictionThread");
845 setDaemon(true);
846 this.cache = new WeakReference<LruBlockCache>(cache);
847 }
848
849 @Override
850 public void run() {
851 enteringRun = true;
852 while (this.go) {
853 synchronized(this) {
854 try {
855 this.wait(1000 * 10
856 } catch(InterruptedException e) {
857 LOG.warn("Interrupted eviction thread ", e);
858 Thread.currentThread().interrupt();
859 }
860 }
861 LruBlockCache cache = this.cache.get();
862 if (cache == null) break;
863 cache.evict();
864 }
865 }
866
867 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="NN_NAKED_NOTIFY",
868 justification="This is what we want")
869 public void evict() {
870 synchronized(this) {
871 this.notifyAll();
872 }
873 }
874
875 synchronized void shutdown() {
876 this.go = false;
877 this.notifyAll();
878 }
879
880
881
882
883 boolean isEnteringRun() {
884 return this.enteringRun;
885 }
886 }
887
888
889
890
891 static class StatisticsThread extends Thread {
892 private final LruBlockCache lru;
893
894 public StatisticsThread(LruBlockCache lru) {
895 super("LruBlockCacheStats");
896 setDaemon(true);
897 this.lru = lru;
898 }
899
900 @Override
901 public void run() {
902 lru.logStats();
903 }
904 }
905
906 public void logStats() {
907
908 long totalSize = heapSize();
909 long freeSize = maxSize - totalSize;
910 LruBlockCache.LOG.info("totalSize=" + StringUtils.byteDesc(totalSize) + ", " +
911 "freeSize=" + StringUtils.byteDesc(freeSize) + ", " +
912 "max=" + StringUtils.byteDesc(this.maxSize) + ", " +
913 "blockCount=" + getBlockCount() + ", " +
914 "accesses=" + stats.getRequestCount() + ", " +
915 "hits=" + stats.getHitCount() + ", " +
916 "hitRatio=" + (stats.getHitCount() == 0 ?
917 "0" : (StringUtils.formatPercent(stats.getHitRatio(), 2)+ ", ")) + ", " +
918 "cachingAccesses=" + stats.getRequestCachingCount() + ", " +
919 "cachingHits=" + stats.getHitCachingCount() + ", " +
920 "cachingHitsRatio=" + (stats.getHitCachingCount() == 0 ?
921 "0,": (StringUtils.formatPercent(stats.getHitCachingRatio(), 2) + ", ")) +
922 "evictions=" + stats.getEvictionCount() + ", " +
923 "evicted=" + stats.getEvictedCount() + ", " +
924 "evictedPerRun=" + stats.evictedPerEviction());
925 }
926
927
928
929
930
931
932
933 public CacheStats getStats() {
934 return this.stats;
935 }
936
937 public final static long CACHE_FIXED_OVERHEAD = ClassSize.align(
938 (4 * Bytes.SIZEOF_LONG) + (9 * ClassSize.REFERENCE) +
939 (6 * Bytes.SIZEOF_FLOAT) + (2 * Bytes.SIZEOF_BOOLEAN)
940 + ClassSize.OBJECT);
941
942 @Override
943 public long heapSize() {
944 return getCurrentSize();
945 }
946
947 public static long calculateOverhead(long maxSize, long blockSize, int concurrency){
948
949 return CACHE_FIXED_OVERHEAD + ClassSize.CONCURRENT_HASHMAP +
950 ((long)Math.ceil(maxSize*1.2/blockSize)
951 * ClassSize.CONCURRENT_HASHMAP_ENTRY) +
952 ((long)concurrency * ClassSize.CONCURRENT_HASHMAP_SEGMENT);
953 }
954
955 @Override
956 public Iterator<CachedBlock> iterator() {
957 final Iterator<LruCachedBlock> iterator = map.values().iterator();
958
959 return new Iterator<CachedBlock>() {
960 private final long now = System.nanoTime();
961
962 @Override
963 public boolean hasNext() {
964 return iterator.hasNext();
965 }
966
967 @Override
968 public CachedBlock next() {
969 final LruCachedBlock b = iterator.next();
970 return new CachedBlock() {
971 @Override
972 public String toString() {
973 return BlockCacheUtil.toString(this, now);
974 }
975
976 @Override
977 public BlockPriority getBlockPriority() {
978 return b.getPriority();
979 }
980
981 @Override
982 public BlockType getBlockType() {
983 return b.getBuffer().getBlockType();
984 }
985
986 @Override
987 public long getOffset() {
988 return b.getCacheKey().getOffset();
989 }
990
991 @Override
992 public long getSize() {
993 return b.getBuffer().heapSize();
994 }
995
996 @Override
997 public long getCachedTime() {
998 return b.getCachedTime();
999 }
1000
1001 @Override
1002 public String getFilename() {
1003 return b.getCacheKey().getHfileName();
1004 }
1005
1006 @Override
1007 public int compareTo(CachedBlock other) {
1008 int diff = this.getFilename().compareTo(other.getFilename());
1009 if (diff != 0) return diff;
1010 diff = Long.compare(this.getOffset(), other.getOffset());
1011 if (diff != 0) return diff;
1012 if (other.getCachedTime() < 0 || this.getCachedTime() < 0) {
1013 throw new IllegalStateException("" + this.getCachedTime() + ", " +
1014 other.getCachedTime());
1015 }
1016 return Long.compare(other.getCachedTime(), this.getCachedTime());
1017 }
1018
1019 @Override
1020 public int hashCode() {
1021 return b.hashCode();
1022 }
1023
1024 @Override
1025 public boolean equals(Object obj) {
1026 if (obj instanceof CachedBlock) {
1027 CachedBlock cb = (CachedBlock)obj;
1028 return compareTo(cb) == 0;
1029 } else {
1030 return false;
1031 }
1032 }
1033 };
1034 }
1035
1036 @Override
1037 public void remove() {
1038 throw new UnsupportedOperationException();
1039 }
1040 };
1041 }
1042
1043
1044
1045 long acceptableSize() {
1046 return (long)Math.floor(this.maxSize * this.acceptableFactor);
1047 }
1048 private long minSize() {
1049 return (long)Math.floor(this.maxSize * this.minFactor);
1050 }
1051 private long singleSize() {
1052 return (long)Math.floor(this.maxSize * this.singleFactor * this.minFactor);
1053 }
1054 private long multiSize() {
1055 return (long)Math.floor(this.maxSize * this.multiFactor * this.minFactor);
1056 }
1057 private long memorySize() {
1058 return (long)Math.floor(this.maxSize * this.memoryFactor * this.minFactor);
1059 }
1060
1061 public void shutdown() {
1062 if (victimHandler != null)
1063 victimHandler.shutdown();
1064 this.scheduleThreadPool.shutdown();
1065 for (int i = 0; i < 10; i++) {
1066 if (!this.scheduleThreadPool.isShutdown()) {
1067 try {
1068 Thread.sleep(10);
1069 } catch (InterruptedException e) {
1070 LOG.warn("Interrupted while sleeping");
1071 Thread.currentThread().interrupt();
1072 break;
1073 }
1074 }
1075 }
1076
1077 if (!this.scheduleThreadPool.isShutdown()) {
1078 List<Runnable> runnables = this.scheduleThreadPool.shutdownNow();
1079 LOG.debug("Still running " + runnables);
1080 }
1081 this.evictionThread.shutdown();
1082 }
1083
1084
1085 @VisibleForTesting
1086 public void clearCache() {
1087 this.map.clear();
1088 this.elements.set(0);
1089 }
1090
1091
1092
1093
1094
1095 @VisibleForTesting
1096 SortedSet<String> getCachedFileNamesForTest() {
1097 SortedSet<String> fileNames = new TreeSet<String>();
1098 for (BlockCacheKey cacheKey : map.keySet()) {
1099 fileNames.add(cacheKey.getHfileName());
1100 }
1101 return fileNames;
1102 }
1103
1104 @VisibleForTesting
1105 Map<BlockType, Integer> getBlockTypeCountsForTest() {
1106 Map<BlockType, Integer> counts =
1107 new EnumMap<BlockType, Integer>(BlockType.class);
1108 for (LruCachedBlock cb : map.values()) {
1109 BlockType blockType = ((Cacheable)cb.getBuffer()).getBlockType();
1110 Integer count = counts.get(blockType);
1111 counts.put(blockType, (count == null ? 0 : count) + 1);
1112 }
1113 return counts;
1114 }
1115
1116 @VisibleForTesting
1117 public Map<DataBlockEncoding, Integer> getEncodingCountsForTest() {
1118 Map<DataBlockEncoding, Integer> counts =
1119 new EnumMap<DataBlockEncoding, Integer>(DataBlockEncoding.class);
1120 for (LruCachedBlock block : map.values()) {
1121 DataBlockEncoding encoding =
1122 ((HFileBlock) block.getBuffer()).getDataBlockEncoding();
1123 Integer count = counts.get(encoding);
1124 counts.put(encoding, (count == null ? 0 : count) + 1);
1125 }
1126 return counts;
1127 }
1128
1129 public void setVictimCache(BlockCache handler) {
1130 assert victimHandler == null;
1131 victimHandler = handler;
1132 }
1133
1134 @VisibleForTesting
1135 Map<BlockCacheKey, LruCachedBlock> getMapForTests() {
1136 return map;
1137 }
1138
1139 BlockCache getVictimHandler() {
1140 return this.victimHandler;
1141 }
1142
1143 @Override
1144 public BlockCache[] getBlockCaches() {
1145 return null;
1146 }
1147 }