Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -1280,20 +1280,13 @@ public interface FSReader {
* or -1 if unknown
* @param pread true to use pread, otherwise use the stream read.
* @param updateMetrics update the metrics or not.
* @param intoHeap allocate the block's ByteBuff by {@link ByteBuffAllocator} or JVM heap.
* For LRUBlockCache, we must ensure that the block to cache is an heap
* one, because the memory occupation is based on heap now, also for
* {@link CombinedBlockCache}, we use the heap LRUBlockCache as L1 cache to
* cache small blocks such as IndexBlock or MetaBlock for faster access. So
* introduce an flag here to decide whether allocate from JVM heap or not
* so that we can avoid an extra off-heap to heap memory copy when using
* LRUBlockCache. For most cases, we known what's the expected block type
* we'll read, while for some special case (Example:
* HFileReaderImpl#readNextDataBlock()), we cannot pre-decide what's the
* expected block type, then we can only allocate block's ByteBuff from
* {@link ByteBuffAllocator} firstly, and then when caching it in
* {@link LruBlockCache} we'll check whether the ByteBuff is from heap or
* not, if not then we'll clone it to an heap one and cache it.
* @param intoHeap whether to allocate the block's ByteBuff on the JVM heap instead of
* using {@link ByteBuffAllocator}. Some cache paths require heap-backed
* blocks. Allocating directly on heap avoids an extra off-heap-to-heap
* copy when the block is cached. When the expected block type is not known
* in advance, the block may first be allocated by
* {@link ByteBuffAllocator} and copied to heap later if required by the
* cache.
* @return the newly read block
*/
HFileBlock readBlockData(long offset, long onDiskSize, boolean pread, boolean updateMetrics,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1301,8 +1301,8 @@ private boolean shouldUseHeap(BlockType expectedBlockType, boolean cacheBlock) {
}

if (!cacheConf.isCombinedBlockCache()) {
// Block to cache in LruBlockCache must be an heap one, if caching enabled. So just allocate
// block memory from heap for saving an extra off-heap to heap copying in that case.
// The non-combined cache path requires cached blocks to be heap-backed. Allocate cacheable
// blocks on heap directly to avoid an extra off-heap-to-heap copy.
return cacheBlock;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
import org.apache.yetus.audience.InterfaceAudience;

/**
* Represents an entry in the {@link LruBlockCache}.
* Represents a cached entry in {@link org.apache.hadoop.hbase.io.hfile.cache.LruCacheEngine}.
* <p>
* Makes the block memory-aware with {@link HeapSize} and Comparable to sort by access time for the
* LRU. It also takes care of priority by either instantiating as in-memory or handling the
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -56,12 +56,13 @@
* </ul>
* <p>
* The important difference from {@link CacheEngine} is the abstraction level. A {@code CacheEngine}
* is a storage back-end implemented by a concrete cache such as LruBlockCache, BucketCache, or a
* future CarrotCache engine. {@code CacheAccessService} is the HBase-facing facade above topology
* and engines. It receives request intent through {@link CacheRequestContext} and insertion intent
* through {@link CacheWriteContext}. A service implementation may initially delegate to an existing
* {@code BlockCache}; a later implementation may delegate to {@link CacheTopology},
* {@link CachePlacementAdmissionPolicy}, and one or more {@link CacheEngine} instances.
* is a storage back-end implemented by a concrete cache such as {@link LruCacheEngine},
* BucketCache, or a future CarrotCache engine. {@code CacheAccessService} is the HBase-facing
* facade above topology and engines. It receives request intent through {@link CacheRequestContext}
* and insertion intent through {@link CacheWriteContext}. A service implementation may initially
* delegate to an existing {@code BlockCache}; a later implementation may delegate to
* {@link CacheTopology}, {@link CachePlacementAdmissionPolicy}, and one or more {@link CacheEngine}
* instances.
* </p>
* <p>
* Normal HBase read and write path code should depend on this interface rather than accessing
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,9 @@
* orchestration, admission control, placement decisions, or promotion/demotion across cache levels.
* </p>
* <p>
* This interface is intentionally aligned with the storage-oriented subset of the current
* {@code BlockCache} contract so that existing implementations such as LruBlockCache and
* BucketCache can be migrated incrementally with minimal behavioral risk.
* </p>
* Storage-engine abstraction used by the pluggable block cache architecture. Native
* implementations, such as {@link LruCacheEngine}, implement this contract directly. Legacy
* {@code BlockCache} implementations can be integrated through compatibility adapters.
* <p>
* Responsibilities of a {@code CacheEngine} include:
* </p>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@ public class LruCacheEngine implements ResizableCacheEngine, HeapSize, Iterable<
* Hard capacity limit. Inserts are rejected once the cache exceeds this factor multiplied by the
* acceptable size.
*/
static final String LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME =
public static final String LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME =
"hbase.lru.blockcache.hard.capacity.limit.factor";

private static final String LRU_SINGLE_PERCENTAGE_CONFIG_NAME =
Expand Down Expand Up @@ -1177,7 +1177,7 @@ public long getBloomBlockCount() {
* Returns the background eviction thread.
* @return eviction thread, or {@code null} if background eviction is disabled
*/
EvictionThread getEvictionThread() {
public EvictionThread getEvictionThread() {
return evictionThread;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -491,9 +491,9 @@ public long size() {
* <p>
* This method aggregates {@link CacheEngine#getCurrentSize()} rather than
* {@link CacheEngine#getCurrentDataSize()} so topology-backed diagnostics preserve legacy
* {@link BlockCache#getCurrentSize()} semantics. For cache implementations such as
* {@code LruBlockCache}, current size may include metadata and cache overhead in addition to
* cached data bytes.
* {@link BlockCache#getCurrentSize()} semantics. For some cache engines, such as
* {@link LruCacheEngine}, current size may include metadata and cache overhead in addition to
* cached block data.
* </p>
* @return aggregate current total cache size in bytes
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -215,8 +215,7 @@ public static BlockCache getBlockCache(CacheAccessService cacheAccessService, Ca
.orElseThrow(() -> new IllegalArgumentException("No cache engine found for tier " + tier));

if (!(engine instanceof BlockCacheBackedCacheEngine)) {
throw new IllegalArgumentException(
"Cache engine for tier " + tier + " must be a BlockCacheBackedCacheEngine");
return null;
}

return ((BlockCacheBackedCacheEngine) engine).getBlockCache();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.hbase.io.hfile;

import static org.apache.hadoop.hbase.io.ByteBuffAllocator.HEAP;
import static org.junit.jupiter.api.Assertions.assertFalse;

import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.io.hfile.cache.CacheEngine;
import org.apache.hadoop.hbase.nio.ByteBuff;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class BlockCacheTestUtils {

private static final Logger LOG = LoggerFactory.getLogger(BlockCacheTestUtils.class);

private BlockCacheTestUtils() {
}

static void testMultiThreadGetAndEvictBlockInternal(BlockCache cache) throws Exception {
int size = 100;
int length = HConstants.HFILEBLOCK_HEADER_SIZE + size;
byte[] byteArr = new byte[length];
HFileContext meta = new HFileContextBuilder().build();
BlockCacheKey key = new BlockCacheKey("key1", 0);
HFileBlock blk = new HFileBlock(BlockType.DATA, size, size, -1,
ByteBuff.wrap(ByteBuffer.wrap(byteArr, 0, size)), HFileBlock.FILL_HEADER, -1, 52, -1, meta,
HEAP);
AtomicBoolean err1 = new AtomicBoolean(false);
Thread t1 = new Thread(() -> {
for (int i = 0; i < 10000 && !err1.get(); i++) {
try {
cache.getBlock(key, false, false, true);
} catch (Exception e) {
err1.set(true);
LOG.info("Cache block or get block failure: ", e);
}
}
});

AtomicBoolean err2 = new AtomicBoolean(false);
Thread t2 = new Thread(() -> {
for (int i = 0; i < 10000 && !err2.get(); i++) {
try {
cache.evictBlock(key);
} catch (Exception e) {
err2.set(true);
LOG.info("Evict block failure: ", e);
}
}
});

AtomicBoolean err3 = new AtomicBoolean(false);
Thread t3 = new Thread(() -> {
for (int i = 0; i < 10000 && !err3.get(); i++) {
try {
cache.cacheBlock(key, blk);
} catch (Exception e) {
err3.set(true);
LOG.info("Cache block failure: ", e);
}
}
});
t1.start();
t2.start();
t3.start();
t1.join();
t2.join();
t3.join();
assertFalse(err1.get());
assertFalse(err2.get());
assertFalse(err3.get());
}

public static void testMultiThreadGetAndEvictBlockInternal(CacheEngine cache) throws Exception {
int size = 100;
int length = HConstants.HFILEBLOCK_HEADER_SIZE + size;
byte[] byteArr = new byte[length];
HFileContext meta = new HFileContextBuilder().build();
BlockCacheKey key = new BlockCacheKey("key1", 0);
HFileBlock blk = new HFileBlock(BlockType.DATA, size, size, -1,
ByteBuff.wrap(ByteBuffer.wrap(byteArr, 0, size)), HFileBlock.FILL_HEADER, -1, 52, -1, meta,
HEAP);
AtomicBoolean err1 = new AtomicBoolean(false);
Thread t1 = new Thread(() -> {
for (int i = 0; i < 10000 && !err1.get(); i++) {
try {
cache.getBlock(key, false, false, true);
} catch (Exception e) {
err1.set(true);
LOG.info("Cache block or get block failure: ", e);
}
}
});

AtomicBoolean err2 = new AtomicBoolean(false);
Thread t2 = new Thread(() -> {
for (int i = 0; i < 10000 && !err2.get(); i++) {
try {
cache.evictBlock(key);
} catch (Exception e) {
err2.set(true);
LOG.info("Evict block failure: ", e);
}
}
});

AtomicBoolean err3 = new AtomicBoolean(false);
Thread t3 = new Thread(() -> {
for (int i = 0; i < 10000 && !err3.get(); i++) {
try {
cache.cacheBlock(key, blk);
} catch (Exception e) {
err3.set(true);
LOG.info("Cache block failure: ", e);
}
}
});
t1.start();
t2.start();
t3.start();
t1.join();
t2.join();
t3.join();
assertFalse(err1.get());
assertFalse(err2.get());
assertFalse(err3.get());
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache;
import org.apache.hadoop.hbase.io.hfile.cache.CacheEngine;
import org.apache.hadoop.hbase.nio.ByteBuff;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ChecksumType;
Expand Down Expand Up @@ -377,6 +378,22 @@ public static void getBlockAndAssertEquals(BlockCache cache, BlockCacheKey key,
}
}

public static void getBlockAndAssertEquals(CacheEngine cache, BlockCacheKey key,
Cacheable blockToCache, ByteBuffer destBuffer, ByteBuffer expectedBuffer) {
destBuffer.clear();
cache.cacheBlock(key, blockToCache);
Cacheable actualBlock = cache.getBlock(key, false, false, false);
try {
actualBlock.serialize(destBuffer, true);
assertEquals(expectedBuffer, destBuffer);
} finally {
// Release the reference count increased by getBlock.
if (actualBlock != null) {
actualBlock.release();
}
}
}

public static void testConvertToJSON(BlockCache toBeTested) {
try {
String json = BlockCacheUtil.toJSON(toBeTested);
Expand All @@ -385,4 +402,13 @@ public static void testConvertToJSON(BlockCache toBeTested) {
fail("conversion to JSON should not fail, exception is:" + e);
}
}

public static void testConvertToJSON(CacheEngine toBeTested) {
try {
String json = BlockCacheUtil.toJSON(toBeTested);
assertNotNull(json);
} catch (Exception e) {
fail("conversion to JSON should not fail, exception is:" + e);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@
import org.apache.hadoop.hbase.io.hfile.cache.CacheAccessService;
import org.apache.hadoop.hbase.io.hfile.cache.CacheAccessServiceTestFactory;
import org.apache.hadoop.hbase.io.hfile.cache.CacheAccessServices;
import org.apache.hadoop.hbase.io.hfile.cache.LruCacheEngine;
import org.apache.hadoop.hbase.io.hfile.cache.TopologyBackedCacheAccessServices;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.regionserver.HRegion;
Expand Down Expand Up @@ -168,8 +169,8 @@ private static List<CacheAccessService> getCacheServices() throws IOException {
// default
BlockCache defaultBlockCache = BlockCacheFactory.createBlockCache(conf);
caches.add(CacheAccessServices.fromBlockCache(defaultBlockCache));
// set LruBlockCache.LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME to 2.0f due to HBASE-16287
TEST_UTIL.getConfiguration().setFloat(LruBlockCache.LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME,
// set LruCacheEngine.LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME to 2.0f due to HBASE-16287
TEST_UTIL.getConfiguration().setFloat(LruCacheEngine.LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME,
2.0f);
// memory
CacheAccessService lru =
Expand Down Expand Up @@ -200,27 +201,30 @@ public static Stream<Arguments> parameters() throws IOException {
return params.stream();
}

/**
* Clears all blocks visible through the cache access service.
* @param cache cache access service to clear
* @throws InterruptedException if interrupted while waiting for asynchronous cache removal
*/
private void clearBlockCache(CacheAccessService cache) throws InterruptedException {
// TODO: HBASE-30018 refactor later
BlockCache blockCache = TopologyBackedCacheAccessServices.getBlockCache(cache);
if (blockCache instanceof LruBlockCache) {
((LruBlockCache) blockCache).clearCache();
} else {
// BucketCache may not return all cached blocks(blocks in write queue), so check it here.
for (int clearCount = 0; cache.getBlockCount() > 0; clearCount++) {
if (clearCount > 0) {
LOG.warn("clear block cache " + cache + " " + clearCount + " times, "
+ cache.getBlockCount() + " blocks remaining");
Thread.sleep(10);
}
for (CachedBlock block : Lists.newArrayList(blockCache)) {
BlockCacheKey key = new BlockCacheKey(block.getFilename(), block.getOffset());
// CombinedBucketCache may need evict two times.
for (int evictCount = 0; cache.evictBlock(key); evictCount++) {
if (evictCount > 1) {
LOG.warn("evict block " + block + " in " + cache + " " + evictCount
+ " times, maybe a bug here");
}
for (int clearCount = 0; cache.getBlockCount() > 0; clearCount++) {
if (clearCount > 0) {
LOG.warn("clear block cache " + cache + " " + clearCount + " times, "
+ cache.getBlockCount() + " blocks remaining");
Thread.sleep(10);
}

Iterable<CachedBlock> cachedBlocks = CacheAccessServices.asCachedBlockIterable(cache)
.orElseThrow(() -> new IllegalStateException("Cache does not support block iteration"));

for (CachedBlock block : Lists.newArrayList(cachedBlocks)) {
BlockCacheKey key = new BlockCacheKey(block.getFilename(), block.getOffset());

// A block may exist in more than one tier, so keep evicting until it is gone.
for (int evictCount = 0; cache.evictBlock(key); evictCount++) {
if (evictCount > 1) {
LOG.warn("evict block " + block + " in " + cache + " " + evictCount
+ " times, maybe a bug here");
}
}
}
Expand All @@ -240,7 +244,7 @@ public void setUp() throws IOException {
cowType.shouldBeCached(BlockType.LEAF_INDEX));
conf.setBoolean(CacheConfig.CACHE_BLOOM_BLOCKS_ON_WRITE_KEY,
cowType.shouldBeCached(BlockType.BLOOM_CHUNK));
cacheConf = new CacheConfig(conf, TopologyBackedCacheAccessServices.getBlockCache(cache));
cacheConf = new CacheConfig(conf, cache);
fs = HFileSystem.get(conf);
}

Expand Down
Loading
Loading