View Javadoc

1   /**
2    * Copyright The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one or more
5    * contributor license agreements. See the NOTICE file distributed with this
6    * work for additional information regarding copyright ownership. The ASF
7    * licenses this file to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance with the License.
9    * You may obtain a copy of the License at
10   *
11   * http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
15   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16   * License for the specific language governing permissions and limitations
17   * under the License.
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   * Basic test of BucketCache.Puts and gets.
50   * <p>
51   * Tests will ensure that blocks' data correctness under several threads concurrency
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 }, // TODO: why is 8k the default blocksize for these tests?
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    * Return a random element from {@code a}.
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      * Test the allocator first
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     // Fill the allocated extents by choosing a random blocksize. Continues selecting blocks until
147     // the cache is completely filled.
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   // BucketCache.cacheBlock is async, it first adds block to ramCache and writeQueue, then writer
189   // threads will flush it to the bucket and put reference entry in backingMap.
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     // Add blocks
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     // persist cache to file
249     bucketCache.shutdown();
250 
251     // restore cache from file
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     // persist cache to file
257     bucketCache.shutdown();
258 
259     // reconfig buckets sizes, the biggest bucket is small than constructedBlockSize (8k or 16k)
260     // so it can't restore cache from file
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 
271   @Test
272   public void testOffsetProducesPositiveOutput() {
273     //This number is picked because it produces negative output if the values isn't ensured to be positive.
274     //See HBASE-18757 for more information.
275     long testValue = 549888460800L;
276     BucketCache.BucketEntry bucketEntry = new BucketCache.BucketEntry(testValue, 10, 10L, true);
277     assertEquals(testValue, bucketEntry.offset());
278   }
279 }