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