1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.io.hfile.bucket;
20
21 import static org.junit.Assert.assertEquals;
22 import static org.junit.Assert.assertFalse;
23 import static org.junit.Assert.assertNull;
24 import static org.junit.Assert.assertTrue;
25
26 import java.io.FileNotFoundException;
27 import java.io.IOException;
28 import java.nio.ByteBuffer;
29 import java.util.ArrayList;
30 import java.util.Arrays;
31 import java.util.List;
32 import java.util.Map;
33 import java.util.Random;
34 import java.util.Set;
35 import java.util.concurrent.locks.ReentrantReadWriteLock;
36
37 import com.google.common.collect.ImmutableMap;
38 import org.apache.hadoop.conf.Configuration;
39 import org.apache.hadoop.fs.Path;
40 import org.apache.hadoop.hbase.HBaseConfiguration;
41 import org.apache.hadoop.hbase.HBaseTestingUtility;
42 import org.apache.hadoop.hbase.HConstants;
43 import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
44 import org.apache.hadoop.hbase.io.hfile.BlockType;
45 import org.apache.hadoop.hbase.io.hfile.CacheTestUtils;
46 import org.apache.hadoop.hbase.io.hfile.CacheTestUtils.HFileBlockPair;
47 import org.apache.hadoop.hbase.io.hfile.Cacheable;
48 import org.apache.hadoop.hbase.io.hfile.HFileBlock;
49 import org.apache.hadoop.hbase.io.hfile.HFileContext;
50 import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
51 import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.BucketSizeInfo;
52 import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.IndexStatistics;
53 import org.apache.hadoop.hbase.testclassification.SmallTests;
54 import org.junit.After;
55 import org.junit.Before;
56 import org.junit.Test;
57 import org.junit.experimental.categories.Category;
58 import org.junit.runner.RunWith;
59 import org.junit.runners.Parameterized;
60
61
62
63
64
65
66 @RunWith(Parameterized.class)
67 @Category(SmallTests.class)
68 public class TestBucketCache {
69
70 private static final Random RAND = new Random();
71
72 @Parameterized.Parameters(name = "{index}: blockSize={0}, bucketSizes={1}")
73 public static Iterable<Object[]> data() {
74 return Arrays.asList(new Object[][] {
75 { 8192, null },
76 {
77 16 * 1024,
78 new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024, 8 * 1024 + 1024, 16 * 1024 + 1024,
79 28 * 1024 + 1024, 32 * 1024 + 1024, 64 * 1024 + 1024, 96 * 1024 + 1024,
80 128 * 1024 + 1024 } } });
81 }
82
83 @Parameterized.Parameter(0)
84 public int constructedBlockSize;
85
86 @Parameterized.Parameter(1)
87 public int[] constructedBlockSizes;
88
89 BucketCache cache;
90 final int CACHE_SIZE = 1000000;
91 final int NUM_BLOCKS = 100;
92 final int BLOCK_SIZE = CACHE_SIZE / NUM_BLOCKS;
93 final int NUM_THREADS = 100;
94 final int NUM_QUERIES = 10000;
95
96 final long capacitySize = 32 * 1024 * 1024;
97 final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS;
98 final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS;
99 String ioEngineName = "heap";
100
101 private class MockedBucketCache extends BucketCache {
102
103 public MockedBucketCache(String ioEngineName, long capacity, int blockSize, int[] bucketSizes,
104 int writerThreads, int writerQLen, String persistencePath) throws FileNotFoundException,
105 IOException {
106 super(ioEngineName, capacity, blockSize, bucketSizes, writerThreads, writerQLen,
107 persistencePath);
108 super.wait_when_cache = true;
109 }
110
111 @Override
112 public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
113 boolean cacheDataInL1) {
114 super.cacheBlock(cacheKey, buf, inMemory, cacheDataInL1);
115 }
116
117 @Override
118 public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
119 super.cacheBlock(cacheKey, buf);
120 }
121 }
122
123 @Before
124 public void setup() throws FileNotFoundException, IOException {
125 cache =
126 new MockedBucketCache(ioEngineName, capacitySize, constructedBlockSize,
127 constructedBlockSizes, writeThreads, writerQLen, null);
128 }
129
130 @After
131 public void tearDown() {
132 cache.shutdown();
133 }
134
135
136
137
138 private static <T> T randFrom(List<T> a) {
139 return a.get(RAND.nextInt(a.size()));
140 }
141
142 @Test
143 public void testBucketAllocator() throws BucketAllocatorException {
144 BucketAllocator mAllocator = cache.getAllocator();
145
146
147
148 final List<Integer> BLOCKSIZES = Arrays.asList(4 * 1024, 8 * 1024, 64 * 1024, 96 * 1024);
149
150 boolean full = false;
151 ArrayList<Long> allocations = new ArrayList<Long>();
152
153
154 List<Integer> tmp = new ArrayList<Integer>(BLOCKSIZES);
155 while (!full) {
156 Integer blockSize = null;
157 try {
158 blockSize = randFrom(tmp);
159 allocations.add(mAllocator.allocateBlock(blockSize));
160 } catch (CacheFullException cfe) {
161 tmp.remove(blockSize);
162 if (tmp.isEmpty()) full = true;
163 }
164 }
165
166 for (Integer blockSize : BLOCKSIZES) {
167 BucketSizeInfo bucketSizeInfo = mAllocator.roundUpToBucketSizeInfo(blockSize);
168 IndexStatistics indexStatistics = bucketSizeInfo.statistics();
169 assertEquals("unexpected freeCount for " + bucketSizeInfo, 0, indexStatistics.freeCount());
170 }
171
172 for (long offset : allocations) {
173 assertEquals(mAllocator.sizeOfAllocation(offset), mAllocator.freeBlock(offset));
174 }
175 assertEquals(0, mAllocator.getUsedSize());
176 }
177
178 @Test
179 public void testCacheSimple() throws Exception {
180 CacheTestUtils.testCacheSimple(cache, BLOCK_SIZE, NUM_QUERIES);
181 }
182
183 @Test
184 public void testCacheMultiThreadedSingleKey() throws Exception {
185 CacheTestUtils.hammerSingleKey(cache, BLOCK_SIZE, 2 * NUM_THREADS, 2 * NUM_QUERIES);
186 }
187
188 @Test
189 public void testHeapSizeChanges() throws Exception {
190 cache.stopWriterThreads();
191 CacheTestUtils.testHeapSizeChanges(cache, BLOCK_SIZE);
192 }
193
194 private void waitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey)
195 throws InterruptedException {
196 while (!cache.backingMap.containsKey(cacheKey) || cache.ramCache.containsKey(cacheKey)) {
197 Thread.sleep(100);
198 }
199 }
200
201
202
203 private void cacheAndWaitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey,
204 Cacheable block) throws InterruptedException {
205 cache.cacheBlock(cacheKey, block);
206 waitUntilFlushedToBucket(cache, cacheKey);
207 }
208
209 @Test
210 public void testMemoryLeak() throws Exception {
211 final BlockCacheKey cacheKey = new BlockCacheKey("dummy", 1L);
212 cacheAndWaitUntilFlushedToBucket(cache, cacheKey, new CacheTestUtils.ByteArrayCacheable(
213 new byte[10]));
214 long lockId = cache.backingMap.get(cacheKey).offset();
215 ReentrantReadWriteLock lock = cache.offsetLock.getLock(lockId);
216 lock.writeLock().lock();
217 Thread evictThread = new Thread("evict-block") {
218
219 @Override
220 public void run() {
221 cache.evictBlock(cacheKey);
222 }
223
224 };
225 evictThread.start();
226 cache.offsetLock.waitForWaiters(lockId, 1);
227 cache.blockEvicted(cacheKey, cache.backingMap.remove(cacheKey), true);
228 cacheAndWaitUntilFlushedToBucket(cache, cacheKey, new CacheTestUtils.ByteArrayCacheable(
229 new byte[10]));
230 lock.writeLock().unlock();
231 evictThread.join();
232 assertEquals(1L, cache.getBlockCount());
233 assertTrue(cache.getCurrentSize() > 0L);
234 assertTrue("We should have a block!", cache.iterator().hasNext());
235 }
236
237 @Test
238 public void testRetrieveFromFile() throws Exception {
239 HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
240 Path testDir = TEST_UTIL.getDataTestDir();
241 TEST_UTIL.getTestFileSystem().mkdirs(testDir);
242
243 BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
244 constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen, testDir
245 + "/bucket.persistence");
246 long usedSize = bucketCache.getAllocator().getUsedSize();
247 assertTrue(usedSize == 0);
248
249 HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1);
250
251 for (HFileBlockPair block : blocks) {
252 bucketCache.cacheBlock(block.getBlockName(), block.getBlock());
253 }
254 for (HFileBlockPair block : blocks) {
255 cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock());
256 }
257 usedSize = bucketCache.getAllocator().getUsedSize();
258 assertTrue(usedSize != 0);
259
260 bucketCache.shutdown();
261
262
263 bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
264 constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen, testDir
265 + "/bucket.persistence");
266 assertEquals(usedSize, bucketCache.getAllocator().getUsedSize());
267
268 bucketCache.shutdown();
269
270
271
272 int[] smallBucketSizes = new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024 };
273 bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
274 constructedBlockSize, smallBucketSizes, writeThreads,
275 writerQLen, testDir + "/bucket.persistence");
276 assertEquals(0, bucketCache.getAllocator().getUsedSize());
277 assertEquals(0, bucketCache.backingMap.size());
278
279 TEST_UTIL.cleanupTestDir();
280 }
281
282 @Test
283 public void testGetPartitionSize() throws IOException {
284
285 validateGetPartitionSize(cache, BucketCache.DEFAULT_SINGLE_FACTOR, BucketCache.DEFAULT_MIN_FACTOR);
286
287 Configuration conf = HBaseConfiguration.create();
288 conf.setFloat(BucketCache.MIN_FACTOR_CONFIG_NAME, 0.5f);
289 conf.setFloat(BucketCache.SINGLE_FACTOR_CONFIG_NAME, 0.1f);
290 conf.setFloat(BucketCache.MULTI_FACTOR_CONFIG_NAME, 0.7f);
291 conf.setFloat(BucketCache.MEMORY_FACTOR_CONFIG_NAME, 0.2f);
292
293 BucketCache cache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
294 constructedBlockSizes, writeThreads, writerQLen, null, 100, conf);
295
296 validateGetPartitionSize(cache, 0.1f, 0.5f);
297 validateGetPartitionSize(cache, 0.7f, 0.5f);
298 validateGetPartitionSize(cache, 0.2f, 0.5f);
299 }
300
301 @Test
302 public void testValidBucketCacheConfigs() throws IOException {
303 Configuration conf = HBaseConfiguration.create();
304 conf.setFloat(BucketCache.ACCEPT_FACTOR_CONFIG_NAME, 0.9f);
305 conf.setFloat(BucketCache.MIN_FACTOR_CONFIG_NAME, 0.5f);
306 conf.setFloat(BucketCache.EXTRA_FREE_FACTOR_CONFIG_NAME, 0.5f);
307 conf.setFloat(BucketCache.SINGLE_FACTOR_CONFIG_NAME, 0.1f);
308 conf.setFloat(BucketCache.MULTI_FACTOR_CONFIG_NAME, 0.7f);
309 conf.setFloat(BucketCache.MEMORY_FACTOR_CONFIG_NAME, 0.2f);
310
311 BucketCache cache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
312 constructedBlockSizes, writeThreads, writerQLen, null, 100, conf);
313
314 assertEquals(BucketCache.ACCEPT_FACTOR_CONFIG_NAME + " failed to propagate.", cache.getAcceptableFactor(), 0.9f, 0);
315 assertEquals(BucketCache.MIN_FACTOR_CONFIG_NAME + " failed to propagate.", cache.getMinFactor(), 0.5f, 0);
316 assertEquals(BucketCache.EXTRA_FREE_FACTOR_CONFIG_NAME + " failed to propagate.", cache.getExtraFreeFactor(), 0.5f, 0);
317 assertEquals(BucketCache.SINGLE_FACTOR_CONFIG_NAME + " failed to propagate.", cache.getSingleFactor(), 0.1f, 0);
318 assertEquals(BucketCache.MULTI_FACTOR_CONFIG_NAME + " failed to propagate.", cache.getMultiFactor(), 0.7f, 0);
319 assertEquals(BucketCache.MEMORY_FACTOR_CONFIG_NAME + " failed to propagate.", cache.getMemoryFactor(), 0.2f, 0);
320 }
321
322 @Test
323 public void testInvalidAcceptFactorConfig() throws IOException {
324 float[] configValues = {-1f, 0.2f, 0.86f, 1.05f};
325 boolean[] expectedOutcomes = {false, false, true, false};
326 Map<String, float[]> configMappings = ImmutableMap.of(BucketCache.ACCEPT_FACTOR_CONFIG_NAME, configValues);
327 Configuration conf = HBaseConfiguration.create();
328 checkConfigValues(conf, configMappings, expectedOutcomes);
329 }
330
331 @Test
332 public void testInvalidMinFactorConfig() throws IOException {
333 float[] configValues = {-1f, 0f, 0.96f, 1.05f};
334
335 boolean[] expectedOutcomes = {false, true, false, false};
336 Map<String, float[]> configMappings = ImmutableMap.of(BucketCache.MIN_FACTOR_CONFIG_NAME, configValues);
337 Configuration conf = HBaseConfiguration.create();
338 checkConfigValues(conf, configMappings, expectedOutcomes);
339 }
340
341 @Test
342 public void testInvalidExtraFreeFactorConfig() throws IOException {
343 float[] configValues = {-1f, 0f, 0.2f, 1.05f};
344
345 boolean[] expectedOutcomes = {false, true, true, true};
346 Map<String, float[]> configMappings = ImmutableMap.of(BucketCache.EXTRA_FREE_FACTOR_CONFIG_NAME, configValues);
347 Configuration conf = HBaseConfiguration.create();
348 checkConfigValues(conf, configMappings, expectedOutcomes);
349 }
350
351 @Test
352 public void testInvalidCacheSplitFactorConfig() throws IOException {
353 float[] singleFactorConfigValues = {0.2f, 0f, -0.2f, 1f};
354 float[] multiFactorConfigValues = {0.4f, 0f, 1f, .05f};
355 float[] memoryFactorConfigValues = {0.4f, 0f, 0.2f, .5f};
356
357 boolean[] expectedOutcomes = {true, false, false, false};
358 Map<String, float[]> configMappings = ImmutableMap.of(BucketCache.SINGLE_FACTOR_CONFIG_NAME,
359 singleFactorConfigValues, BucketCache.MULTI_FACTOR_CONFIG_NAME, multiFactorConfigValues,
360 BucketCache.MEMORY_FACTOR_CONFIG_NAME, memoryFactorConfigValues);
361 Configuration conf = HBaseConfiguration.create();
362 checkConfigValues(conf, configMappings, expectedOutcomes);
363 }
364
365 private void checkConfigValues(Configuration conf, Map<String, float[]> configMap, boolean[] expectSuccess) throws IOException {
366 Set<String> configNames = configMap.keySet();
367 for (int i = 0; i < expectSuccess.length; i++) {
368 try {
369 for (String configName : configNames) {
370 conf.setFloat(configName, configMap.get(configName)[i]);
371 }
372 new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
373 constructedBlockSizes, writeThreads, writerQLen, null, 100, conf);
374 assertTrue("Created BucketCache and expected it to succeed: " + expectSuccess[i] + ", but it actually was: " + !expectSuccess[i], expectSuccess[i]);
375 } catch (IllegalArgumentException e) {
376 assertFalse("Created BucketCache and expected it to succeed: " + expectSuccess[i] + ", but it actually was: " + !expectSuccess[i], expectSuccess[i]);
377 }
378 }
379 }
380
381 private void validateGetPartitionSize(BucketCache bucketCache, float partitionFactor, float minFactor) {
382 long expectedOutput = (long) Math.floor(bucketCache.getAllocator().getTotalSize() * partitionFactor * minFactor);
383 assertEquals(expectedOutput, bucketCache.getPartitionSize(partitionFactor));
384 }
385
386 @Test
387 public void testOffsetProducesPositiveOutput() {
388
389
390 long testValue = 549888460800L;
391 BucketCache.BucketEntry bucketEntry = new BucketCache.BucketEntry(testValue, 10, 10L, true);
392 assertEquals(testValue, bucketEntry.offset());
393 }
394
395 @Test
396 public void testCacheBlockNextBlockMetadataMissing() throws Exception {
397 int size = 100;
398 int length = HConstants.HFILEBLOCK_HEADER_SIZE + size;
399 byte[] byteArr = new byte[length];
400 ByteBuffer buf = ByteBuffer.wrap(byteArr, 0, size);
401 HFileContext meta = new HFileContextBuilder().build();
402 HFileBlock blockWithNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1, buf,
403 HFileBlock.FILL_HEADER, -1, 52, -1, meta);
404 HFileBlock blockWithoutNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1, buf,
405 HFileBlock.FILL_HEADER, -1, -1, -1, meta);
406
407 BlockCacheKey key = new BlockCacheKey("key1", 0);
408 ByteBuffer actualBuffer = ByteBuffer.allocate(length);
409 ByteBuffer block1Buffer = ByteBuffer.allocate(length);
410 ByteBuffer block2Buffer = ByteBuffer.allocate(length);
411 blockWithNextBlockMetadata.serialize(block1Buffer, true);
412 blockWithoutNextBlockMetadata.serialize(block2Buffer, true);
413
414
415 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata,
416 actualBuffer, block1Buffer);
417
418 waitUntilFlushedToBucket(cache, key);
419
420 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithoutNextBlockMetadata,
421 actualBuffer, block1Buffer);
422
423
424 cache.evictBlock(key);
425 assertNull(cache.getBlock(key, false, false, false));
426 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithoutNextBlockMetadata,
427 actualBuffer, block2Buffer);
428
429 waitUntilFlushedToBucket(cache, key);
430
431 CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata,
432 actualBuffer, block1Buffer);
433 }
434 }