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    while (bucketCache.backingMap.size() == 0) {
168      Thread.sleep(10);
169    }
170    Iterator<Map.Entry<BlockCacheKey, BucketEntry>> it =
171      bucketCache.backingMap.entrySet().iterator();
172    // Evict Blocks from cache
173    bucketCache.evictBlock(it.next().getKey());
174    bucketCache.evictBlock(it.next().getKey());
175    int retries = 0;
176    while (!PrefetchExecutor.isCompleted(storeFile) && retries < 5) {
177      Thread.sleep(500);
178      retries++;
179    }
180    assertTrue(retries < 5);
181    assertFalse(bucketCache.fullyCachedFiles.containsKey(storeFile.getName()));
182    cleanupBucketCache(bucketCache);
183  }
184
185  public void readStoreFile(Path storeFilePath, long offset, FileSystem fs, CacheConfig cacheConf,
186    Configuration conf, BucketCache bucketCache) throws Exception {
187    // Open the file
188    HFile.Reader reader = HFile.createReader(fs, storeFilePath, cacheConf, true, conf);
189
190    while (!reader.prefetchComplete()) {
191      // Sleep for a bit
192      Thread.sleep(1000);
193    }
194    HFileBlock block = reader.readBlock(offset, -1, false, true, false, true, null, null);
195    BlockCacheKey blockCacheKey = new BlockCacheKey(reader.getName(), offset);
196    BucketEntry be = bucketCache.backingMap.get(blockCacheKey);
197    boolean isCached = bucketCache.getBlock(blockCacheKey, true, false, true) != null;
198
199    if (
200      block.getBlockType() == BlockType.DATA || block.getBlockType() == BlockType.ROOT_INDEX
201        || block.getBlockType() == BlockType.INTERMEDIATE_INDEX
202    ) {
203      assertTrue(isCached);
204    }
205  }
206
207  public Path writeStoreFile(String fname, Configuration conf, CacheConfig cacheConf, FileSystem fs)
208    throws IOException {
209    Path storeFileParentDir = new Path(TEST_UTIL.getDataTestDir(), fname);
210    HFileContext meta = new HFileContextBuilder().withBlockSize(DATA_BLOCK_SIZE).build();
211    StoreFileWriter sfw = new StoreFileWriter.Builder(conf, cacheConf, fs)
212      .withOutputDir(storeFileParentDir).withFileContext(meta).build();
213    Random rand = ThreadLocalRandom.current();
214    final int rowLen = 32;
215    for (int i = 0; i < NUM_KV; ++i) {
216      byte[] k = RandomKeyValueUtil.randomOrderedKey(rand, i);
217      byte[] v = RandomKeyValueUtil.randomValue(rand);
218      int cfLen = rand.nextInt(k.length - rowLen + 1);
219      KeyValue kv = new KeyValue(k, 0, rowLen, k, rowLen, cfLen, k, rowLen + cfLen,
220        k.length - rowLen - cfLen, rand.nextLong(), generateKeyType(rand), v, 0, v.length);
221      sfw.append(kv);
222    }
223
224    sfw.close();
225    return sfw.getPath();
226  }
227
228  public static KeyValue.Type generateKeyType(Random rand) {
229    if (rand.nextBoolean()) {
230      // Let's make half of KVs puts.
231      return KeyValue.Type.Put;
232    } else {
233      KeyValue.Type keyType = KeyValue.Type.values()[1 + rand.nextInt(NUM_VALID_KEY_TYPES)];
234      if (keyType == KeyValue.Type.Minimum || keyType == KeyValue.Type.Maximum) {
235        throw new RuntimeException("Generated an invalid key type: " + keyType + ". "
236          + "Probably the layout of KeyValue.Type has changed.");
237      }
238      return keyType;
239    }
240  }
241
242}