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.assertTrue;
23
24 import java.io.FileNotFoundException;
25 import java.io.IOException;
26 import java.util.ArrayList;
27 import java.util.Arrays;
28 import java.util.List;
29 import java.util.Random;
30 import java.util.concurrent.locks.ReentrantReadWriteLock;
31
32 import org.apache.hadoop.fs.Path;
33 import org.apache.hadoop.hbase.HBaseTestingUtility;
34 import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
35 import org.apache.hadoop.hbase.io.hfile.CacheTestUtils;
36 import org.apache.hadoop.hbase.io.hfile.CacheTestUtils.HFileBlockPair;
37 import org.apache.hadoop.hbase.io.hfile.Cacheable;
38 import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.BucketSizeInfo;
39 import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.IndexStatistics;
40 import org.apache.hadoop.hbase.testclassification.SmallTests;
41 import org.junit.After;
42 import org.junit.Before;
43 import org.junit.Test;
44 import org.junit.experimental.categories.Category;
45 import org.junit.runner.RunWith;
46 import org.junit.runners.Parameterized;
47
48
49
50
51
52
53 @RunWith(Parameterized.class)
54 @Category(SmallTests.class)
55 public class TestBucketCache {
56
57 private static final Random RAND = new Random();
58
59 @Parameterized.Parameters(name = "{index}: blockSize={0}, bucketSizes={1}")
60 public static Iterable<Object[]> data() {
61 return Arrays.asList(new Object[][] {
62 { 8192, null },
63 {
64 16 * 1024,
65 new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024, 8 * 1024 + 1024, 16 * 1024 + 1024,
66 28 * 1024 + 1024, 32 * 1024 + 1024, 64 * 1024 + 1024, 96 * 1024 + 1024,
67 128 * 1024 + 1024 } } });
68 }
69
70 @Parameterized.Parameter(0)
71 public int constructedBlockSize;
72
73 @Parameterized.Parameter(1)
74 public int[] constructedBlockSizes;
75
76 BucketCache cache;
77 final int CACHE_SIZE = 1000000;
78 final int NUM_BLOCKS = 100;
79 final int BLOCK_SIZE = CACHE_SIZE / NUM_BLOCKS;
80 final int NUM_THREADS = 100;
81 final int NUM_QUERIES = 10000;
82
83 final long capacitySize = 32 * 1024 * 1024;
84 final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS;
85 final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS;
86 String ioEngineName = "heap";
87 String persistencePath = null;
88
89 private class MockedBucketCache extends BucketCache {
90
91 public MockedBucketCache(String ioEngineName, long capacity, int blockSize, int[] bucketSizes,
92 int writerThreads, int writerQLen, String persistencePath) throws FileNotFoundException,
93 IOException {
94 super(ioEngineName, capacity, blockSize, bucketSizes, writerThreads, writerQLen,
95 persistencePath);
96 super.wait_when_cache = true;
97 }
98
99 @Override
100 public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory,
101 boolean cacheDataInL1) {
102 if (super.getBlock(cacheKey, true, false, true) != null) {
103 throw new RuntimeException("Cached an already cached block");
104 }
105 super.cacheBlock(cacheKey, buf, inMemory, cacheDataInL1);
106 }
107
108 @Override
109 public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
110 if (super.getBlock(cacheKey, true, false, true) != null) {
111 throw new RuntimeException("Cached an already cached block");
112 }
113 super.cacheBlock(cacheKey, buf);
114 }
115 }
116
117 @Before
118 public void setup() throws FileNotFoundException, IOException {
119 cache =
120 new MockedBucketCache(ioEngineName, capacitySize, constructedBlockSize,
121 constructedBlockSizes, writeThreads, writerQLen, persistencePath);
122 }
123
124 @After
125 public void tearDown() {
126 cache.shutdown();
127 }
128
129
130
131
132 private static <T> T randFrom(List<T> a) {
133 return a.get(RAND.nextInt(a.size()));
134 }
135
136 @Test
137 public void testBucketAllocator() throws BucketAllocatorException {
138 BucketAllocator mAllocator = cache.getAllocator();
139
140
141
142 final List<Integer> BLOCKSIZES = Arrays.asList(4 * 1024, 8 * 1024, 64 * 1024, 96 * 1024);
143
144 boolean full = false;
145 ArrayList<Long> allocations = new ArrayList<Long>();
146
147
148 List<Integer> tmp = new ArrayList<Integer>(BLOCKSIZES);
149 while (!full) {
150 Integer blockSize = null;
151 try {
152 blockSize = randFrom(tmp);
153 allocations.add(mAllocator.allocateBlock(blockSize));
154 } catch (CacheFullException cfe) {
155 tmp.remove(blockSize);
156 if (tmp.isEmpty()) full = true;
157 }
158 }
159
160 for (Integer blockSize : BLOCKSIZES) {
161 BucketSizeInfo bucketSizeInfo = mAllocator.roundUpToBucketSizeInfo(blockSize);
162 IndexStatistics indexStatistics = bucketSizeInfo.statistics();
163 assertEquals("unexpected freeCount for " + bucketSizeInfo, 0, indexStatistics.freeCount());
164 }
165
166 for (long offset : allocations) {
167 assertEquals(mAllocator.sizeOfAllocation(offset), mAllocator.freeBlock(offset));
168 }
169 assertEquals(0, mAllocator.getUsedSize());
170 }
171
172 @Test
173 public void testCacheSimple() throws Exception {
174 CacheTestUtils.testCacheSimple(cache, BLOCK_SIZE, NUM_QUERIES);
175 }
176
177 @Test
178 public void testCacheMultiThreadedSingleKey() throws Exception {
179 CacheTestUtils.hammerSingleKey(cache, BLOCK_SIZE, 2 * NUM_THREADS, 2 * NUM_QUERIES);
180 }
181
182 @Test
183 public void testHeapSizeChanges() throws Exception {
184 cache.stopWriterThreads();
185 CacheTestUtils.testHeapSizeChanges(cache, BLOCK_SIZE);
186 }
187
188
189
190 private void cacheAndWaitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey,
191 Cacheable block) throws InterruptedException {
192 cache.cacheBlock(cacheKey, block);
193 while (!cache.backingMap.containsKey(cacheKey)) {
194 Thread.sleep(100);
195 }
196 }
197
198 @Test
199 public void testMemoryLeak() throws Exception {
200 final BlockCacheKey cacheKey = new BlockCacheKey("dummy", 1L);
201 cacheAndWaitUntilFlushedToBucket(cache, cacheKey, new CacheTestUtils.ByteArrayCacheable(
202 new byte[10]));
203 long lockId = cache.backingMap.get(cacheKey).offset();
204 ReentrantReadWriteLock lock = cache.offsetLock.getLock(lockId);
205 lock.writeLock().lock();
206 Thread evictThread = new Thread("evict-block") {
207
208 @Override
209 public void run() {
210 cache.evictBlock(cacheKey);
211 }
212
213 };
214 evictThread.start();
215 cache.offsetLock.waitForWaiters(lockId, 1);
216 cache.blockEvicted(cacheKey, cache.backingMap.remove(cacheKey), true);
217 cacheAndWaitUntilFlushedToBucket(cache, cacheKey, new CacheTestUtils.ByteArrayCacheable(
218 new byte[10]));
219 lock.writeLock().unlock();
220 evictThread.join();
221 assertEquals(1L, cache.getBlockCount());
222 assertTrue(cache.getCurrentSize() > 0L);
223 assertTrue("We should have a block!", cache.iterator().hasNext());
224 }
225
226 @Test
227 public void testRetrieveFromFile() throws Exception {
228 HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
229 Path testDir = TEST_UTIL.getDataTestDir();
230 TEST_UTIL.getTestFileSystem().mkdirs(testDir);
231
232 BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
233 constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen, testDir
234 + "/bucket.persistence");
235 long usedSize = bucketCache.getAllocator().getUsedSize();
236 assertTrue(usedSize == 0);
237
238 HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1);
239
240 for (HFileBlockPair block : blocks) {
241 bucketCache.cacheBlock(block.getBlockName(), block.getBlock());
242 }
243 for (HFileBlockPair block : blocks) {
244 cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock());
245 }
246 usedSize = bucketCache.getAllocator().getUsedSize();
247 assertTrue(usedSize != 0);
248
249 bucketCache.shutdown();
250
251
252 bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
253 constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen, testDir
254 + "/bucket.persistence");
255 assertEquals(usedSize, bucketCache.getAllocator().getUsedSize());
256
257 bucketCache.shutdown();
258
259
260
261 int[] smallBucketSizes = new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024 };
262 bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
263 constructedBlockSize, smallBucketSizes, writeThreads,
264 writerQLen, testDir + "/bucket.persistence");
265 assertEquals(0, bucketCache.getAllocator().getUsedSize());
266 assertEquals(0, bucketCache.backingMap.size());
267
268 TEST_UTIL.cleanupTestDir();
269 }
270 }