001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.io.hfile.bucket; 019 020import static org.apache.hadoop.hbase.io.hfile.CacheConfig.BUCKETCACHE_PERSIST_INTERVAL_KEY; 021import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.ACCEPT_FACTOR_CONFIG_NAME; 022import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.BACKING_MAP_PERSISTENCE_CHUNK_SIZE; 023import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.BLOCK_ORPHAN_GRACE_PERIOD; 024import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.DEFAULT_ERROR_TOLERATION_DURATION; 025import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.DEFAULT_MIN_FACTOR; 026import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.DEFAULT_SINGLE_FACTOR; 027import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.EXTRA_FREE_FACTOR_CONFIG_NAME; 028import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.MEMORY_FACTOR_CONFIG_NAME; 029import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.MIN_FACTOR_CONFIG_NAME; 030import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.MULTI_FACTOR_CONFIG_NAME; 031import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.QUEUE_ADDITION_WAIT_TIME; 032import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.SINGLE_FACTOR_CONFIG_NAME; 033import static org.junit.jupiter.api.Assertions.assertEquals; 034import static org.junit.jupiter.api.Assertions.assertFalse; 035import static org.junit.jupiter.api.Assertions.assertNotEquals; 036import static org.junit.jupiter.api.Assertions.assertNotNull; 037import static org.junit.jupiter.api.Assertions.assertNull; 038import static org.junit.jupiter.api.Assertions.assertSame; 039import static org.junit.jupiter.api.Assertions.assertTrue; 040import static org.junit.jupiter.api.Assertions.fail; 041import static org.mockito.Mockito.mock; 042import static org.mockito.Mockito.when; 043 044import java.io.File; 045import java.io.IOException; 046import java.lang.reflect.Field; 047import java.nio.ByteBuffer; 048import java.util.ArrayList; 049import java.util.Arrays; 050import java.util.Collection; 051import java.util.HashMap; 052import java.util.List; 053import java.util.Map; 054import java.util.Set; 055import java.util.concurrent.ThreadLocalRandom; 056import java.util.concurrent.atomic.LongAdder; 057import java.util.concurrent.locks.ReentrantReadWriteLock; 058import java.util.stream.Stream; 059import org.apache.hadoop.conf.Configuration; 060import org.apache.hadoop.fs.Path; 061import org.apache.hadoop.hbase.HBaseConfiguration; 062import org.apache.hadoop.hbase.HBaseParameterizedTestTemplate; 063import org.apache.hadoop.hbase.HBaseTestingUtil; 064import org.apache.hadoop.hbase.HConstants; 065import org.apache.hadoop.hbase.Waiter; 066import org.apache.hadoop.hbase.io.ByteBuffAllocator; 067import org.apache.hadoop.hbase.io.hfile.BlockCacheKey; 068import org.apache.hadoop.hbase.io.hfile.BlockPriority; 069import org.apache.hadoop.hbase.io.hfile.BlockType; 070import org.apache.hadoop.hbase.io.hfile.CacheStats; 071import org.apache.hadoop.hbase.io.hfile.CacheTestUtils; 072import org.apache.hadoop.hbase.io.hfile.CacheTestUtils.HFileBlockPair; 073import org.apache.hadoop.hbase.io.hfile.Cacheable; 074import org.apache.hadoop.hbase.io.hfile.HFileBlock; 075import org.apache.hadoop.hbase.io.hfile.HFileContext; 076import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder; 077import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.BucketSizeInfo; 078import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.IndexStatistics; 079import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.RAMCache; 080import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.RAMQueueEntry; 081import org.apache.hadoop.hbase.nio.ByteBuff; 082import org.apache.hadoop.hbase.regionserver.HRegion; 083import org.apache.hadoop.hbase.regionserver.HStore; 084import org.apache.hadoop.hbase.regionserver.HStoreFile; 085import org.apache.hadoop.hbase.testclassification.IOTests; 086import org.apache.hadoop.hbase.testclassification.LargeTests; 087import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 088import org.apache.hadoop.hbase.util.Pair; 089import org.apache.hadoop.hbase.util.Threads; 090import org.junit.jupiter.api.AfterEach; 091import org.junit.jupiter.api.BeforeEach; 092import org.junit.jupiter.api.Tag; 093import org.junit.jupiter.api.TestTemplate; 094import org.junit.jupiter.params.provider.Arguments; 095import org.mockito.Mockito; 096import org.slf4j.Logger; 097import org.slf4j.LoggerFactory; 098 099import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap; 100 101/** 102 * Basic test of BucketCache.Puts and gets. 103 * <p> 104 * Tests will ensure that blocks' data correctness under several threads concurrency 105 */ 106@Tag(IOTests.TAG) 107@Tag(LargeTests.TAG) 108@HBaseParameterizedTestTemplate(name = "{index}: blockSize={0}, bucketSizes={1}") 109public class TestBucketCache { 110 111 private static final Logger LOG = LoggerFactory.getLogger(TestBucketCache.class); 112 113 public static Stream<Arguments> parameters() { 114 // TODO: why is 8k the default blocksize for these tests? 115 return Stream.of(Arguments.of(8192, null), 116 Arguments.of(16 * 1024, 117 new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024, 8 * 1024 + 1024, 16 * 1024 + 1024, 118 28 * 1024 + 1024, 32 * 1024 + 1024, 64 * 1024 + 1024, 96 * 1024 + 1024, 119 128 * 1024 + 1024 })); 120 } 121 122 private final int constructedBlockSize; 123 private final int[] constructedBlockSizes; 124 125 public TestBucketCache(int constructedBlockSize, int[] constructedBlockSizes) { 126 this.constructedBlockSize = constructedBlockSize; 127 this.constructedBlockSizes = constructedBlockSizes; 128 } 129 130 BucketCache cache; 131 final int CACHE_SIZE = 1000000; 132 final int NUM_BLOCKS = 100; 133 final int BLOCK_SIZE = CACHE_SIZE / NUM_BLOCKS; 134 final int NUM_THREADS = 100; 135 final int NUM_QUERIES = 10000; 136 137 final long capacitySize = 32 * 1024 * 1024; 138 final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS; 139 final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS; 140 private String ioEngineName = "offheap"; 141 142 private static final HBaseTestingUtil HBASE_TESTING_UTILITY = new HBaseTestingUtil(); 143 144 private static class MockedBucketCache extends BucketCache { 145 146 public MockedBucketCache(String ioEngineName, long capacity, int blockSize, int[] bucketSizes, 147 int writerThreads, int writerQLen, String persistencePath) throws IOException { 148 super(ioEngineName, capacity, blockSize, bucketSizes, writerThreads, writerQLen, 149 persistencePath); 150 } 151 152 @Override 153 public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory) { 154 super.cacheBlock(cacheKey, buf, inMemory); 155 } 156 157 @Override 158 public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) { 159 super.cacheBlock(cacheKey, buf); 160 } 161 } 162 163 @BeforeEach 164 public void setup() throws IOException { 165 cache = new MockedBucketCache(ioEngineName, capacitySize, constructedBlockSize, 166 constructedBlockSizes, writeThreads, writerQLen, null); 167 } 168 169 @AfterEach 170 public void tearDown() { 171 cache.shutdown(); 172 } 173 174 /** 175 * Test Utility to create test dir and return name 176 * @return return name of created dir 177 * @throws IOException throws IOException 178 */ 179 private Path createAndGetTestDir() throws IOException { 180 final Path testDir = HBASE_TESTING_UTILITY.getDataTestDir(); 181 HBASE_TESTING_UTILITY.getTestFileSystem().mkdirs(testDir); 182 return testDir; 183 } 184 185 /** 186 * Return a random element from {@code a}. 187 */ 188 private static <T> T randFrom(List<T> a) { 189 return a.get(ThreadLocalRandom.current().nextInt(a.size())); 190 } 191 192 @TestTemplate 193 public void testBucketAllocator() throws BucketAllocatorException { 194 BucketAllocator mAllocator = cache.getAllocator(); 195 /* 196 * Test the allocator first 197 */ 198 final List<Integer> BLOCKSIZES = Arrays.asList(4 * 1024, 8 * 1024, 64 * 1024, 96 * 1024); 199 200 boolean full = false; 201 ArrayList<Pair<Long, Integer>> allocations = new ArrayList<>(); 202 // Fill the allocated extents by choosing a random blocksize. Continues selecting blocks until 203 // the cache is completely filled. 204 List<Integer> tmp = new ArrayList<>(BLOCKSIZES); 205 while (!full) { 206 Integer blockSize = null; 207 try { 208 blockSize = randFrom(tmp); 209 allocations.add(new Pair<>(mAllocator.allocateBlock(blockSize), blockSize)); 210 } catch (CacheFullException cfe) { 211 tmp.remove(blockSize); 212 if (tmp.isEmpty()) full = true; 213 } 214 } 215 216 for (Integer blockSize : BLOCKSIZES) { 217 BucketSizeInfo bucketSizeInfo = mAllocator.roundUpToBucketSizeInfo(blockSize); 218 IndexStatistics indexStatistics = bucketSizeInfo.statistics(); 219 assertEquals(0, indexStatistics.freeCount(), "unexpected freeCount for " + bucketSizeInfo); 220 221 // we know the block sizes above are multiples of 1024, but default bucket sizes give an 222 // additional 1024 on top of that so this counts towards fragmentation in our test 223 // real life may have worse fragmentation because blocks may not be perfectly sized to block 224 // size, given encoding/compression and large rows 225 assertEquals(1024 * indexStatistics.totalCount(), indexStatistics.fragmentationBytes()); 226 } 227 228 mAllocator.logDebugStatistics(); 229 230 for (Pair<Long, Integer> allocation : allocations) { 231 assertEquals(mAllocator.sizeOfAllocation(allocation.getFirst()), 232 mAllocator.freeBlock(allocation.getFirst(), allocation.getSecond())); 233 } 234 assertEquals(0, mAllocator.getUsedSize()); 235 } 236 237 @TestTemplate 238 public void testCacheSimple() throws Exception { 239 CacheTestUtils.testCacheSimple(cache, BLOCK_SIZE, NUM_QUERIES); 240 } 241 242 @TestTemplate 243 public void testCacheMultiThreadedSingleKey() throws Exception { 244 CacheTestUtils.hammerSingleKey(cache, 2 * NUM_THREADS, 2 * NUM_QUERIES); 245 } 246 247 @TestTemplate 248 public void testHeapSizeChanges() throws Exception { 249 cache.stopWriterThreads(); 250 CacheTestUtils.testHeapSizeChanges(cache, BLOCK_SIZE); 251 } 252 253 public static void waitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey) 254 throws InterruptedException { 255 Waiter.waitFor(HBaseConfiguration.create(), 10000, 256 () -> (cache.backingMap.containsKey(cacheKey) && !cache.ramCache.containsKey(cacheKey))); 257 } 258 259 public static void waitUntilAllFlushedToBucket(BucketCache cache) throws InterruptedException { 260 while (!cache.ramCache.isEmpty()) { 261 Thread.sleep(100); 262 } 263 Thread.sleep(1000); 264 } 265 266 // BucketCache.cacheBlock is async, it first adds block to ramCache and writeQueue, then writer 267 // threads will flush it to the bucket and put reference entry in backingMap. 268 private void cacheAndWaitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey, 269 Cacheable block, boolean waitWhenCache) throws InterruptedException { 270 cache.cacheBlock(cacheKey, block, false, waitWhenCache); 271 waitUntilFlushedToBucket(cache, cacheKey); 272 } 273 274 @TestTemplate 275 public void testMemoryLeak() throws Exception { 276 final BlockCacheKey cacheKey = new BlockCacheKey("dummy", 1L); 277 cacheAndWaitUntilFlushedToBucket(cache, cacheKey, 278 new CacheTestUtils.ByteArrayCacheable(new byte[10]), true); 279 BucketEntry oldEntry = cache.backingMap.get(cacheKey); 280 long lockId = oldEntry.offset(); 281 ReentrantReadWriteLock lock = cache.offsetLock.getLock(lockId); 282 Thread evictThread = new Thread("evict-block") { 283 @Override 284 public void run() { 285 cache.evictBlock(cacheKey); 286 } 287 }; 288 BucketEntry replacementEntry; 289 lock.writeLock().lock(); 290 try { 291 evictThread.start(); 292 cache.offsetLock.waitForWaiters(lockId, 1); 293 assertTrue(cache.backingMap.remove(cacheKey, oldEntry)); 294 cache.blockEvicted(cacheKey, oldEntry, true, true); 295 assertEquals(0, cache.getBlockCount()); 296 cache.cacheBlock(cacheKey, new CacheTestUtils.ByteArrayCacheable(new byte[10]), false, true); 297 // The replacement is published before taking its offset lock. Full flush must wait until the 298 // old entry's lock is released. 299 Waiter.waitFor(HBaseConfiguration.create(), 10000, () -> { 300 BucketEntry currentEntry = cache.backingMap.get(cacheKey); 301 return currentEntry != null && currentEntry != oldEntry; 302 }); 303 replacementEntry = cache.backingMap.get(cacheKey); 304 } finally { 305 lock.writeLock().unlock(); 306 } 307 waitUntilFlushedToBucket(cache, cacheKey); 308 assertEquals(1, cache.getBlockCount()); 309 evictThread.join(); 310 /** 311 * <pre> 312 * The asserts here before HBASE-21957 are: 313 * assertEquals(1L, cache.getBlockCount()); 314 * assertTrue(cache.getCurrentSize() > 0L); 315 * assertTrue("We should have a block!", cache.iterator().hasNext()); 316 * 317 * The asserts here after HBASE-21957 are: 318 * assertEquals(0, cache.getBlockCount()); 319 * assertEquals(cache.getCurrentSize(), 0L); 320 * 321 * I think the asserts before HBASE-21957 is more reasonable,because 322 * {@link BucketCache#evictBlock} should only evict the {@link BucketEntry} 323 * it had seen, and newly added Block after the {@link BucketEntry} 324 * it had seen should not be evicted. 325 * </pre> 326 */ 327 assertSame(replacementEntry, cache.backingMap.get(cacheKey)); 328 assertTrue(cache.blocksByHFile.contains(cacheKey)); 329 assertEquals(1L, cache.getBlockCount()); 330 assertTrue(cache.getCurrentSize() > 0L); 331 assertTrue(cache.iterator().hasNext(), "We should have a block!"); 332 } 333 334 @TestTemplate 335 public void testRetrieveFromFile() throws Exception { 336 Path testDir = createAndGetTestDir(); 337 String ioEngineName = "file:" + testDir + "/bucket.cache"; 338 testRetrievalUtils(testDir, ioEngineName); 339 int[] smallBucketSizes = new int[] { 3 * 1024, 5 * 1024 }; 340 String persistencePath = testDir + "/bucket.persistence"; 341 BucketCache bucketCache = null; 342 try { 343 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 344 smallBucketSizes, writeThreads, writerQLen, persistencePath); 345 assertTrue(bucketCache.waitForCacheInitialization(10000)); 346 assertFalse(new File(persistencePath).exists()); 347 assertEquals(0, bucketCache.getAllocator().getUsedSize()); 348 assertEquals(0, bucketCache.backingMap.size()); 349 } finally { 350 bucketCache.shutdown(); 351 HBASE_TESTING_UTILITY.cleanupTestDir(); 352 } 353 } 354 355 @TestTemplate 356 public void testRetrieveFromMMap() throws Exception { 357 final Path testDir = createAndGetTestDir(); 358 final String ioEngineName = "mmap:" + testDir + "/bucket.cache"; 359 testRetrievalUtils(testDir, ioEngineName); 360 } 361 362 @TestTemplate 363 public void testRetrieveFromPMem() throws Exception { 364 final Path testDir = createAndGetTestDir(); 365 final String ioEngineName = "pmem:" + testDir + "/bucket.cache"; 366 testRetrievalUtils(testDir, ioEngineName); 367 int[] smallBucketSizes = new int[] { 3 * 1024, 5 * 1024 }; 368 String persistencePath = testDir + "/bucket.persistence" + EnvironmentEdgeManager.currentTime(); 369 BucketCache bucketCache = null; 370 try { 371 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 372 smallBucketSizes, writeThreads, writerQLen, persistencePath); 373 assertTrue(bucketCache.waitForCacheInitialization(10000)); 374 assertFalse(new File(persistencePath).exists()); 375 assertEquals(0, bucketCache.getAllocator().getUsedSize()); 376 assertEquals(0, bucketCache.backingMap.size()); 377 } finally { 378 bucketCache.shutdown(); 379 HBASE_TESTING_UTILITY.cleanupTestDir(); 380 } 381 } 382 383 private void testRetrievalUtils(Path testDir, String ioEngineName) 384 throws IOException, InterruptedException { 385 final String persistencePath = 386 testDir + "/bucket.persistence" + EnvironmentEdgeManager.currentTime(); 387 BucketCache bucketCache = null; 388 try { 389 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 390 constructedBlockSizes, writeThreads, writerQLen, persistencePath); 391 assertTrue(bucketCache.waitForCacheInitialization(10000)); 392 long usedSize = bucketCache.getAllocator().getUsedSize(); 393 assertEquals(0, usedSize); 394 HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1); 395 for (HFileBlockPair block : blocks) { 396 bucketCache.cacheBlock(block.getBlockName(), block.getBlock()); 397 } 398 for (HFileBlockPair block : blocks) { 399 cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock(), 400 false); 401 } 402 usedSize = bucketCache.getAllocator().getUsedSize(); 403 assertNotEquals(0, usedSize); 404 BucketEntry persistedEntry = bucketCache.backingMap.values().iterator().next(); 405 assertEquals(1, persistedEntry.refCnt()); 406 bucketCache.shutdown(); 407 assertEquals(0, persistedEntry.refCnt()); 408 assertTrue(bucketCache.backingMap.isEmpty()); 409 assertTrue(new File(persistencePath).exists()); 410 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 411 constructedBlockSizes, writeThreads, writerQLen, persistencePath); 412 assertTrue(bucketCache.waitForCacheInitialization(10000)); 413 414 assertEquals(usedSize, bucketCache.getAllocator().getUsedSize()); 415 } finally { 416 if (bucketCache != null) { 417 bucketCache.shutdown(); 418 } 419 } 420 assertTrue(new File(persistencePath).exists()); 421 } 422 423 @TestTemplate 424 public void testRetrieveUnsupportedIOE() throws Exception { 425 try { 426 final Path testDir = createAndGetTestDir(); 427 final String ioEngineName = testDir + "/bucket.cache"; 428 testRetrievalUtils(testDir, ioEngineName); 429 fail("Should have thrown IllegalArgumentException because of unsupported IOEngine!!"); 430 } catch (IllegalArgumentException e) { 431 assertEquals("Don't understand io engine name for cache- prefix with file:, " 432 + "files:, mmap: or offheap", e.getMessage()); 433 } 434 } 435 436 @TestTemplate 437 public void testRetrieveFromMultipleFiles() throws Exception { 438 final Path testDirInitial = createAndGetTestDir(); 439 final Path newTestDir = new HBaseTestingUtil().getDataTestDir(); 440 HBASE_TESTING_UTILITY.getTestFileSystem().mkdirs(newTestDir); 441 String ioEngineName = 442 new StringBuilder("files:").append(testDirInitial).append("/bucket1.cache") 443 .append(FileIOEngine.FILE_DELIMITER).append(newTestDir).append("/bucket2.cache").toString(); 444 testRetrievalUtils(testDirInitial, ioEngineName); 445 int[] smallBucketSizes = new int[] { 3 * 1024, 5 * 1024 }; 446 String persistencePath = testDirInitial + "/bucket.persistence"; 447 BucketCache bucketCache = null; 448 try { 449 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 450 smallBucketSizes, writeThreads, writerQLen, persistencePath); 451 assertTrue(bucketCache.waitForCacheInitialization(10000)); 452 assertFalse(new File(persistencePath).exists()); 453 assertEquals(0, bucketCache.getAllocator().getUsedSize()); 454 assertEquals(0, bucketCache.backingMap.size()); 455 } finally { 456 bucketCache.shutdown(); 457 HBASE_TESTING_UTILITY.cleanupTestDir(); 458 } 459 } 460 461 @TestTemplate 462 public void testRetrieveFromFileWithoutPersistence() throws Exception { 463 BucketCache bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 464 constructedBlockSizes, writeThreads, writerQLen, null); 465 assertTrue(bucketCache.waitForCacheInitialization(10000)); 466 try { 467 final Path testDir = createAndGetTestDir(); 468 String ioEngineName = "file:" + testDir + "/bucket.cache"; 469 long usedSize = bucketCache.getAllocator().getUsedSize(); 470 assertEquals(0, usedSize); 471 HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1); 472 for (HFileBlockPair block : blocks) { 473 bucketCache.cacheBlock(block.getBlockName(), block.getBlock()); 474 } 475 for (HFileBlockPair block : blocks) { 476 cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock(), 477 false); 478 } 479 usedSize = bucketCache.getAllocator().getUsedSize(); 480 assertNotEquals(0, usedSize); 481 bucketCache.shutdown(); 482 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 483 constructedBlockSizes, writeThreads, writerQLen, null); 484 assertTrue(bucketCache.waitForCacheInitialization(10000)); 485 assertEquals(0, bucketCache.getAllocator().getUsedSize()); 486 } finally { 487 bucketCache.shutdown(); 488 HBASE_TESTING_UTILITY.cleanupTestDir(); 489 } 490 } 491 492 @TestTemplate 493 public void testBucketAllocatorLargeBuckets() throws BucketAllocatorException { 494 long availableSpace = 20 * 1024L * 1024 * 1024; 495 int[] bucketSizes = new int[] { 1024, 1024 * 1024, 1024 * 1024 * 1024 }; 496 BucketAllocator allocator = new BucketAllocator(availableSpace, bucketSizes); 497 assertTrue(allocator.getBuckets().length > 0); 498 } 499 500 @TestTemplate 501 public void testGetPartitionSize() throws IOException { 502 // Test default values 503 validateGetPartitionSize(cache, DEFAULT_SINGLE_FACTOR, DEFAULT_MIN_FACTOR); 504 505 Configuration conf = HBaseConfiguration.create(); 506 conf.setFloat(MIN_FACTOR_CONFIG_NAME, 0.5f); 507 conf.setFloat(SINGLE_FACTOR_CONFIG_NAME, 0.1f); 508 conf.setFloat(MULTI_FACTOR_CONFIG_NAME, 0.7f); 509 conf.setFloat(MEMORY_FACTOR_CONFIG_NAME, 0.2f); 510 511 BucketCache cache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 512 constructedBlockSizes, writeThreads, writerQLen, null, 100, conf); 513 assertTrue(cache.waitForCacheInitialization(10000)); 514 515 validateGetPartitionSize(cache, 0.1f, 0.5f); 516 validateGetPartitionSize(cache, 0.7f, 0.5f); 517 validateGetPartitionSize(cache, 0.2f, 0.5f); 518 } 519 520 @TestTemplate 521 public void testCacheSizeCapacity() throws IOException { 522 // Test cache capacity (capacity / blockSize) < Integer.MAX_VALUE 523 validateGetPartitionSize(cache, DEFAULT_SINGLE_FACTOR, DEFAULT_MIN_FACTOR); 524 Configuration conf = HBaseConfiguration.create(); 525 conf.setFloat(BucketCache.MIN_FACTOR_CONFIG_NAME, 0.5f); 526 conf.setFloat(SINGLE_FACTOR_CONFIG_NAME, 0.1f); 527 conf.setFloat(MULTI_FACTOR_CONFIG_NAME, 0.7f); 528 conf.setFloat(MEMORY_FACTOR_CONFIG_NAME, 0.2f); 529 try { 530 new BucketCache(ioEngineName, Long.MAX_VALUE, 1, constructedBlockSizes, writeThreads, 531 writerQLen, null, 100, conf); 532 fail("Should have thrown IllegalArgumentException because of large cache capacity!"); 533 } catch (IllegalArgumentException e) { 534 assertEquals("Cache capacity is too large, only support 32TB now", e.getMessage()); 535 } 536 } 537 538 @TestTemplate 539 public void testValidBucketCacheConfigs() throws IOException { 540 Configuration conf = HBaseConfiguration.create(); 541 conf.setFloat(ACCEPT_FACTOR_CONFIG_NAME, 0.9f); 542 conf.setFloat(MIN_FACTOR_CONFIG_NAME, 0.5f); 543 conf.setFloat(EXTRA_FREE_FACTOR_CONFIG_NAME, 0.5f); 544 conf.setFloat(SINGLE_FACTOR_CONFIG_NAME, 0.1f); 545 conf.setFloat(MULTI_FACTOR_CONFIG_NAME, 0.7f); 546 conf.setFloat(MEMORY_FACTOR_CONFIG_NAME, 0.2f); 547 548 BucketCache cache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 549 constructedBlockSizes, writeThreads, writerQLen, null, 100, conf); 550 assertTrue(cache.waitForCacheInitialization(10000)); 551 552 assertEquals(0.9f, cache.getAcceptableFactor(), 0, 553 ACCEPT_FACTOR_CONFIG_NAME + " failed to propagate."); 554 assertEquals(0.5f, cache.getMinFactor(), 0, MIN_FACTOR_CONFIG_NAME + " failed to propagate."); 555 assertEquals(0.5f, cache.getExtraFreeFactor(), 0, 556 EXTRA_FREE_FACTOR_CONFIG_NAME + " failed to propagate."); 557 assertEquals(0.1f, cache.getSingleFactor(), 0, 558 SINGLE_FACTOR_CONFIG_NAME + " failed to propagate."); 559 assertEquals(0.7f, cache.getMultiFactor(), 0, 560 MULTI_FACTOR_CONFIG_NAME + " failed to propagate."); 561 assertEquals(0.2f, cache.getMemoryFactor(), 0, 562 MEMORY_FACTOR_CONFIG_NAME + " failed to propagate."); 563 } 564 565 @TestTemplate 566 public void testInvalidAcceptFactorConfig() throws IOException { 567 float[] configValues = { -1f, 0.2f, 0.86f, 1.05f }; 568 boolean[] expectedOutcomes = { false, false, true, false }; 569 Map<String, float[]> configMappings = ImmutableMap.of(ACCEPT_FACTOR_CONFIG_NAME, configValues); 570 Configuration conf = HBaseConfiguration.create(); 571 checkConfigValues(conf, configMappings, expectedOutcomes); 572 } 573 574 @TestTemplate 575 public void testInvalidMinFactorConfig() throws IOException { 576 float[] configValues = { -1f, 0f, 0.96f, 1.05f }; 577 // throws due to <0, in expected range, minFactor > acceptableFactor, > 1.0 578 boolean[] expectedOutcomes = { false, true, false, false }; 579 Map<String, float[]> configMappings = ImmutableMap.of(MIN_FACTOR_CONFIG_NAME, configValues); 580 Configuration conf = HBaseConfiguration.create(); 581 checkConfigValues(conf, configMappings, expectedOutcomes); 582 } 583 584 @TestTemplate 585 public void testInvalidExtraFreeFactorConfig() throws IOException { 586 float[] configValues = { -1f, 0f, 0.2f, 1.05f }; 587 // throws due to <0, in expected range, in expected range, config can be > 1.0 588 boolean[] expectedOutcomes = { false, true, true, true }; 589 Map<String, float[]> configMappings = 590 ImmutableMap.of(EXTRA_FREE_FACTOR_CONFIG_NAME, configValues); 591 Configuration conf = HBaseConfiguration.create(); 592 checkConfigValues(conf, configMappings, expectedOutcomes); 593 } 594 595 @TestTemplate 596 public void testInvalidCacheSplitFactorConfig() throws IOException { 597 float[] singleFactorConfigValues = { 0.2f, 0f, -0.2f, 1f }; 598 float[] multiFactorConfigValues = { 0.4f, 0f, 1f, .05f }; 599 float[] memoryFactorConfigValues = { 0.4f, 0f, 0.2f, .5f }; 600 // All configs add up to 1.0 and are between 0 and 1.0, configs don't add to 1.0, configs can't 601 // be negative, configs don't add to 1.0 602 boolean[] expectedOutcomes = { true, false, false, false }; 603 Map<String, 604 float[]> configMappings = ImmutableMap.of(SINGLE_FACTOR_CONFIG_NAME, singleFactorConfigValues, 605 MULTI_FACTOR_CONFIG_NAME, multiFactorConfigValues, MEMORY_FACTOR_CONFIG_NAME, 606 memoryFactorConfigValues); 607 Configuration conf = HBaseConfiguration.create(); 608 checkConfigValues(conf, configMappings, expectedOutcomes); 609 } 610 611 private void checkConfigValues(Configuration conf, Map<String, float[]> configMap, 612 boolean[] expectSuccess) throws IOException { 613 Set<String> configNames = configMap.keySet(); 614 for (int i = 0; i < expectSuccess.length; i++) { 615 try { 616 for (String configName : configNames) { 617 conf.setFloat(configName, configMap.get(configName)[i]); 618 } 619 BucketCache cache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 620 constructedBlockSizes, writeThreads, writerQLen, null, 100, conf); 621 assertTrue(cache.waitForCacheInitialization(10000)); 622 assertTrue(expectSuccess[i], "Created BucketCache and expected it to succeed: " 623 + expectSuccess[i] + ", but it actually was: " + !expectSuccess[i]); 624 } catch (IllegalArgumentException e) { 625 assertFalse(expectSuccess[i], "Created BucketCache and expected it to succeed: " 626 + expectSuccess[i] + ", but it actually was: " + !expectSuccess[i]); 627 } 628 } 629 } 630 631 private void validateGetPartitionSize(BucketCache bucketCache, float partitionFactor, 632 float minFactor) { 633 long expectedOutput = 634 (long) Math.floor(bucketCache.getAllocator().getTotalSize() * partitionFactor * minFactor); 635 assertEquals(expectedOutput, bucketCache.getPartitionSize(partitionFactor)); 636 } 637 638 @TestTemplate 639 public void testOffsetProducesPositiveOutput() { 640 // This number is picked because it produces negative output if the values isn't ensured to be 641 // positive. See HBASE-18757 for more information. 642 long testValue = 549888460800L; 643 BucketEntry bucketEntry = new BucketEntry(testValue, 10, 10, 10L, true, (entry) -> { 644 return ByteBuffAllocator.NONE; 645 }, ByteBuffAllocator.HEAP); 646 assertEquals(testValue, bucketEntry.offset()); 647 } 648 649 @TestTemplate 650 public void testEvictionCount() throws InterruptedException { 651 int size = 100; 652 int length = HConstants.HFILEBLOCK_HEADER_SIZE + size; 653 ByteBuffer buf1 = ByteBuffer.allocate(size), buf2 = ByteBuffer.allocate(size); 654 HFileContext meta = new HFileContextBuilder().build(); 655 ByteBuffAllocator allocator = ByteBuffAllocator.HEAP; 656 HFileBlock blockWithNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1, 657 ByteBuff.wrap(buf1), HFileBlock.FILL_HEADER, -1, 52, -1, meta, allocator); 658 HFileBlock blockWithoutNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1, 659 ByteBuff.wrap(buf2), HFileBlock.FILL_HEADER, -1, -1, -1, meta, allocator); 660 661 BlockCacheKey key = new BlockCacheKey("testEvictionCount", 0); 662 ByteBuffer actualBuffer = ByteBuffer.allocate(length); 663 ByteBuffer block1Buffer = ByteBuffer.allocate(length); 664 ByteBuffer block2Buffer = ByteBuffer.allocate(length); 665 blockWithNextBlockMetadata.serialize(block1Buffer, true); 666 blockWithoutNextBlockMetadata.serialize(block2Buffer, true); 667 668 // Add blockWithNextBlockMetadata, expect blockWithNextBlockMetadata back. 669 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer, 670 block1Buffer); 671 672 waitUntilFlushedToBucket(cache, key); 673 674 assertEquals(0, cache.getStats().getEvictionCount()); 675 676 // evict call should return 1, but then eviction count be 0 677 assertEquals(1, cache.evictBlocksByHfileName("testEvictionCount")); 678 assertEquals(0, cache.getStats().getEvictionCount()); 679 680 // add back 681 key = new BlockCacheKey("testEvictionCount", 0); 682 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer, 683 block1Buffer); 684 waitUntilFlushedToBucket(cache, key); 685 686 // should not increment 687 assertTrue(cache.evictBlock(key)); 688 assertEquals(0, cache.getStats().getEvictionCount()); 689 690 // add back 691 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer, 692 block1Buffer); 693 waitUntilFlushedToBucket(cache, key); 694 695 // should finally increment eviction count 696 cache.freeSpace("testing"); 697 assertEquals(1, cache.getStats().getEvictionCount()); 698 } 699 700 @TestTemplate 701 public void testCacheBlockNextBlockMetadataMissing() throws Exception { 702 int size = 100; 703 int length = HConstants.HFILEBLOCK_HEADER_SIZE + size; 704 ByteBuffer buf1 = ByteBuffer.allocate(size), buf2 = ByteBuffer.allocate(size); 705 HFileContext meta = new HFileContextBuilder().build(); 706 ByteBuffAllocator allocator = ByteBuffAllocator.HEAP; 707 HFileBlock blockWithNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1, 708 ByteBuff.wrap(buf1), HFileBlock.FILL_HEADER, -1, 52, -1, meta, allocator); 709 HFileBlock blockWithoutNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1, 710 ByteBuff.wrap(buf2), HFileBlock.FILL_HEADER, -1, -1, -1, meta, allocator); 711 712 BlockCacheKey key = new BlockCacheKey("testCacheBlockNextBlockMetadataMissing", 0); 713 ByteBuffer actualBuffer = ByteBuffer.allocate(length); 714 ByteBuffer block1Buffer = ByteBuffer.allocate(length); 715 ByteBuffer block2Buffer = ByteBuffer.allocate(length); 716 blockWithNextBlockMetadata.serialize(block1Buffer, true); 717 blockWithoutNextBlockMetadata.serialize(block2Buffer, true); 718 719 // Add blockWithNextBlockMetadata, expect blockWithNextBlockMetadata back. 720 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer, 721 block1Buffer); 722 723 waitUntilFlushedToBucket(cache, key); 724 assertNotNull(cache.backingMap.get(key)); 725 assertEquals(1, cache.backingMap.get(key).refCnt()); 726 assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt()); 727 assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt()); 728 729 // Add blockWithoutNextBlockMetada, expect blockWithNextBlockMetadata back. 730 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithoutNextBlockMetadata, actualBuffer, 731 block1Buffer); 732 assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt()); 733 assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt()); 734 assertEquals(1, cache.backingMap.get(key).refCnt()); 735 736 // Clear and add blockWithoutNextBlockMetadata 737 assertTrue(cache.evictBlock(key)); 738 assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt()); 739 assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt()); 740 741 assertNull(cache.getBlock(key, false, false, false)); 742 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithoutNextBlockMetadata, actualBuffer, 743 block2Buffer); 744 745 waitUntilFlushedToBucket(cache, key); 746 assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt()); 747 assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt()); 748 749 // Add blockWithNextBlockMetadata, expect blockWithNextBlockMetadata to replace. 750 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer, 751 block1Buffer); 752 753 waitUntilFlushedToBucket(cache, key); 754 assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt()); 755 assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt()); 756 } 757 758 @TestTemplate 759 public void testRAMCache() { 760 int size = 100; 761 int length = HConstants.HFILEBLOCK_HEADER_SIZE + size; 762 byte[] byteArr = new byte[length]; 763 ByteBuffer buf = ByteBuffer.wrap(byteArr, 0, size); 764 HFileContext meta = new HFileContextBuilder().build(); 765 766 RAMCache cache = new RAMCache(); 767 BlockCacheKey key1 = new BlockCacheKey("file-1", 1); 768 BlockCacheKey key2 = new BlockCacheKey("file-2", 2); 769 HFileBlock blk1 = new HFileBlock(BlockType.DATA, size, size, -1, ByteBuff.wrap(buf), 770 HFileBlock.FILL_HEADER, -1, 52, -1, meta, ByteBuffAllocator.HEAP); 771 HFileBlock blk2 = new HFileBlock(BlockType.DATA, size, size, -1, ByteBuff.wrap(buf), 772 HFileBlock.FILL_HEADER, -1, -1, -1, meta, ByteBuffAllocator.HEAP); 773 RAMQueueEntry re1 = new RAMQueueEntry(key1, blk1, 1, false, false, false); 774 RAMQueueEntry re2 = new RAMQueueEntry(key1, blk2, 1, false, false, false); 775 776 assertFalse(cache.containsKey(key1)); 777 assertNull(cache.putIfAbsent(key1, re1)); 778 assertEquals(2, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt()); 779 780 assertNotNull(cache.putIfAbsent(key1, re2)); 781 assertEquals(2, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt()); 782 assertEquals(1, ((HFileBlock) re2.getData()).getBufferReadOnly().refCnt()); 783 784 assertNull(cache.putIfAbsent(key2, re2)); 785 assertEquals(2, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt()); 786 assertEquals(2, ((HFileBlock) re2.getData()).getBufferReadOnly().refCnt()); 787 788 cache.remove(key1); 789 assertEquals(1, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt()); 790 assertEquals(2, ((HFileBlock) re2.getData()).getBufferReadOnly().refCnt()); 791 792 cache.clear(); 793 assertEquals(1, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt()); 794 assertEquals(1, ((HFileBlock) re2.getData()).getBufferReadOnly().refCnt()); 795 } 796 797 @TestTemplate 798 public void testFreeBlockWhenIOEngineWriteFailure() throws IOException { 799 // initialize an block. 800 int size = 100, offset = 20; 801 int length = HConstants.HFILEBLOCK_HEADER_SIZE + size; 802 ByteBuffer buf = ByteBuffer.allocate(length); 803 HFileContext meta = new HFileContextBuilder().build(); 804 HFileBlock block = new HFileBlock(BlockType.DATA, size, size, -1, ByteBuff.wrap(buf), 805 HFileBlock.FILL_HEADER, offset, 52, -1, meta, ByteBuffAllocator.HEAP); 806 807 // initialize an mocked ioengine. 808 IOEngine ioEngine = Mockito.mock(IOEngine.class); 809 when(ioEngine.usesSharedMemory()).thenReturn(false); 810 // Mockito.doNothing().when(ioEngine).write(Mockito.any(ByteBuffer.class), Mockito.anyLong()); 811 Mockito.doThrow(RuntimeException.class).when(ioEngine).write(Mockito.any(ByteBuffer.class), 812 Mockito.anyLong()); 813 Mockito.doThrow(RuntimeException.class).when(ioEngine).write(Mockito.any(ByteBuff.class), 814 Mockito.anyLong()); 815 816 // create an bucket allocator. 817 long availableSpace = 1024 * 1024 * 1024L; 818 BucketAllocator allocator = new BucketAllocator(availableSpace, null); 819 820 BlockCacheKey key = new BlockCacheKey("dummy", 1L); 821 RAMQueueEntry re = new RAMQueueEntry(key, block, 1, true, false, false); 822 823 assertEquals(0, allocator.getUsedSize()); 824 try { 825 re.writeToCache(ioEngine, allocator, null, null, 826 ByteBuffer.allocate(HFileBlock.BLOCK_METADATA_SPACE), Long.MAX_VALUE); 827 fail(); 828 } catch (Exception e) { 829 } 830 assertEquals(0, allocator.getUsedSize()); 831 } 832 833 /** 834 * This test is for HBASE-26295, {@link BucketEntry} which is restored from a persistence file 835 * could not be freed even if corresponding {@link HFileBlock} is evicted from 836 * {@link BucketCache}. 837 */ 838 @TestTemplate 839 public void testFreeBucketEntryRestoredFromFile() throws Exception { 840 BucketCache bucketCache = null; 841 try { 842 final Path dataTestDir = createAndGetTestDir(); 843 844 String ioEngineName = "file:" + dataTestDir + "/bucketNoRecycler.cache"; 845 String persistencePath = dataTestDir + "/bucketNoRecycler.persistence"; 846 847 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 848 constructedBlockSizes, writeThreads, writerQLen, persistencePath); 849 assertTrue(bucketCache.waitForCacheInitialization(10000)); 850 long usedByteSize = bucketCache.getAllocator().getUsedSize(); 851 assertEquals(0, usedByteSize); 852 853 HFileBlockPair[] hfileBlockPairs = 854 CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1); 855 // Add blocks 856 for (HFileBlockPair hfileBlockPair : hfileBlockPairs) { 857 bucketCache.cacheBlock(hfileBlockPair.getBlockName(), hfileBlockPair.getBlock()); 858 } 859 860 for (HFileBlockPair hfileBlockPair : hfileBlockPairs) { 861 cacheAndWaitUntilFlushedToBucket(bucketCache, hfileBlockPair.getBlockName(), 862 hfileBlockPair.getBlock(), false); 863 } 864 usedByteSize = bucketCache.getAllocator().getUsedSize(); 865 assertNotEquals(0, usedByteSize); 866 // persist cache to file 867 bucketCache.shutdown(); 868 assertTrue(new File(persistencePath).exists()); 869 870 // restore cache from file 871 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 872 constructedBlockSizes, writeThreads, writerQLen, persistencePath); 873 assertTrue(bucketCache.waitForCacheInitialization(10000)); 874 assertEquals(usedByteSize, bucketCache.getAllocator().getUsedSize()); 875 876 for (HFileBlockPair hfileBlockPair : hfileBlockPairs) { 877 BlockCacheKey blockCacheKey = hfileBlockPair.getBlockName(); 878 bucketCache.evictBlock(blockCacheKey); 879 } 880 assertEquals(0, bucketCache.getAllocator().getUsedSize()); 881 assertEquals(0, bucketCache.backingMap.size()); 882 } finally { 883 bucketCache.shutdown(); 884 HBASE_TESTING_UTILITY.cleanupTestDir(); 885 } 886 } 887 888 @TestTemplate 889 public void testBlockAdditionWaitWhenCache() throws Exception { 890 BucketCache bucketCache = null; 891 try { 892 final Path dataTestDir = createAndGetTestDir(); 893 894 String ioEngineName = "file:" + dataTestDir + "/bucketNoRecycler.cache"; 895 String persistencePath = dataTestDir + "/bucketNoRecycler.persistence"; 896 897 Configuration config = HBASE_TESTING_UTILITY.getConfiguration(); 898 config.setLong(QUEUE_ADDITION_WAIT_TIME, 1000); 899 900 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 901 constructedBlockSizes, 1, 1, persistencePath, DEFAULT_ERROR_TOLERATION_DURATION, config); 902 assertTrue(bucketCache.waitForCacheInitialization(10000)); 903 long usedByteSize = bucketCache.getAllocator().getUsedSize(); 904 assertEquals(0, usedByteSize); 905 906 HFileBlockPair[] hfileBlockPairs = 907 CacheTestUtils.generateHFileBlocks(constructedBlockSize, 10); 908 String[] names = CacheTestUtils.getHFileNames(hfileBlockPairs); 909 // Add blocks 910 for (HFileBlockPair hfileBlockPair : hfileBlockPairs) { 911 bucketCache.cacheBlock(hfileBlockPair.getBlockName(), hfileBlockPair.getBlock(), false, 912 true); 913 } 914 915 // Max wait for 10 seconds. 916 long timeout = 10000; 917 // Wait for blocks size to match the number of blocks. 918 while (bucketCache.backingMap.size() != 10) { 919 if (timeout <= 0) break; 920 Threads.sleep(100); 921 timeout -= 100; 922 } 923 for (HFileBlockPair hfileBlockPair : hfileBlockPairs) { 924 assertTrue(bucketCache.backingMap.containsKey(hfileBlockPair.getBlockName())); 925 } 926 usedByteSize = bucketCache.getAllocator().getUsedSize(); 927 assertNotEquals(0, usedByteSize); 928 // persist cache to file 929 bucketCache.shutdown(); 930 assertTrue(new File(persistencePath).exists()); 931 932 // restore cache from file 933 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 934 constructedBlockSizes, writeThreads, writerQLen, persistencePath); 935 assertTrue(bucketCache.waitForCacheInitialization(10000)); 936 assertEquals(usedByteSize, bucketCache.getAllocator().getUsedSize()); 937 BlockCacheKey[] newKeys = CacheTestUtils.regenerateKeys(hfileBlockPairs, names); 938 for (BlockCacheKey key : newKeys) { 939 bucketCache.evictBlock(key); 940 } 941 assertEquals(0, bucketCache.getAllocator().getUsedSize()); 942 assertEquals(0, bucketCache.backingMap.size()); 943 } finally { 944 if (bucketCache != null) { 945 bucketCache.shutdown(); 946 } 947 HBASE_TESTING_UTILITY.cleanupTestDir(); 948 } 949 } 950 951 @TestTemplate 952 public void testOnConfigurationChange() throws Exception { 953 BucketCache bucketCache = null; 954 try { 955 final Path dataTestDir = createAndGetTestDir(); 956 957 String ioEngineName = "file:" + dataTestDir + "/bucketNoRecycler.cache"; 958 959 Configuration config = HBASE_TESTING_UTILITY.getConfiguration(); 960 961 bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 962 constructedBlockSizes, 1, 1, null, DEFAULT_ERROR_TOLERATION_DURATION, config); 963 964 assertTrue(bucketCache.waitForCacheInitialization(10000)); 965 966 config.setFloat(ACCEPT_FACTOR_CONFIG_NAME, 0.9f); 967 config.setFloat(MIN_FACTOR_CONFIG_NAME, 0.8f); 968 config.setFloat(EXTRA_FREE_FACTOR_CONFIG_NAME, 0.15f); 969 config.setFloat(SINGLE_FACTOR_CONFIG_NAME, 0.2f); 970 config.setFloat(MULTI_FACTOR_CONFIG_NAME, 0.6f); 971 config.setFloat(MEMORY_FACTOR_CONFIG_NAME, 0.2f); 972 config.setLong(QUEUE_ADDITION_WAIT_TIME, 100); 973 config.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, 500); 974 config.setLong(BACKING_MAP_PERSISTENCE_CHUNK_SIZE, 1000); 975 976 bucketCache.onConfigurationChange(config); 977 978 assertEquals(0.9f, bucketCache.getAcceptableFactor(), 0.01); 979 assertEquals(0.8f, bucketCache.getMinFactor(), 0.01); 980 assertEquals(0.15f, bucketCache.getExtraFreeFactor(), 0.01); 981 assertEquals(0.2f, bucketCache.getSingleFactor(), 0.01); 982 assertEquals(0.6f, bucketCache.getMultiFactor(), 0.01); 983 assertEquals(0.2f, bucketCache.getMemoryFactor(), 0.01); 984 assertEquals(100L, bucketCache.getQueueAdditionWaitTime()); 985 assertEquals(500L, bucketCache.getBucketcachePersistInterval()); 986 assertEquals(1000L, bucketCache.getPersistenceChunkSize()); 987 988 } finally { 989 if (bucketCache != null) { 990 bucketCache.shutdown(); 991 } 992 HBASE_TESTING_UTILITY.cleanupTestDir(); 993 } 994 } 995 996 @TestTemplate 997 public void testNotifyFileCachingCompletedSuccess() throws Exception { 998 BucketCache bucketCache = null; 999 try { 1000 Path filePath = 1001 new Path(HBASE_TESTING_UTILITY.getDataTestDir(), "testNotifyFileCachingCompletedSuccess"); 1002 bucketCache = testNotifyFileCachingCompletedForTenBlocks(filePath, 10, false); 1003 if (bucketCache.getStats().getFailedInserts() > 0) { 1004 LOG.info("There were {} fail inserts, " 1005 + "will assert if total blocks in backingMap equals (10 - failInserts) " 1006 + "and file isn't listed as fully cached.", bucketCache.getStats().getFailedInserts()); 1007 assertEquals(10 - bucketCache.getStats().getFailedInserts(), bucketCache.backingMap.size()); 1008 assertFalse(bucketCache.fullyCachedFiles.containsKey(filePath.getName())); 1009 } else { 1010 assertTrue(bucketCache.fullyCachedFiles.containsKey(filePath.getName())); 1011 } 1012 } finally { 1013 if (bucketCache != null) { 1014 bucketCache.shutdown(); 1015 } 1016 HBASE_TESTING_UTILITY.cleanupTestDir(); 1017 } 1018 } 1019 1020 @TestTemplate 1021 public void testNotifyFileCachingCompletedForEncodedDataSuccess() throws Exception { 1022 BucketCache bucketCache = null; 1023 try { 1024 Path filePath = new Path(HBASE_TESTING_UTILITY.getDataTestDir(), 1025 "testNotifyFileCachingCompletedForEncodedDataSuccess"); 1026 bucketCache = testNotifyFileCachingCompletedForTenBlocks(filePath, 10, true); 1027 if (bucketCache.getStats().getFailedInserts() > 0) { 1028 LOG.info("There were {} fail inserts, " 1029 + "will assert if total blocks in backingMap equals (10 - failInserts) " 1030 + "and file isn't listed as fully cached.", bucketCache.getStats().getFailedInserts()); 1031 assertEquals(10 - bucketCache.getStats().getFailedInserts(), bucketCache.backingMap.size()); 1032 assertFalse(bucketCache.fullyCachedFiles.containsKey(filePath.getName())); 1033 } else { 1034 assertTrue(bucketCache.fullyCachedFiles.containsKey(filePath.getName())); 1035 } 1036 } finally { 1037 if (bucketCache != null) { 1038 bucketCache.shutdown(); 1039 } 1040 HBASE_TESTING_UTILITY.cleanupTestDir(); 1041 } 1042 } 1043 1044 @TestTemplate 1045 public void testNotifyFileCachingCompletedNotAllCached() throws Exception { 1046 BucketCache bucketCache = null; 1047 try { 1048 Path filePath = new Path(HBASE_TESTING_UTILITY.getDataTestDir(), 1049 "testNotifyFileCachingCompletedNotAllCached"); 1050 // Deliberately passing more blocks than we have created to test that 1051 // notifyFileCachingCompleted will not consider the file fully cached 1052 bucketCache = testNotifyFileCachingCompletedForTenBlocks(filePath, 12, false); 1053 assertFalse(bucketCache.fullyCachedFiles.containsKey(filePath.getName())); 1054 } finally { 1055 if (bucketCache != null) { 1056 bucketCache.shutdown(); 1057 } 1058 HBASE_TESTING_UTILITY.cleanupTestDir(); 1059 } 1060 } 1061 1062 private BucketCache testNotifyFileCachingCompletedForTenBlocks(Path filePath, 1063 int totalBlocksToCheck, boolean encoded) throws Exception { 1064 final Path dataTestDir = createAndGetTestDir(); 1065 String ioEngineName = "file:" + dataTestDir + "/bucketNoRecycler.cache"; 1066 BucketCache bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize, 1067 constructedBlockSizes, 1, 1, null); 1068 assertTrue(bucketCache.waitForCacheInitialization(10000)); 1069 long usedByteSize = bucketCache.getAllocator().getUsedSize(); 1070 assertEquals(0, usedByteSize); 1071 HFileBlockPair[] hfileBlockPairs = 1072 CacheTestUtils.generateBlocksForPath(constructedBlockSize, 10, filePath, encoded); 1073 // Add blocks 1074 for (HFileBlockPair hfileBlockPair : hfileBlockPairs) { 1075 bucketCache.cacheBlock(hfileBlockPair.getBlockName(), hfileBlockPair.getBlock(), false, true); 1076 } 1077 bucketCache.notifyFileCachingCompleted(filePath, totalBlocksToCheck, totalBlocksToCheck, 1078 totalBlocksToCheck * constructedBlockSize); 1079 return bucketCache; 1080 } 1081 1082 @TestTemplate 1083 public void testEvictOrphansOutOfGracePeriod() throws Exception { 1084 BucketCache bucketCache = testEvictOrphans(0); 1085 assertEquals(10, bucketCache.getBackingMap().size()); 1086 assertEquals(0, bucketCache.blocksByHFile.stream() 1087 .filter(key -> key.getHfileName().equals("testEvictOrphans-orphan")).count()); 1088 } 1089 1090 @TestTemplate 1091 public void testEvictOrphansWithinGracePeriod() throws Exception { 1092 BucketCache bucketCache = testEvictOrphans(60 * 60 * 1000L); 1093 assertEquals(18, bucketCache.getBackingMap().size()); 1094 assertTrue(bucketCache.blocksByHFile.stream() 1095 .filter(key -> key.getHfileName().equals("testEvictOrphans-orphan")).count() > 0); 1096 } 1097 1098 private BucketCache testEvictOrphans(long orphanEvictionGracePeriod) throws Exception { 1099 Path validFile = new Path(HBASE_TESTING_UTILITY.getDataTestDir(), "testEvictOrphans-valid"); 1100 Path orphanFile = new Path(HBASE_TESTING_UTILITY.getDataTestDir(), "testEvictOrphans-orphan"); 1101 Map<String, HRegion> onlineRegions = new HashMap<>(); 1102 List<HStore> stores = new ArrayList<>(); 1103 Collection<HStoreFile> storeFiles = new ArrayList<>(); 1104 HRegion mockedRegion = mock(HRegion.class); 1105 HStore mockedStore = mock(HStore.class); 1106 HStoreFile mockedStoreFile = mock(HStoreFile.class); 1107 when(mockedStoreFile.getPath()).thenReturn(validFile); 1108 storeFiles.add(mockedStoreFile); 1109 when(mockedStore.getStorefiles()).thenReturn(storeFiles); 1110 stores.add(mockedStore); 1111 when(mockedRegion.getStores()).thenReturn(stores); 1112 onlineRegions.put("mocked_region", mockedRegion); 1113 HBASE_TESTING_UTILITY.getConfiguration().setDouble(MIN_FACTOR_CONFIG_NAME, 0.99); 1114 HBASE_TESTING_UTILITY.getConfiguration().setDouble(ACCEPT_FACTOR_CONFIG_NAME, 1); 1115 HBASE_TESTING_UTILITY.getConfiguration().setDouble(EXTRA_FREE_FACTOR_CONFIG_NAME, 0.01); 1116 HBASE_TESTING_UTILITY.getConfiguration().setLong(BLOCK_ORPHAN_GRACE_PERIOD, 1117 orphanEvictionGracePeriod); 1118 BucketCache bucketCache = new BucketCache(ioEngineName, (constructedBlockSize + 1024) * 21, 1119 constructedBlockSize, new int[] { constructedBlockSize + 1024 }, 1, 1, null, 60 * 1000, 1120 HBASE_TESTING_UTILITY.getConfiguration(), onlineRegions); 1121 HFileBlockPair[] validBlockPairs = 1122 CacheTestUtils.generateBlocksForPath(constructedBlockSize, 10, validFile, false); 1123 HFileBlockPair[] orphanBlockPairs = 1124 CacheTestUtils.generateBlocksForPath(constructedBlockSize, 10, orphanFile, false); 1125 for (HFileBlockPair pair : validBlockPairs) { 1126 bucketCache.cacheBlockWithWait(pair.getBlockName(), pair.getBlock(), false, true); 1127 } 1128 waitUntilAllFlushedToBucket(bucketCache); 1129 assertEquals(10, bucketCache.getBackingMap().size()); 1130 bucketCache.freeSpace("test"); 1131 assertEquals(10, bucketCache.getBackingMap().size()); 1132 for (HFileBlockPair pair : orphanBlockPairs) { 1133 bucketCache.cacheBlockWithWait(pair.getBlockName(), pair.getBlock(), false, true); 1134 } 1135 waitUntilAllFlushedToBucket(bucketCache); 1136 assertEquals(20, bucketCache.getBackingMap().size()); 1137 bucketCache.freeSpace("test"); 1138 return bucketCache; 1139 } 1140 1141 @TestTemplate 1142 public void testBlockPriority() throws Exception { 1143 HFileBlockPair block = CacheTestUtils.generateHFileBlocks(BLOCK_SIZE, 1)[0]; 1144 cacheAndWaitUntilFlushedToBucket(cache, block.getBlockName(), block.getBlock(), true); 1145 assertEquals(cache.backingMap.get(block.getBlockName()).getPriority(), BlockPriority.SINGLE); 1146 cache.getBlock(block.getBlockName(), true, false, true); 1147 assertEquals(cache.backingMap.get(block.getBlockName()).getPriority(), BlockPriority.MULTI); 1148 } 1149 1150 @TestTemplate 1151 public void testIOTimePerHitReturnsZeroWhenNoHits() 1152 throws NoSuchFieldException, IllegalAccessException { 1153 CacheStats cacheStats = cache.getStats(); 1154 assertTrue(cacheStats instanceof BucketCacheStats); 1155 BucketCacheStats bucketCacheStats = (BucketCacheStats) cacheStats; 1156 1157 Field field = BucketCacheStats.class.getDeclaredField("ioHitCount"); 1158 field.setAccessible(true); 1159 LongAdder ioHitCount = (LongAdder) field.get(bucketCacheStats); 1160 1161 assertEquals(0, ioHitCount.sum()); 1162 double ioTimePerHit = bucketCacheStats.getIOTimePerHit(); 1163 assertEquals(0, ioTimePerHit, 0.0); 1164 } 1165}