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.junit.Assert.assertFalse; 021import static org.junit.Assert.assertTrue; 022 023import java.io.File; 024import java.io.IOException; 025import java.util.Iterator; 026import java.util.Map; 027import java.util.Random; 028import java.util.concurrent.ThreadLocalRandom; 029import org.apache.hadoop.conf.Configuration; 030import org.apache.hadoop.fs.FileSystem; 031import org.apache.hadoop.fs.Path; 032import org.apache.hadoop.hbase.HBaseClassTestRule; 033import org.apache.hadoop.hbase.HBaseTestingUtil; 034import org.apache.hadoop.hbase.KeyValue; 035import org.apache.hadoop.hbase.fs.HFileSystem; 036import org.apache.hadoop.hbase.io.hfile.BlockCacheKey; 037import org.apache.hadoop.hbase.io.hfile.BlockType; 038import org.apache.hadoop.hbase.io.hfile.CacheConfig; 039import org.apache.hadoop.hbase.io.hfile.HFile; 040import org.apache.hadoop.hbase.io.hfile.HFileBlock; 041import org.apache.hadoop.hbase.io.hfile.HFileContext; 042import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder; 043import org.apache.hadoop.hbase.io.hfile.PrefetchExecutor; 044import org.apache.hadoop.hbase.io.hfile.RandomKeyValueUtil; 045import org.apache.hadoop.hbase.regionserver.StoreFileWriter; 046import org.apache.hadoop.hbase.testclassification.IOTests; 047import org.apache.hadoop.hbase.testclassification.MediumTests; 048import org.junit.ClassRule; 049import org.junit.Test; 050import org.junit.experimental.categories.Category; 051import org.junit.rules.TestName; 052 053@Category({ IOTests.class, MediumTests.class }) 054public class TestBucketCachePersister { 055 056 @ClassRule 057 public static final HBaseClassTestRule CLASS_RULE = 058 HBaseClassTestRule.forClass(TestBucketCachePersister.class); 059 060 public TestName name = new TestName(); 061 062 public int constructedBlockSize = 16 * 1024; 063 064 public int[] constructedBlockSizes = 065 new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024, 8 * 1024 + 1024, 16 * 1024 + 1024, 066 28 * 1024 + 1024, 32 * 1024 + 1024, 64 * 1024 + 1024, 96 * 1024 + 1024, 128 * 1024 + 1024 }; 067 068 private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 069 070 private static final int NUM_VALID_KEY_TYPES = KeyValue.Type.values().length - 2; 071 private static final int DATA_BLOCK_SIZE = 2048; 072 private static final int NUM_KV = 1000; 073 074 final long capacitySize = 32 * 1024 * 1024; 075 final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS; 076 final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS; 077 Path testDir; 078 079 public Configuration setupBucketCacheConfig(long bucketCachePersistInterval) throws IOException { 080 Configuration conf; 081 conf = TEST_UTIL.getConfiguration(); 082 conf.setBoolean(CacheConfig.PREFETCH_BLOCKS_ON_OPEN_KEY, true); 083 testDir = TEST_UTIL.getDataTestDir(); 084 TEST_UTIL.getTestFileSystem().mkdirs(testDir); 085 conf.setLong(CacheConfig.BUCKETCACHE_PERSIST_INTERVAL_KEY, bucketCachePersistInterval); 086 return conf; 087 } 088 089 public BucketCache setupBucketCache(Configuration conf, String persistentCacheFile) 090 throws IOException { 091 BucketCache bucketCache = new BucketCache("file:" + testDir + "/" + persistentCacheFile, 092 capacitySize, constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen, 093 testDir + "/bucket.persistence", 60 * 1000, conf); 094 return bucketCache; 095 } 096 097 public void cleanupBucketCache(BucketCache bucketCache) throws IOException { 098 bucketCache.shutdown(); 099 TEST_UTIL.cleanupDataTestDirOnTestFS(String.valueOf(testDir)); 100 assertFalse(TEST_UTIL.getTestFileSystem().exists(testDir)); 101 } 102 103 @Test 104 public void testPrefetchPersistenceCrash() throws Exception { 105 long bucketCachePersistInterval = 3000; 106 Configuration conf = setupBucketCacheConfig(bucketCachePersistInterval); 107 BucketCache bucketCache = setupBucketCache(conf, "testPrefetchPersistenceCrash"); 108 CacheConfig cacheConf = new CacheConfig(conf, bucketCache); 109 FileSystem fs = HFileSystem.get(conf); 110 // Load Cache 111 Path storeFile = writeStoreFile("TestPrefetch0", conf, cacheConf, fs); 112 Path storeFile2 = writeStoreFile("TestPrefetch1", conf, cacheConf, fs); 113 readStoreFile(storeFile, 0, fs, cacheConf, conf, bucketCache); 114 readStoreFile(storeFile2, 0, fs, cacheConf, conf, bucketCache); 115 Thread.sleep(bucketCachePersistInterval); 116 assertTrue(new File(testDir + "/bucket.persistence").exists()); 117 assertTrue(new File(testDir + "/bucket.persistence").delete()); 118 cleanupBucketCache(bucketCache); 119 } 120 121 @Test 122 public void testPrefetchPersistenceCrashNegative() throws Exception { 123 long bucketCachePersistInterval = Long.MAX_VALUE; 124 Configuration conf = setupBucketCacheConfig(bucketCachePersistInterval); 125 BucketCache bucketCache = setupBucketCache(conf, "testPrefetchPersistenceCrashNegative"); 126 CacheConfig cacheConf = new CacheConfig(conf, bucketCache); 127 FileSystem fs = HFileSystem.get(conf); 128 // Load Cache 129 Path storeFile = writeStoreFile("TestPrefetch2", conf, cacheConf, fs); 130 readStoreFile(storeFile, 0, fs, cacheConf, conf, bucketCache); 131 assertFalse(new File(testDir + "/bucket.persistence").exists()); 132 cleanupBucketCache(bucketCache); 133 } 134 135 @Test 136 public void testPrefetchListUponBlockEviction() throws Exception { 137 Configuration conf = setupBucketCacheConfig(200); 138 BucketCache bucketCache = setupBucketCache(conf, "testPrefetchListUponBlockEviction"); 139 CacheConfig cacheConf = new CacheConfig(conf, bucketCache); 140 FileSystem fs = HFileSystem.get(conf); 141 // Load Blocks in cache 142 Path storeFile = writeStoreFile("TestPrefetch3", conf, cacheConf, fs); 143 readStoreFile(storeFile, 0, fs, cacheConf, conf, bucketCache); 144 int retries = 0; 145 while (!bucketCache.fullyCachedFiles.containsKey(storeFile.getName()) && retries < 5) { 146 Thread.sleep(500); 147 retries++; 148 } 149 assertTrue(retries < 5); 150 BlockCacheKey bucketCacheKey = bucketCache.backingMap.entrySet().iterator().next().getKey(); 151 // Evict Blocks from cache 152 bucketCache.evictBlock(bucketCacheKey); 153 assertFalse(bucketCache.fullyCachedFiles.containsKey(storeFile.getName())); 154 cleanupBucketCache(bucketCache); 155 } 156 157 @Test 158 public void testPrefetchBlockEvictionWhilePrefetchRunning() throws Exception { 159 Configuration conf = setupBucketCacheConfig(200); 160 BucketCache bucketCache = 161 setupBucketCache(conf, "testPrefetchBlockEvictionWhilePrefetchRunning"); 162 CacheConfig cacheConf = new CacheConfig(conf, bucketCache); 163 FileSystem fs = HFileSystem.get(conf); 164 // Load Blocks in cache 165 Path storeFile = writeStoreFile("TestPrefetch3", conf, cacheConf, fs); 166 HFile.createReader(fs, storeFile, cacheConf, true, conf); 167 boolean evicted = false; 168 while (!PrefetchExecutor.isCompleted(storeFile)) { 169 if (bucketCache.backingMap.size() > 0 && !evicted) { 170 Iterator<Map.Entry<BlockCacheKey, BucketEntry>> it = 171 bucketCache.backingMap.entrySet().iterator(); 172 // Evict a data block from cache 173 Map.Entry<BlockCacheKey, BucketEntry> entry = it.next(); 174 while (it.hasNext() && !evicted) { 175 if (entry.getKey().getBlockType().equals(BlockType.DATA)) { 176 evicted = bucketCache.evictBlock(it.next().getKey()); 177 } 178 } 179 } 180 Thread.sleep(10); 181 } 182 assertFalse(bucketCache.fullyCachedFiles.containsKey(storeFile.getName())); 183 cleanupBucketCache(bucketCache); 184 } 185 186 public void readStoreFile(Path storeFilePath, long offset, FileSystem fs, CacheConfig cacheConf, 187 Configuration conf, BucketCache bucketCache) throws Exception { 188 // Open the file 189 HFile.Reader reader = HFile.createReader(fs, storeFilePath, cacheConf, true, conf); 190 191 while (!reader.prefetchComplete()) { 192 // Sleep for a bit 193 Thread.sleep(1000); 194 } 195 HFileBlock block = reader.readBlock(offset, -1, false, true, false, true, null, null); 196 BlockCacheKey blockCacheKey = new BlockCacheKey(reader.getName(), offset); 197 BucketEntry be = bucketCache.backingMap.get(blockCacheKey); 198 boolean isCached = bucketCache.getBlock(blockCacheKey, true, false, true) != null; 199 200 if ( 201 block.getBlockType() == BlockType.DATA || block.getBlockType() == BlockType.ROOT_INDEX 202 || block.getBlockType() == BlockType.INTERMEDIATE_INDEX 203 ) { 204 assertTrue(isCached); 205 } 206 } 207 208 public Path writeStoreFile(String fname, Configuration conf, CacheConfig cacheConf, FileSystem fs) 209 throws IOException { 210 Path storeFileParentDir = new Path(TEST_UTIL.getDataTestDir(), fname); 211 HFileContext meta = new HFileContextBuilder().withBlockSize(DATA_BLOCK_SIZE).build(); 212 StoreFileWriter sfw = new StoreFileWriter.Builder(conf, cacheConf, fs) 213 .withOutputDir(storeFileParentDir).withFileContext(meta).build(); 214 Random rand = ThreadLocalRandom.current(); 215 final int rowLen = 32; 216 for (int i = 0; i < NUM_KV; ++i) { 217 byte[] k = RandomKeyValueUtil.randomOrderedKey(rand, i); 218 byte[] v = RandomKeyValueUtil.randomValue(rand); 219 int cfLen = rand.nextInt(k.length - rowLen + 1); 220 KeyValue kv = new KeyValue(k, 0, rowLen, k, rowLen, cfLen, k, rowLen + cfLen, 221 k.length - rowLen - cfLen, rand.nextLong(), generateKeyType(rand), v, 0, v.length); 222 sfw.append(kv); 223 } 224 225 sfw.close(); 226 return sfw.getPath(); 227 } 228 229 public static KeyValue.Type generateKeyType(Random rand) { 230 if (rand.nextBoolean()) { 231 // Let's make half of KVs puts. 232 return KeyValue.Type.Put; 233 } else { 234 KeyValue.Type keyType = KeyValue.Type.values()[1 + rand.nextInt(NUM_VALID_KEY_TYPES)]; 235 if (keyType == KeyValue.Type.Minimum || keyType == KeyValue.Type.Maximum) { 236 throw new RuntimeException("Generated an invalid key type: " + keyType + ". " 237 + "Probably the layout of KeyValue.Type has changed."); 238 } 239 return keyType; 240 } 241 } 242 243}