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.jupiter.api.Assertions.assertFalse;
021import static org.junit.jupiter.api.Assertions.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.CountDownLatch;
029import java.util.concurrent.ThreadLocalRandom;
030import java.util.concurrent.TimeUnit;
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.fs.FileSystem;
033import org.apache.hadoop.fs.Path;
034import org.apache.hadoop.hbase.HBaseTestingUtil;
035import org.apache.hadoop.hbase.KeyValue;
036import org.apache.hadoop.hbase.fs.HFileSystem;
037import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
038import org.apache.hadoop.hbase.io.hfile.BlockType;
039import org.apache.hadoop.hbase.io.hfile.CacheConfig;
040import org.apache.hadoop.hbase.io.hfile.HFile;
041import org.apache.hadoop.hbase.io.hfile.HFileBlock;
042import org.apache.hadoop.hbase.io.hfile.HFileContext;
043import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
044import org.apache.hadoop.hbase.io.hfile.PrefetchExecutor;
045import org.apache.hadoop.hbase.io.hfile.RandomKeyValueUtil;
046import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
047import org.apache.hadoop.hbase.testclassification.IOTests;
048import org.apache.hadoop.hbase.testclassification.MediumTests;
049import org.junit.jupiter.api.Tag;
050import org.junit.jupiter.api.Test;
051import org.slf4j.Logger;
052import org.slf4j.LoggerFactory;
053
054@Tag(IOTests.TAG)
055@Tag(MediumTests.TAG)
056public class TestBucketCachePersister {
057
058  public int constructedBlockSize = 16 * 1024;
059
060  private static final Logger LOG = LoggerFactory.getLogger(TestBucketCachePersister.class);
061
062  public int[] constructedBlockSizes =
063    new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024, 8 * 1024 + 1024, 16 * 1024 + 1024,
064      28 * 1024 + 1024, 32 * 1024 + 1024, 64 * 1024 + 1024, 96 * 1024 + 1024, 128 * 1024 + 1024 };
065
066  private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
067
068  private static final int NUM_VALID_KEY_TYPES = KeyValue.Type.values().length - 2;
069  private static final int DATA_BLOCK_SIZE = 2048;
070  private static final int NUM_KV = 1000;
071
072  final long capacitySize = 32 * 1024 * 1024;
073  final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS;
074  final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS;
075  Path testDir;
076
077  public Configuration setupBucketCacheConfig(long bucketCachePersistInterval) throws IOException {
078    Configuration conf;
079    conf = TEST_UTIL.getConfiguration();
080    conf.setBoolean(CacheConfig.PREFETCH_BLOCKS_ON_OPEN_KEY, true);
081    testDir = TEST_UTIL.getDataTestDir();
082    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
083    conf.setLong(CacheConfig.BUCKETCACHE_PERSIST_INTERVAL_KEY, bucketCachePersistInterval);
084    return conf;
085  }
086
087  public BucketCache setupBucketCache(Configuration conf, String persistentCacheFile)
088    throws IOException {
089    BucketCache bucketCache = new BucketCache("file:" + testDir + "/" + persistentCacheFile,
090      capacitySize, constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen,
091      testDir + "/bucket.persistence", 60 * 1000, conf);
092    return bucketCache;
093  }
094
095  public void cleanupBucketCache(BucketCache bucketCache) throws IOException {
096    bucketCache.shutdown();
097    TEST_UTIL.cleanupDataTestDirOnTestFS(String.valueOf(testDir));
098    assertFalse(TEST_UTIL.getTestFileSystem().exists(testDir));
099  }
100
101  @Test
102  public void testPersistenceIntervalChange() throws Exception {
103    // Keep the initial interval above the assertion timeout so a stale interval fails the test.
104    Configuration conf = setupBucketCacheConfig(TimeUnit.SECONDS.toMillis(10));
105    CountDownLatch firstPersistence = new CountDownLatch(1);
106    CountDownLatch resumePersister = new CountDownLatch(1);
107    CountDownLatch secondPersistence = new CountDownLatch(1);
108    BucketCache bucketCache = new BucketCache("file:" + testDir + "/interval.cache", capacitySize,
109      constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen,
110      testDir + "/bucket.persistence", 60 * 1000, conf) {
111      @Override
112      public void setCacheInconsistent(boolean inconsistent) {
113        super.setCacheInconsistent(inconsistent);
114        if (!inconsistent) {
115          if (firstPersistence.getCount() > 0) {
116            firstPersistence.countDown();
117            // Pause after the first persistence, before the persister starts its next sleep.
118            try {
119              resumePersister.await();
120            } catch (InterruptedException e) {
121              Thread.currentThread().interrupt();
122            }
123          } else {
124            secondPersistence.countDown();
125          }
126        }
127      }
128    };
129    try {
130      assertTrue(bucketCache.waitForCacheInitialization(10000));
131      bucketCache.setCacheInconsistent(true);
132      assertTrue(firstPersistence.await(30, TimeUnit.SECONDS),
133        "The persister should persist using the initial interval");
134
135      conf.setLong(CacheConfig.BUCKETCACHE_PERSIST_INTERVAL_KEY, TimeUnit.SECONDS.toMillis(1));
136      bucketCache.onConfigurationChange(conf);
137      bucketCache.setCacheInconsistent(true);
138      resumePersister.countDown();
139
140      assertTrue(secondPersistence.await(5, TimeUnit.SECONDS),
141        "The next persistence should use the updated interval");
142    } finally {
143      resumePersister.countDown();
144      cleanupBucketCache(bucketCache);
145    }
146  }
147
148  @Test
149  public void testPrefetchPersistenceCrash() throws Exception {
150    long bucketCachePersistInterval = 3000;
151    Configuration conf = setupBucketCacheConfig(bucketCachePersistInterval);
152    BucketCache bucketCache = setupBucketCache(conf, "testPrefetchPersistenceCrash");
153    CacheConfig cacheConf = new CacheConfig(conf, bucketCache);
154    FileSystem fs = HFileSystem.get(conf);
155    // Load Cache
156    Path storeFile = writeStoreFile("TestPrefetch0", conf, cacheConf, fs);
157    Path storeFile2 = writeStoreFile("TestPrefetch1", conf, cacheConf, fs);
158    readStoreFile(storeFile, 0, fs, cacheConf, conf, bucketCache);
159    readStoreFile(storeFile2, 0, fs, cacheConf, conf, bucketCache);
160    Thread.sleep(bucketCachePersistInterval);
161    assertTrue(new File(testDir + "/bucket.persistence").exists());
162    assertTrue(new File(testDir + "/bucket.persistence").delete());
163    cleanupBucketCache(bucketCache);
164  }
165
166  @Test
167  public void testPrefetchPersistenceCrashNegative() throws Exception {
168    long bucketCachePersistInterval = Long.MAX_VALUE;
169    Configuration conf = setupBucketCacheConfig(bucketCachePersistInterval);
170    BucketCache bucketCache = setupBucketCache(conf, "testPrefetchPersistenceCrashNegative");
171    CacheConfig cacheConf = new CacheConfig(conf, bucketCache);
172    FileSystem fs = HFileSystem.get(conf);
173    // Load Cache
174    Path storeFile = writeStoreFile("TestPrefetch2", conf, cacheConf, fs);
175    readStoreFile(storeFile, 0, fs, cacheConf, conf, bucketCache);
176    assertFalse(new File(testDir + "/bucket.persistence").exists());
177    cleanupBucketCache(bucketCache);
178  }
179
180  @Test
181  public void testPrefetchListUponBlockEviction() throws Exception {
182    Configuration conf = setupBucketCacheConfig(200);
183    BucketCache bucketCache = setupBucketCache(conf, "testPrefetchListUponBlockEviction");
184    CacheConfig cacheConf = new CacheConfig(conf, bucketCache);
185    FileSystem fs = HFileSystem.get(conf);
186    // Load Blocks in cache
187    Path storeFile = writeStoreFile("TestPrefetch3", conf, cacheConf, fs);
188    readStoreFile(storeFile, 0, fs, cacheConf, conf, bucketCache);
189    int retries = 0;
190    while (!bucketCache.fullyCachedFiles.containsKey(storeFile.getName()) && retries < 5) {
191      Thread.sleep(500);
192      retries++;
193    }
194    assertTrue(retries < 5);
195    BlockCacheKey bucketCacheKey = bucketCache.backingMap.entrySet().iterator().next().getKey();
196    // Evict Blocks from cache
197    bucketCache.evictBlock(bucketCacheKey);
198    assertFalse(bucketCache.fullyCachedFiles.containsKey(storeFile.getName()));
199    cleanupBucketCache(bucketCache);
200  }
201
202  @Test
203  public void testPrefetchBlockEvictionWhilePrefetchRunning() throws Exception {
204    Configuration conf = setupBucketCacheConfig(200);
205    BucketCache bucketCache =
206      setupBucketCache(conf, "testPrefetchBlockEvictionWhilePrefetchRunning");
207    CacheConfig cacheConf = new CacheConfig(conf, bucketCache);
208    FileSystem fs = HFileSystem.get(conf);
209    // Load Blocks in cache
210    Path storeFile = writeStoreFile("TestPrefetch3", conf, cacheConf, fs);
211    HFile.createReader(fs, storeFile, cacheConf, true, conf);
212    boolean evicted = false;
213    while (!PrefetchExecutor.isCompleted(storeFile)) {
214      LOG.debug("Entered loop as prefetch for {} is still running.", storeFile);
215      if (bucketCache.backingMap.size() > 0 && !evicted) {
216        Iterator<Map.Entry<BlockCacheKey, BucketEntry>> it =
217          bucketCache.backingMap.entrySet().iterator();
218        // Evict a data block from cache
219        Map.Entry<BlockCacheKey, BucketEntry> entry = it.next();
220        while (it.hasNext() && !evicted) {
221          if (entry.getKey().getBlockType().equals(BlockType.DATA)) {
222            evicted = bucketCache.evictBlock(it.next().getKey());
223            LOG.debug("Attempted eviction for {}. Succeeded? {}", storeFile, evicted);
224          }
225        }
226      }
227      Thread.sleep(10);
228    }
229    assertFalse(bucketCache.fullyCachedFiles.containsKey(storeFile.getName()));
230    cleanupBucketCache(bucketCache);
231  }
232
233  public void readStoreFile(Path storeFilePath, long offset, FileSystem fs, CacheConfig cacheConf,
234    Configuration conf, BucketCache bucketCache) throws Exception {
235    // Open the file
236    HFile.Reader reader = HFile.createReader(fs, storeFilePath, cacheConf, true, conf);
237
238    while (!reader.prefetchComplete()) {
239      // Sleep for a bit
240      Thread.sleep(1000);
241    }
242    HFileBlock block = reader.readBlock(offset, -1, false, true, false, true, null, null);
243    BlockCacheKey blockCacheKey = new BlockCacheKey(reader.getName(), offset);
244    BucketEntry be = bucketCache.backingMap.get(blockCacheKey);
245    boolean isCached = bucketCache.getBlock(blockCacheKey, true, false, true) != null;
246
247    if (
248      block.getBlockType() == BlockType.DATA || block.getBlockType() == BlockType.ROOT_INDEX
249        || block.getBlockType() == BlockType.INTERMEDIATE_INDEX
250    ) {
251      assertTrue(isCached);
252    }
253  }
254
255  public Path writeStoreFile(String fname, Configuration conf, CacheConfig cacheConf, FileSystem fs)
256    throws IOException {
257    Path storeFileParentDir = new Path(TEST_UTIL.getDataTestDir(), fname);
258    HFileContext meta = new HFileContextBuilder().withBlockSize(DATA_BLOCK_SIZE).build();
259    StoreFileWriter sfw = new StoreFileWriter.Builder(conf, cacheConf, fs)
260      .withOutputDir(storeFileParentDir).withFileContext(meta).build();
261    Random rand = ThreadLocalRandom.current();
262    final int rowLen = 32;
263    for (int i = 0; i < NUM_KV; ++i) {
264      byte[] k = RandomKeyValueUtil.randomOrderedKey(rand, i);
265      byte[] v = RandomKeyValueUtil.randomValue(rand);
266      int cfLen = rand.nextInt(k.length - rowLen + 1);
267      KeyValue kv = new KeyValue(k, 0, rowLen, k, rowLen, cfLen, k, rowLen + cfLen,
268        k.length - rowLen - cfLen, rand.nextLong(), generateKeyType(rand), v, 0, v.length);
269      sfw.append(kv);
270    }
271
272    sfw.close();
273    return sfw.getPath();
274  }
275
276  public static KeyValue.Type generateKeyType(Random rand) {
277    if (rand.nextBoolean()) {
278      // Let's make half of KVs puts.
279      return KeyValue.Type.Put;
280    } else {
281      KeyValue.Type keyType = KeyValue.Type.values()[1 + rand.nextInt(NUM_VALID_KEY_TYPES)];
282      if (keyType == KeyValue.Type.Minimum || keyType == KeyValue.Type.Maximum) {
283        throw new RuntimeException("Generated an invalid key type: " + keyType + ". "
284          + "Probably the layout of KeyValue.Type has changed.");
285      }
286      return keyType;
287    }
288  }
289
290}