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