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