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.hamcrest.MatcherAssert.assertThat;
021import static org.hamcrest.Matchers.both;
022import static org.hamcrest.Matchers.hasKey;
023import static org.hamcrest.io.FileMatchers.anExistingFile;
024import static org.junit.jupiter.api.Assertions.assertEquals;
025import static org.junit.jupiter.api.Assertions.assertNotEquals;
026import static org.junit.jupiter.api.Assertions.assertTrue;
027
028import java.io.File;
029import java.io.IOException;
030import java.util.Random;
031import java.util.concurrent.ThreadLocalRandom;
032import java.util.stream.Stream;
033import org.apache.hadoop.conf.Configuration;
034import org.apache.hadoop.fs.FileSystem;
035import org.apache.hadoop.fs.Path;
036import org.apache.hadoop.hbase.HBaseParameterizedTestTemplate;
037import org.apache.hadoop.hbase.HBaseTestingUtil;
038import org.apache.hadoop.hbase.KeyValue;
039import org.apache.hadoop.hbase.Waiter;
040import org.apache.hadoop.hbase.fs.HFileSystem;
041import org.apache.hadoop.hbase.io.hfile.CacheConfig;
042import org.apache.hadoop.hbase.io.hfile.HFile;
043import org.apache.hadoop.hbase.io.hfile.HFileContext;
044import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
045import org.apache.hadoop.hbase.io.hfile.PrefetchExecutor;
046import org.apache.hadoop.hbase.io.hfile.RandomKeyValueUtil;
047import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
048import org.apache.hadoop.hbase.testclassification.IOTests;
049import org.apache.hadoop.hbase.testclassification.SmallTests;
050import org.junit.jupiter.api.AfterEach;
051import org.junit.jupiter.api.BeforeEach;
052import org.junit.jupiter.api.Tag;
053import org.junit.jupiter.api.TestTemplate;
054import org.junit.jupiter.params.provider.Arguments;
055
056@Tag(IOTests.TAG)
057@Tag(SmallTests.TAG)
058@HBaseParameterizedTestTemplate(name = "{index}: blockSize={0}, bucketSizes={1}")
059public class TestPrefetchPersistence {
060
061  private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
062
063  private static final int NUM_VALID_KEY_TYPES = KeyValue.Type.values().length - 2;
064  private static final int DATA_BLOCK_SIZE = 2048;
065  private static final int NUM_KV = 1000;
066
067  public static Stream<Arguments> parameters() {
068    return Stream.of(Arguments.of(16 * 1024,
069      new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024, 8 * 1024 + 1024, 16 * 1024 + 1024,
070        28 * 1024 + 1024, 32 * 1024 + 1024, 64 * 1024 + 1024, 96 * 1024 + 1024,
071        128 * 1024 + 1024 }));
072  }
073
074  private final int constructedBlockSize;
075  private final int[] constructedBlockSizes;
076
077  public TestPrefetchPersistence(int constructedBlockSize, int[] constructedBlockSizes) {
078    this.constructedBlockSize = constructedBlockSize;
079    this.constructedBlockSizes = constructedBlockSizes;
080  }
081
082  private Configuration conf;
083  private CacheConfig cacheConf;
084  private FileSystem fs;
085  String prefetchPersistencePath;
086  Path testDir;
087
088  BucketCache bucketCache;
089
090  final long capacitySize = 32 * 1024 * 1024;
091  final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS;
092  final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS;
093
094  @BeforeEach
095  public void setup() throws IOException {
096    conf = TEST_UTIL.getConfiguration();
097    conf.setBoolean(CacheConfig.PREFETCH_BLOCKS_ON_OPEN_KEY, true);
098    conf.setInt(PrefetchExecutor.PREFETCH_DELAY, 0);
099    PrefetchExecutor.loadConfiguration(conf);
100    testDir = TEST_UTIL.getDataTestDir();
101    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
102    fs = HFileSystem.get(conf);
103  }
104
105  @TestTemplate
106  public void testPrefetchPersistence() throws Exception {
107    bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
108      constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen,
109      testDir + "/bucket.persistence", 60 * 1000, conf);
110    bucketCache.waitForCacheInitialization(10000);
111    cacheConf = new CacheConfig(conf, bucketCache);
112
113    long usedSize = bucketCache.getAllocator().getUsedSize();
114    assertEquals(0, usedSize);
115    assertTrue(new File(testDir + "/bucket.cache").exists());
116    // Load Cache
117    Path storeFile = writeStoreFile("TestPrefetch0");
118    Path storeFile2 = writeStoreFile("TestPrefetch1");
119    readStoreFile(storeFile);
120    readStoreFile(storeFile2);
121    usedSize = bucketCache.getAllocator().getUsedSize();
122    assertNotEquals(0, usedSize);
123
124    bucketCache.shutdown();
125    assertThat(new File(testDir + "/bucket.persistence"), anExistingFile());
126    bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
127      constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen,
128      testDir + "/bucket.persistence", 60 * 1000, conf);
129    bucketCache.waitForCacheInitialization(10000);
130    cacheConf = new CacheConfig(conf, bucketCache);
131    assertNotEquals(usedSize, 0);
132    assertThat(bucketCache.fullyCachedFiles,
133      both(hasKey(storeFile.getName())).and(hasKey(storeFile2.getName())));
134  }
135
136  @AfterEach
137  public void cleanup() {
138    TEST_UTIL.cleanupTestDir();
139  }
140
141  public void readStoreFile(Path storeFilePath) throws Exception {
142    // Open the file
143    HFile.Reader reader = HFile.createReader(fs, storeFilePath, cacheConf, true, conf);
144    Waiter.waitFor(conf, 30000, () -> reader.prefetchComplete()
145      || bucketCache.fullyCachedFiles.containsKey(storeFilePath.getName()));
146  }
147
148  public Path writeStoreFile(String fname) throws IOException {
149    Path storeFileParentDir = new Path(TEST_UTIL.getDataTestDir(), fname);
150    HFileContext meta = new HFileContextBuilder().withBlockSize(DATA_BLOCK_SIZE).build();
151    StoreFileWriter sfw = new StoreFileWriter.Builder(conf, cacheConf, fs)
152      .withOutputDir(storeFileParentDir).withFileContext(meta).build();
153    Random rand = ThreadLocalRandom.current();
154    final int rowLen = 32;
155    for (int i = 0; i < NUM_KV; ++i) {
156      byte[] k = RandomKeyValueUtil.randomOrderedKey(rand, i);
157      byte[] v = RandomKeyValueUtil.randomValue(rand);
158      int cfLen = rand.nextInt(k.length - rowLen + 1);
159      KeyValue kv = new KeyValue(k, 0, rowLen, k, rowLen, cfLen, k, rowLen + cfLen,
160        k.length - rowLen - cfLen, rand.nextLong(), generateKeyType(rand), v, 0, v.length);
161      sfw.append(kv);
162    }
163
164    sfw.close();
165    return sfw.getPath();
166  }
167
168  public static KeyValue.Type generateKeyType(Random rand) {
169    if (rand.nextBoolean()) {
170      // Let's make half of KVs puts.
171      return KeyValue.Type.Put;
172    } else {
173      KeyValue.Type keyType = KeyValue.Type.values()[1 + rand.nextInt(NUM_VALID_KEY_TYPES)];
174      if (keyType == KeyValue.Type.Minimum || keyType == KeyValue.Type.Maximum) {
175        throw new RuntimeException("Generated an invalid key type: " + keyType + ". "
176          + "Probably the layout of KeyValue.Type has changed.");
177      }
178      return keyType;
179    }
180  }
181
182}