Copilot commented on code in PR #8653:
URL: https://github.com/apache/hbase/pull/8653#discussion_r4065482965
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/cache/TieredInclusiveTopology.java:
##########
@@ -110,4 +110,20 @@ public void shutdown() {
l1.shutdown();
l2.shutdown();
}
+
+ /**
+ * Handles a capacity-driven eviction from this inclusive topology.
+ * <p>
+ * L1 eviction does not require demotion because inclusive placement
maintains the corresponding
+ * block in L2. L2 pressure eviction likewise does not cause movement to
another tier.
+ * </p>
+ * @param cacheKey key identifying the evicted block
+ * @param block evicted block
+ * @param sourceEngine engine that evicted the block
+ * @return {@code false}, because no additional placement is required
+ */
+ @Override
+ public boolean handleEviction(BlockCacheKey cacheKey, Cacheable block,
CacheEngine sourceEngine) {
+ return false;
Review Comment:
Inclusive placement writes L1 and L2 in separate calls, and an engine
insertion has no success result. L1 can therefore evict before the L2 write
completes, or while L2 has no copy, and this unconditional return drops the
only remaining block. For an L1 eviction, first check L2 and insert the evicted
block only when L2 does not already contain it; L2 evictions should still leave
the topology.
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/cache/CacheAccessServices.java:
##########
@@ -81,33 +77,59 @@ public static CacheAccessService fromBlockCache(BlockCache
blockCache) {
DefaultHBaseCachePlacementAdmissionPolicy.INSTANCE);
}
+ /**
+ * Creates a {@link CacheAccessService} from the block cache configuration.
+ * @param conf cache configuration
+ * @return configured cache access service, or a disabled service when block
caching is disabled
+ * @throws NullPointerException if {@code conf} is {@code null}
+ */
+ public static CacheAccessService fromConfiguration(Configuration conf) {
+ return fromConfiguration(conf, null);
+ }
+
/**
* Creates a {@link CacheAccessService} from the block cache configuration.
* <p>
- * This method is a compatibility factory for tests and transitional code
paths that want to
- * obtain a {@link CacheAccessService} directly from {@link Configuration},
while still using the
- * existing {@link BlockCacheFactory} and legacy {@link BlockCache}
implementations underneath.
- * </p>
- * <p>
- * The method delegates block-cache construction to
- * {@link BlockCacheFactory#createBlockCache(Configuration)}. If the legacy
factory creates a
- * {@link BlockCache}, the returned service is backed by that cache through
- * {@link TopologyBackedCacheAccessService}. If the legacy factory does not
create a cache, this
- * method returns the disabled/no-op cache access service.
- * </p>
- * <p>
- * This method does not introduce new cache-engine or topology-based runtime
wiring. It is
- * intended only as a bridge while existing HBase tests and integration
paths migrate from direct
- * {@link BlockCache} usage to {@link CacheAccessService}.
+ * Cache implementations that implement {@link CacheEngine} natively are
used directly. Legacy
+ * {@link BlockCache} implementations are adapted to {@link CacheEngine}
until their migration is
+ * complete.
* </p>
- * @param conf configuration used by {@link BlockCacheFactory}
- * @return cache access service created from the configured legacy block
cache, or disabled when
- * no block cache is configured
+ * @param conf cache configuration
+ * @param onlineRegions currently online regions, or {@code null} when
unavailable
+ * @return configured cache access service, or a disabled service when block
caching is disabled
* @throws NullPointerException if {@code conf} is {@code null}
*/
- public static CacheAccessService fromConfiguration(Configuration conf) {
+ public static CacheAccessService fromConfiguration(Configuration conf,
+ Map<String, HRegion> onlineRegions) {
Objects.requireNonNull(conf, "conf must not be null");
- return fromBlockCache(BlockCacheFactory.createBlockCache(conf));
+
+ CacheEngine l1 = BlockCacheFactory.createFirstLevelCacheEngine(conf);
+ if (l1 == null) {
+ return disabled();
+ }
+
+ CachePlacementAdmissionPolicy policy =
DefaultHBaseCachePlacementAdmissionPolicy.INSTANCE;
+
+ boolean useExternal =
conf.getBoolean(BlockCacheFactory.EXTERNAL_BLOCKCACHE_KEY,
+ BlockCacheFactory.EXTERNAL_BLOCKCACHE_DEFAULT);
+
+ if (useExternal) {
+ CacheEngine l2 = BlockCacheFactory.createExternalCacheEngine(conf);
+ if (l2 == null) {
+ return
TopologyBackedCacheAccessServices.fromSingleCacheEngine("single", l1, policy);
+ }
+
+ return
TopologyBackedCacheAccessServices.fromTieredInclusiveCacheEngines("inclusive",
l1, l2,
+ policy);
Review Comment:
This new external-cache branch creates a `TIERED_INCLUSIVE` service, but
`CacheConfig.isCombinedBlockCache()` recognizes only `TIERED_EXCLUSIVE`.
Consequently `HFileReaderImpl.shouldUseHeap()` treats the inclusive L1/L2 cache
as a single on-heap cache and allocates DATA blocks on heap, unlike the legacy
`InclusiveCombinedBlockCache` path. Extend the compatibility check to treat
both tiered topology types as combined caches.
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/cache/LruCacheEngine.java:
##########
@@ -0,0 +1,1549 @@
+/*
+ * 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.cache;
+
+import java.lang.ref.WeakReference;
+import java.util.EnumMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.concurrent.locks.ReentrantLock;
+import org.apache.commons.lang3.mutable.MutableBoolean;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseInterfaceAudience;
+import org.apache.hadoop.hbase.io.HeapSize;
+import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
+import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
+import org.apache.hadoop.hbase.io.hfile.BlockCacheUtil;
+import org.apache.hadoop.hbase.io.hfile.BlockPriority;
+import org.apache.hadoop.hbase.io.hfile.BlockType;
+import org.apache.hadoop.hbase.io.hfile.CacheStats;
+import org.apache.hadoop.hbase.io.hfile.Cacheable;
+import org.apache.hadoop.hbase.io.hfile.CachedBlock;
+import org.apache.hadoop.hbase.io.hfile.HFileBlock;
+import org.apache.hadoop.hbase.io.hfile.LruCachedBlock;
+import org.apache.hadoop.hbase.io.hfile.LruCachedBlockQueue;
+import org.apache.hadoop.hbase.util.ClassSize;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.base.MoreObjects;
+import org.apache.hbase.thirdparty.com.google.common.base.Objects;
+import
org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+/**
+ * Native LRU {@link CacheEngine} implementation.
+ * <p>
+ * This cache is memory-aware using {@link HeapSize}, memory-bound using an
LRU eviction algorithm,
+ * and concurrent. It is backed by a {@link ConcurrentHashMap} and can use a
non-blocking eviction
+ * thread, providing constant-time {@link #cacheBlock(BlockCacheKey,
Cacheable, boolean)} and
+ * {@link #getBlock(BlockCacheKey, boolean, boolean, boolean)} operations.
+ * </p>
+ * <p>
+ * The cache maintains three block-priority levels to provide scan resistance
and support in-memory
+ * column families:
+ * </p>
+ * <ul>
+ * <li>single-access blocks</li>
+ * <li>multiple-access blocks</li>
+ * <li>in-memory blocks</li>
+ * </ul>
+ * <p>
+ * Each priority is assigned a portion of the total cache capacity. During
eviction the cache tries
+ * to preserve the configured relative sizes while allowing unused capacity in
one priority to be
+ * consumed by another.
+ * </p>
+ * <p>
+ * This class is a storage engine only. It does not perform L1/L2
orchestration, victim-cache
+ * delegation, tier placement, admission control, promotion, or demotion.
Those responsibilities
+ * belong to the cache topology and policy layers.
+ * </p>
+ */
[email protected]
+public class LruCacheEngine implements CacheEngine, HeapSize,
Iterable<CachedBlock> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(LruCacheEngine.class);
+
+ /**
+ * Percentage of total size that eviction will evict until.
+ */
+ private static final String LRU_MIN_FACTOR_CONFIG_NAME =
"hbase.lru.blockcache.min.factor";
+
+ /**
+ * Acceptable cache size above which eviction is triggered.
+ */
+ private static final String LRU_ACCEPTABLE_FACTOR_CONFIG_NAME =
+ "hbase.lru.blockcache.acceptable.factor";
+
+ /**
+ * 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 =
+ "hbase.lru.blockcache.hard.capacity.limit.factor";
+
+ private static final String LRU_SINGLE_PERCENTAGE_CONFIG_NAME =
+ "hbase.lru.blockcache.single.percentage";
+
+ private static final String LRU_MULTI_PERCENTAGE_CONFIG_NAME =
+ "hbase.lru.blockcache.multi.percentage";
+
+ private static final String LRU_MEMORY_PERCENTAGE_CONFIG_NAME =
+ "hbase.lru.blockcache.memory.percentage";
+
+ /**
+ * Configuration key that gives data blocks from in-memory HFiles higher
eviction priority.
+ */
+ private static final String LRU_IN_MEMORY_FORCE_MODE_CONFIG_NAME =
+ "hbase.lru.rs.inmemoryforcemode";
+
+ static final float DEFAULT_LOAD_FACTOR = 0.75f;
+ static final int DEFAULT_CONCURRENCY_LEVEL = 16;
+
+ private static final float DEFAULT_MIN_FACTOR = 0.95f;
+ static final float DEFAULT_ACCEPTABLE_FACTOR = 0.99f;
+
+ private static final float DEFAULT_SINGLE_FACTOR = 0.25f;
+ private static final float DEFAULT_MULTI_FACTOR = 0.50f;
+ private static final float DEFAULT_MEMORY_FACTOR = 0.25f;
+
+ private static final float DEFAULT_HARD_CAPACITY_LIMIT_FACTOR = 1.2f;
+
+ private static final boolean DEFAULT_IN_MEMORY_FORCE_MODE = false;
+
+ private static final int STAT_THREAD_PERIOD = 60 * 5;
+
+ private static final String LRU_MAX_BLOCK_SIZE = "hbase.lru.max.block.size";
+
+ private static final long DEFAULT_MAX_BLOCK_SIZE = 16L * 1024L * 1024L;
+
+ /**
+ * Fixed heap overhead of an LRU cache engine instance.
+ */
+ public static final long CACHE_FIXED_OVERHEAD =
+ ClassSize.estimateBase(LruCacheEngine.class, false);
+
+ /**
+ * Cached blocks keyed by their HFile block cache key.
+ * <p>
+ * A {@link ConcurrentHashMap} is required because {@link #getBlock} and
eviction depend on the
+ * atomicity guarantees of {@code computeIfPresent}.
+ * </p>
+ */
+ private transient final ConcurrentHashMap<BlockCacheKey, LruCachedBlock> map;
+
+ /** Lock protecting the eviction process. */
+ private transient final ReentrantLock evictionLock = new ReentrantLock(true);
+
+ /** Maximum size of an individual block accepted by this cache. */
+ private final long maxBlockSize;
+
+ /** Whether an eviction pass is currently running. */
+ private volatile boolean evictionInProgress;
+
+ /** Optional background eviction thread. */
+ private transient final EvictionThread evictionThread;
+
+ /**
+ * Listener notified about capacity-driven block evictions.
+ */
+ private volatile CacheEvictionListener evictionListener;
+
+ /** Executor used to periodically report cache statistics. */
+ private transient final ScheduledExecutorService scheduleThreadPool =
+ Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
+ .setNameFormat("LruCacheEngineStatsExecutor").setDaemon(true).build());
+
+ /** Current total heap size used by the cache. */
+ private final AtomicLong size;
+
+ /** Current heap size of data blocks. */
+ private final LongAdder dataBlockSize = new LongAdder();
+
+ /** Current heap size of index blocks. */
+ private final LongAdder indexBlockSize = new LongAdder();
+
+ /** Current heap size of bloom blocks. */
+ private final LongAdder bloomBlockSize = new LongAdder();
+
+ /** Current number of cached blocks. */
+ private final AtomicLong elements;
+
+ /** Current number of cached data blocks. */
+ private final LongAdder dataBlockElements = new LongAdder();
+
+ /** Current number of cached index blocks. */
+ private final LongAdder indexBlockElements = new LongAdder();
+
+ /** Current number of cached bloom blocks. */
+ private final LongAdder bloomBlockElements = new LongAdder();
+
+ /** Sequential cache access identifier. */
+ private final AtomicLong count;
+
+ /** Hard cache capacity limit factor. */
+ private float hardCapacityLimitFactor;
+
+ /** Cache statistics. */
+ private final CacheStats stats;
+
+ /** Maximum cache size in bytes. */
+ private long maxSize;
+
+ /** Expected average block size. */
+ private long blockSize;
+
+ /** Cache size factor at which eviction is triggered. */
+ private float acceptableFactor;
+
+ /** Cache size factor to which an eviction pass should reduce the cache. */
+ private float minFactor;
+
+ /** Fraction of capacity assigned to single-access blocks. */
+ private float singleFactor;
+
+ /** Fraction of capacity assigned to multiple-access blocks. */
+ private float multiFactor;
+
+ /** Fraction of capacity assigned to in-memory blocks. */
+ private float memoryFactor;
+
+ /** Heap overhead of the cache structure itself. */
+ private long overhead;
+
+ /** Whether data blocks from in-memory HFiles receive stronger retention
priority. */
+ private boolean forceInMemory;
+
+ /**
+ * Creates an LRU cache engine using default configuration values.
+ * @param maxSize maximum size of the cache, in bytes
+ * @param blockSize expected average block size, in bytes
+ */
+ public LruCacheEngine(long maxSize, long blockSize) {
+ this(maxSize, blockSize, true);
+ }
+
+ /**
+ * Creates an LRU cache engine and optionally enables the background
eviction thread.
+ * @param maxSize maximum size of the cache, in bytes
+ * @param blockSize expected average block size, in bytes
+ * @param evictionThread whether background eviction should be enabled
+ */
+ public LruCacheEngine(long maxSize, long blockSize, boolean evictionThread) {
+ this(maxSize, blockSize, evictionThread, (int) Math.ceil(1.2 * maxSize /
blockSize),
+ DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL, DEFAULT_MIN_FACTOR,
DEFAULT_ACCEPTABLE_FACTOR,
+ DEFAULT_SINGLE_FACTOR, DEFAULT_MULTI_FACTOR, DEFAULT_MEMORY_FACTOR,
+ DEFAULT_HARD_CAPACITY_LIMIT_FACTOR, false, DEFAULT_MAX_BLOCK_SIZE);
+ }
+
+ /**
+ * Creates an LRU cache engine using values from the supplied configuration.
+ * @param maxSize maximum size of the cache, in bytes
+ * @param blockSize expected average block size, in bytes
+ * @param evictionThread whether background eviction should be enabled
+ * @param conf cache configuration
+ */
+ public LruCacheEngine(long maxSize, long blockSize, boolean evictionThread,
Configuration conf) {
+ this(maxSize, blockSize, evictionThread, (int) Math.ceil(1.2 * maxSize /
blockSize),
+ DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL,
+ conf.getFloat(LRU_MIN_FACTOR_CONFIG_NAME, DEFAULT_MIN_FACTOR),
+ conf.getFloat(LRU_ACCEPTABLE_FACTOR_CONFIG_NAME,
DEFAULT_ACCEPTABLE_FACTOR),
+ conf.getFloat(LRU_SINGLE_PERCENTAGE_CONFIG_NAME, DEFAULT_SINGLE_FACTOR),
+ conf.getFloat(LRU_MULTI_PERCENTAGE_CONFIG_NAME, DEFAULT_MULTI_FACTOR),
+ conf.getFloat(LRU_MEMORY_PERCENTAGE_CONFIG_NAME, DEFAULT_MEMORY_FACTOR),
+ conf.getFloat(LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME,
DEFAULT_HARD_CAPACITY_LIMIT_FACTOR),
+ conf.getBoolean(LRU_IN_MEMORY_FORCE_MODE_CONFIG_NAME,
DEFAULT_IN_MEMORY_FORCE_MODE),
+ conf.getLong(LRU_MAX_BLOCK_SIZE, DEFAULT_MAX_BLOCK_SIZE));
+ }
+
+ /**
+ * Creates an LRU cache engine using the supplied configuration and
background eviction.
+ * @param maxSize maximum size of the cache, in bytes
+ * @param blockSize expected average block size, in bytes
+ * @param conf cache configuration
+ */
+ public LruCacheEngine(long maxSize, long blockSize, Configuration conf) {
+ this(maxSize, blockSize, true, conf);
+ }
+
+ /**
+ * Creates a fully configured LRU cache engine.
+ * @param maxSize maximum size of this cache, in bytes
+ * @param blockSize expected average size of blocks, in bytes
+ * @param evictionThread whether to run eviction in a background thread
+ * @param mapInitialSize initial size of the backing map
+ * @param mapLoadFactor load factor of the backing map
+ * @param mapConcurrencyLevel concurrency level of the backing map
+ * @param minFactor fraction of maximum size retained after
eviction
+ * @param acceptableFactor fraction of maximum size that triggers eviction
+ * @param singleFactor fraction assigned to single-access blocks
+ * @param multiFactor fraction assigned to multiple-access blocks
+ * @param memoryFactor fraction assigned to in-memory blocks
+ * @param hardLimitFactor hard capacity limit factor
+ * @param forceInMemory whether in-memory HFile blocks receive
stronger retention priority
+ * @param maxBlockSize largest individual block accepted by this cache
+ */
+ public LruCacheEngine(long maxSize, long blockSize, boolean evictionThread,
int mapInitialSize,
+ float mapLoadFactor, int mapConcurrencyLevel, float minFactor, float
acceptableFactor,
+ float singleFactor, float multiFactor, float memoryFactor, float
hardLimitFactor,
+ boolean forceInMemory, long maxBlockSize) {
+ this.maxBlockSize = maxBlockSize;
+
+ if (
+ singleFactor + multiFactor + memoryFactor != 1 || singleFactor < 0 ||
multiFactor < 0
+ || memoryFactor < 0
+ ) {
+ throw new IllegalArgumentException(
+ "Single, multi, and memory factors should be non-negative and total
1.0");
+ }
+
+ if (minFactor >= acceptableFactor) {
+ throw new IllegalArgumentException("minFactor must be smaller than
acceptableFactor");
+ }
+
+ if (minFactor >= 1.0f || acceptableFactor >= 1.0f) {
+ throw new IllegalArgumentException("all factors must be < 1");
+ }
+
+ this.maxSize = maxSize;
+ this.blockSize = blockSize;
+ this.forceInMemory = forceInMemory;
+ this.map = new ConcurrentHashMap<>(mapInitialSize, mapLoadFactor,
mapConcurrencyLevel);
+ this.minFactor = minFactor;
+ this.acceptableFactor = acceptableFactor;
+ this.singleFactor = singleFactor;
+ this.multiFactor = multiFactor;
+ this.memoryFactor = memoryFactor;
+ this.stats = new CacheStats(getClass().getSimpleName());
+ this.count = new AtomicLong(0);
+ this.elements = new AtomicLong(0);
+ this.overhead = calculateOverhead(maxSize, blockSize, mapConcurrencyLevel);
+ this.size = new AtomicLong(this.overhead);
+ this.hardCapacityLimitFactor = hardLimitFactor;
+
+ if (evictionThread) {
+ this.evictionThread = new EvictionThread(this);
+ this.evictionThread.start();
+ } else {
+ this.evictionThread = null;
+ }
+
+ this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
STAT_THREAD_PERIOD,
+ STAT_THREAD_PERIOD, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Returns the human-readable name of this cache engine.
+ * @return cache engine name
+ */
+ @Override
+ public String getName() {
+ return getClass().getSimpleName();
+ }
+
+ /**
+ * Updates the maximum size of this cache.
+ * <p>
+ * If the cache is already larger than the new acceptable size, an eviction
pass is started.
+ * </p>
+ * @param maxSize new maximum size, in bytes
+ */
+ public void setMaxSize(long maxSize) {
+ this.maxSize = maxSize;
+ if (size.get() > acceptableSize() && !evictionInProgress) {
+ runEviction();
+ }
+ }
+
+ /**
+ * Returns a heap-backed reference suitable for storage in this cache.
+ * <p>
+ * Shared-memory {@link HFileBlock}s are cloned onto the heap. Other blocks
are retained before
+ * being referenced by the cache.
+ * </p>
+ * @param buf block to convert
+ * @return heap-backed retained block
+ */
+ private Cacheable asReferencedHeapBlock(Cacheable buf) {
+ if (buf instanceof HFileBlock) {
+ HFileBlock block = (HFileBlock) buf;
+ if (block.isSharedMem()) {
+ return HFileBlock.deepCloneOnHeap(block);
+ }
+ }
+
+ return buf.retain();
+ }
+
+ /**
+ * Caches the specified block.
+ * @param cacheKey block cache key
+ * @param buf block contents
+ * @param inMemory whether the block should receive in-memory priority
+ */
+ @Override
+ public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean
inMemory) {
+ if (buf.heapSize() > maxBlockSize) {
+ if (stats.failInsert() % 50 == 0) {
+ LOG.warn("Trying to cache too large a block " +
cacheKey.getHfileName() + " @ "
+ + cacheKey.getOffset() + " is " + buf.heapSize() + " which is larger
than "
+ + maxBlockSize);
+ }
+ return;
+ }
+
+ LruCachedBlock cachedBlock = map.get(cacheKey);
+ if (
+ cachedBlock != null &&
!BlockCacheUtil.shouldReplaceExistingCacheBlock(this, cacheKey, buf)
+ ) {
+ return;
+ }
+
+ long currentSize = size.get();
+ long currentAcceptableSize = acceptableSize();
+ long hardLimitSize = (long) (hardCapacityLimitFactor *
currentAcceptableSize);
+
+ if (currentSize >= hardLimitSize) {
+ stats.failInsert();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("LruCacheEngine current size " +
StringUtils.byteDesc(currentSize)
+ + " has exceeded acceptable size " +
StringUtils.byteDesc(currentAcceptableSize) + "."
+ + " The hard limit size is " + StringUtils.byteDesc(hardLimitSize)
+ + ", failed to put cacheKey:" + cacheKey + " into LruCacheEngine.");
+ }
+ if (!evictionInProgress) {
+ runEviction();
+ }
+ return;
+ }
+
+ Cacheable referencedBlock = asReferencedHeapBlock(buf);
+ cachedBlock = new LruCachedBlock(cacheKey, referencedBlock,
count.incrementAndGet(), inMemory);
+
+ long newSize = updateSizeMetrics(cachedBlock, false);
+ map.put(cacheKey, cachedBlock);
+
+ long elementCount = elements.incrementAndGet();
+ if (referencedBlock.getBlockType().isBloom()) {
+ bloomBlockElements.increment();
+ } else if (referencedBlock.getBlockType().isIndex()) {
+ indexBlockElements.increment();
+ } else if (referencedBlock.getBlockType().isData()) {
+ dataBlockElements.increment();
+ }
+
+ if (LOG.isTraceEnabled()) {
+ assertCounterSanity(map.size(), elementCount);
+ }
+
+ if (newSize > currentAcceptableSize && !evictionInProgress) {
+ runEviction();
+ }
+ }
+
+ /**
+ * Caches the specified block with normal cache priority.
+ * @param cacheKey block cache key
+ * @param buf block contents
+ */
+ @Override
+ public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
+ cacheBlock(cacheKey, buf, false);
+ }
+
+ /**
+ * Caches the specified block.
+ * <p>
+ * LRU insertion is synchronous, so {@code waitWhenCache} has no effect.
+ * </p>
+ * @param cacheKey block cache key
+ * @param buf block contents
+ * @param inMemory whether the block should receive in-memory priority
+ * @param waitWhenCache whether the caller requests synchronous completion
+ */
+ @Override
+ public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean
inMemory,
+ boolean waitWhenCache) {
+ cacheBlock(cacheKey, buf, inMemory);
+ }
+
+ /**
+ * Checks consistency between the backing-map size and the element counter.
+ * <p>
+ * This method is intended for TRACE-level diagnostics and assertion-enabled
JVMs.
+ * </p>
+ * @param mapSize current backing-map size
+ * @param counterVal current element-counter value
+ */
+ private static void assertCounterSanity(long mapSize, long counterVal) {
+ if (counterVal < 0) {
+ LOG.trace("counterVal overflow. Assertions unreliable. counterVal=" +
counterVal
+ + ", mapSize=" + mapSize);
+ return;
+ }
+
+ if (mapSize < Integer.MAX_VALUE) {
+ double percentageDifference = Math.abs((((double) counterVal) /
((double) mapSize)) - 1.0);
+ if (percentageDifference > 0.05) {
+ LOG.trace("delta between reported and actual size > 5%. counterVal=" +
counterVal
+ + ", mapSize=" + mapSize);
+ }
+ }
+ }
+
+ /**
+ * Updates total and block-type-specific size metrics.
+ * @param cachedBlock cached block whose size should be applied
+ * @param evict whether this operation represents removal
+ * @return new total cache size
+ */
+ private long updateSizeMetrics(LruCachedBlock cachedBlock, boolean evict) {
+ long heapSize = cachedBlock.heapSize();
+ BlockType blockType = cachedBlock.getBuffer().getBlockType();
+
+ if (evict) {
+ heapSize *= -1;
+ }
+
+ if (blockType != null) {
+ if (blockType.isBloom()) {
+ bloomBlockSize.add(heapSize);
+ } else if (blockType.isIndex()) {
+ indexBlockSize.add(heapSize);
+ } else if (blockType.isData()) {
+ dataBlockSize.add(heapSize);
+ }
+ }
+
+ return size.addAndGet(heapSize);
+ }
+
+ /**
+ * Returns the cached block associated with the specified key.
+ * <p>
+ * Lookup is strictly local to this cache engine. A miss is returned to the
topology layer rather
+ * than being delegated to another cache tier.
+ * </p>
+ * @param cacheKey block cache key
+ * @param caching whether the caller caches blocks on misses
+ * @param repeat whether this is a repeated lookup for the same
block
+ * @param updateCacheMetrics whether cache statistics should be updated
+ * @return cached block, or {@code null} if not present
+ */
+ @Override
+ public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching, boolean
repeat,
+ boolean updateCacheMetrics) {
+ LruCachedBlock cachedBlock = map.computeIfPresent(cacheKey, (key, value)
-> {
+ value.getBuffer().retain();
+ return value;
+ });
+
+ if (cachedBlock == null) {
+ if (!repeat && updateCacheMetrics) {
+ stats.miss(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
+ }
+ return null;
+ }
+
+ if (updateCacheMetrics) {
+ stats.hit(caching, cacheKey.isPrimary(), cacheKey.getBlockType());
+ }
+
+ cachedBlock.access(count.incrementAndGet());
+ return cachedBlock.getBuffer();
+ }
+
+ /**
+ * Returns whether the specified block is currently cached.
+ * @param cacheKey block cache key
+ * @return {@code true} if the block is present
+ */
+ public boolean containsBlock(BlockCacheKey cacheKey) {
+ return map.containsKey(cacheKey);
+ }
+
+ /**
+ * Returns whether the specified block is currently cached.
+ * @param cacheKey block cache key
+ * @return optional containing the local cache-membership result
+ */
+ @Override
+ public Optional<Boolean> isAlreadyCached(BlockCacheKey cacheKey) {
+ return Optional.of(containsBlock(cacheKey));
+ }
+
+ /**
+ * Evicts the specified block.
+ * @param cacheKey block cache key
+ * @return {@code true} if a block was found and evicted
+ */
+ @Override
+ public boolean evictBlock(BlockCacheKey cacheKey) {
+ LruCachedBlock cachedBlock = map.get(cacheKey);
+ return cachedBlock != null && evictBlock(cachedBlock, false) > 0;
+ }
+
+ /**
+ * Evicts all cached blocks belonging to the specified HFile.
+ * <p>
+ * This is a linear scan over the cache contents.
+ * </p>
+ * @param hfileName HFile name
+ * @return number of blocks evicted
+ */
+ @Override
+ public int evictBlocksByHfileName(String hfileName) {
+ int numEvicted = 0;
+
+ for (BlockCacheKey key : map.keySet()) {
+ if (key.getHfileName().equals(hfileName) && evictBlock(key)) {
+ numEvicted++;
+ }
+ }
+
+ return numEvicted;
+ }
+
+ /**
+ * Evicts the specified block from this cache.
+ * <p>
+ * For capacity-driven evictions, the configured eviction listener is
notified while a temporary
+ * reference to the evicted block is retained. Explicit invalidations do not
generate eviction
+ * notifications.
+ * </p>
+ * @param block block to evict
+ * @param evictedByEvictionProcess whether the eviction was caused by cache
pressure
+ * @return heap size of the evicted block, or {@code 0} if the block was not
present
+ */
+ protected long evictBlock(LruCachedBlock block, boolean
evictedByEvictionProcess) {
+ final MutableBoolean evicted = new MutableBoolean(false);
+ final CacheEvictionListener listener = evictedByEvictionProcess ?
evictionListener : null;
+ final Cacheable buffer = block.getBuffer();
+
+ if (listener != null) {
+ buffer.retain();
+ }
+
+ try {
+ map.computeIfPresent(block.getCacheKey(), (key, value) -> {
+ value.getBuffer().release();
+ evicted.setTrue();
+ return null;
+ });
Review Comment:
The eviction candidate may be stale by the time this method runs. Retaining
`block.getBuffer()` before the map operation can retain an already-released
buffer after a concurrent explicit eviction. If the key was replaced instead,
`computeIfPresent` removes the newer mapping while the size accounting and
listener still use the stale candidate. Atomically verify and retain the mapped
value inside `computeIfPresent`, then perform accounting and notification using
that actual removed value; `LruBlockCache.java:523-529` documents why the
retain must occur inside the map operation.
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/cache/CacheAccessServices.java:
##########
@@ -81,33 +77,59 @@ public static CacheAccessService fromBlockCache(BlockCache
blockCache) {
DefaultHBaseCachePlacementAdmissionPolicy.INSTANCE);
}
+ /**
+ * Creates a {@link CacheAccessService} from the block cache configuration.
+ * @param conf cache configuration
+ * @return configured cache access service, or a disabled service when block
caching is disabled
+ * @throws NullPointerException if {@code conf} is {@code null}
+ */
+ public static CacheAccessService fromConfiguration(Configuration conf) {
+ return fromConfiguration(conf, null);
+ }
+
/**
* Creates a {@link CacheAccessService} from the block cache configuration.
* <p>
- * This method is a compatibility factory for tests and transitional code
paths that want to
- * obtain a {@link CacheAccessService} directly from {@link Configuration},
while still using the
- * existing {@link BlockCacheFactory} and legacy {@link BlockCache}
implementations underneath.
- * </p>
- * <p>
- * The method delegates block-cache construction to
- * {@link BlockCacheFactory#createBlockCache(Configuration)}. If the legacy
factory creates a
- * {@link BlockCache}, the returned service is backed by that cache through
- * {@link TopologyBackedCacheAccessService}. If the legacy factory does not
create a cache, this
- * method returns the disabled/no-op cache access service.
- * </p>
- * <p>
- * This method does not introduce new cache-engine or topology-based runtime
wiring. It is
- * intended only as a bridge while existing HBase tests and integration
paths migrate from direct
- * {@link BlockCache} usage to {@link CacheAccessService}.
+ * Cache implementations that implement {@link CacheEngine} natively are
used directly. Legacy
+ * {@link BlockCache} implementations are adapted to {@link CacheEngine}
until their migration is
+ * complete.
* </p>
- * @param conf configuration used by {@link BlockCacheFactory}
- * @return cache access service created from the configured legacy block
cache, or disabled when
- * no block cache is configured
+ * @param conf cache configuration
+ * @param onlineRegions currently online regions, or {@code null} when
unavailable
+ * @return configured cache access service, or a disabled service when block
caching is disabled
* @throws NullPointerException if {@code conf} is {@code null}
*/
- public static CacheAccessService fromConfiguration(Configuration conf) {
+ public static CacheAccessService fromConfiguration(Configuration conf,
+ Map<String, HRegion> onlineRegions) {
Objects.requireNonNull(conf, "conf must not be null");
- return fromBlockCache(BlockCacheFactory.createBlockCache(conf));
+
+ CacheEngine l1 = BlockCacheFactory.createFirstLevelCacheEngine(conf);
+ if (l1 == null) {
+ return disabled();
+ }
+
+ CachePlacementAdmissionPolicy policy =
DefaultHBaseCachePlacementAdmissionPolicy.INSTANCE;
+
+ boolean useExternal =
conf.getBoolean(BlockCacheFactory.EXTERNAL_BLOCKCACHE_KEY,
+ BlockCacheFactory.EXTERNAL_BLOCKCACHE_DEFAULT);
+
+ if (useExternal) {
+ CacheEngine l2 = BlockCacheFactory.createExternalCacheEngine(conf);
+ if (l2 == null) {
+ return
TopologyBackedCacheAccessServices.fromSingleCacheEngine("single", l1, policy);
+ }
+
+ return
TopologyBackedCacheAccessServices.fromTieredInclusiveCacheEngines("inclusive",
l1, l2,
+ policy);
+ }
+
+ CacheEngine l2 = BlockCacheFactory.createBucketCacheEngine(conf,
onlineRegions);
+ if (l2 == null) {
+ return TopologyBackedCacheAccessServices.fromSingleCacheEngine("single",
l1, policy);
+ }
+
+ return
TopologyBackedCacheAccessServices.fromTieredExclusiveCacheEngines("combined",
l1, l2,
+ policy);
Review Comment:
Both newly constructed tiered services expose only L1 statistics because
their topologies return `l1.getStats()`. Data blocks and their accesses are
normally handled by L2, so service-level metrics omit L2 hits, misses, and
evictions instead of preserving the aggregate view provided by
`CombinedBlockCache`. Provide an aggregate `CacheStats` view for both tiered
topology implementations before routing configuration-based construction
through them.
##########
hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestCacheConfig.java:
##########
@@ -510,4 +431,40 @@ void testCacheAccessServiceIsNoOpWhenBlockCacheIsNull() {
assertInstanceOf(NoOpCacheAccessService.class, service);
assertFalse(service.isCacheEnabled());
}
+
+ /**
+ * Verifies that a capacity-driven eviction from the native L1 cache engine
is propagated to the
+ * configured L2 cache engine.
+ * @throws Exception if waiting for an L1 eviction to reach L2 fails
+ */
+ @Test
+ public void testL1CapacityEvictionMovesBlockToL2() throws Exception {
+ this.conf.set(HConstants.BUCKET_CACHE_IOENGINE_KEY, "offheap");
+ this.conf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0.001f);
+ this.conf.setInt(HConstants.BUCKET_CACHE_SIZE_KEY, 100);
Review Comment:
`DataCacheEntry` serializes to 1 MiB, but the default BucketCache's largest
bucket is 513 KiB (`BucketAllocator.java:295-298`). Every block demoted by this
loop is therefore rejected by L2, so `getBlockCount()` never increases and the
test times out. Configure a bucket large enough for this test entry or use a
smaller cacheable.
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/cache/CacheAccessServices.java:
##########
@@ -81,33 +77,59 @@ public static CacheAccessService fromBlockCache(BlockCache
blockCache) {
DefaultHBaseCachePlacementAdmissionPolicy.INSTANCE);
}
+ /**
+ * Creates a {@link CacheAccessService} from the block cache configuration.
+ * @param conf cache configuration
+ * @return configured cache access service, or a disabled service when block
caching is disabled
+ * @throws NullPointerException if {@code conf} is {@code null}
+ */
+ public static CacheAccessService fromConfiguration(Configuration conf) {
+ return fromConfiguration(conf, null);
+ }
+
/**
* Creates a {@link CacheAccessService} from the block cache configuration.
* <p>
- * This method is a compatibility factory for tests and transitional code
paths that want to
- * obtain a {@link CacheAccessService} directly from {@link Configuration},
while still using the
- * existing {@link BlockCacheFactory} and legacy {@link BlockCache}
implementations underneath.
- * </p>
- * <p>
- * The method delegates block-cache construction to
- * {@link BlockCacheFactory#createBlockCache(Configuration)}. If the legacy
factory creates a
- * {@link BlockCache}, the returned service is backed by that cache through
- * {@link TopologyBackedCacheAccessService}. If the legacy factory does not
create a cache, this
- * method returns the disabled/no-op cache access service.
- * </p>
- * <p>
- * This method does not introduce new cache-engine or topology-based runtime
wiring. It is
- * intended only as a bridge while existing HBase tests and integration
paths migrate from direct
- * {@link BlockCache} usage to {@link CacheAccessService}.
+ * Cache implementations that implement {@link CacheEngine} natively are
used directly. Legacy
+ * {@link BlockCache} implementations are adapted to {@link CacheEngine}
until their migration is
+ * complete.
* </p>
- * @param conf configuration used by {@link BlockCacheFactory}
- * @return cache access service created from the configured legacy block
cache, or disabled when
- * no block cache is configured
+ * @param conf cache configuration
+ * @param onlineRegions currently online regions, or {@code null} when
unavailable
+ * @return configured cache access service, or a disabled service when block
caching is disabled
* @throws NullPointerException if {@code conf} is {@code null}
*/
- public static CacheAccessService fromConfiguration(Configuration conf) {
+ public static CacheAccessService fromConfiguration(Configuration conf,
+ Map<String, HRegion> onlineRegions) {
Objects.requireNonNull(conf, "conf must not be null");
- return fromBlockCache(BlockCacheFactory.createBlockCache(conf));
+
+ CacheEngine l1 = BlockCacheFactory.createFirstLevelCacheEngine(conf);
+ if (l1 == null) {
+ return disabled();
+ }
+
+ CachePlacementAdmissionPolicy policy =
DefaultHBaseCachePlacementAdmissionPolicy.INSTANCE;
+
+ boolean useExternal =
conf.getBoolean(BlockCacheFactory.EXTERNAL_BLOCKCACHE_KEY,
+ BlockCacheFactory.EXTERNAL_BLOCKCACHE_DEFAULT);
+
+ if (useExternal) {
+ CacheEngine l2 = BlockCacheFactory.createExternalCacheEngine(conf);
+ if (l2 == null) {
+ return
TopologyBackedCacheAccessServices.fromSingleCacheEngine("single", l1, policy);
+ }
+
+ return
TopologyBackedCacheAccessServices.fromTieredInclusiveCacheEngines("inclusive",
l1, l2,
+ policy);
+ }
+
+ CacheEngine l2 = BlockCacheFactory.createBucketCacheEngine(conf,
onlineRegions);
+ if (l2 == null) {
+ return TopologyBackedCacheAccessServices.fromSingleCacheEngine("single",
l1, policy);
+ }
+
+ return
TopologyBackedCacheAccessServices.fromTieredExclusiveCacheEngines("combined",
l1, l2,
+ policy);
Review Comment:
This new exclusive topology path exposes an existing double update in
`TopologyBackedCacheAccessService`: its exclusive lookup passes
`updateCacheMetrics=true` into the selected engine and then records the same
hit or miss again in `updateBlockMetrics`. Both the native L1 and adapted L2
honor the first update, so every lookup is double-counted. Disable engine-level
updates for that lookup and retain only the service's explicit update, matching
`CombinedBlockCache`.
##########
hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/cache/LruCacheEngine.java:
##########
@@ -0,0 +1,1549 @@
+/*
+ * 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.cache;
+
+import java.lang.ref.WeakReference;
+import java.util.EnumMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.PriorityQueue;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.concurrent.locks.ReentrantLock;
+import org.apache.commons.lang3.mutable.MutableBoolean;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseInterfaceAudience;
+import org.apache.hadoop.hbase.io.HeapSize;
+import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
+import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
+import org.apache.hadoop.hbase.io.hfile.BlockCacheUtil;
+import org.apache.hadoop.hbase.io.hfile.BlockPriority;
+import org.apache.hadoop.hbase.io.hfile.BlockType;
+import org.apache.hadoop.hbase.io.hfile.CacheStats;
+import org.apache.hadoop.hbase.io.hfile.Cacheable;
+import org.apache.hadoop.hbase.io.hfile.CachedBlock;
+import org.apache.hadoop.hbase.io.hfile.HFileBlock;
+import org.apache.hadoop.hbase.io.hfile.LruCachedBlock;
+import org.apache.hadoop.hbase.io.hfile.LruCachedBlockQueue;
+import org.apache.hadoop.hbase.util.ClassSize;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.base.MoreObjects;
+import org.apache.hbase.thirdparty.com.google.common.base.Objects;
+import
org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+/**
+ * Native LRU {@link CacheEngine} implementation.
+ * <p>
+ * This cache is memory-aware using {@link HeapSize}, memory-bound using an
LRU eviction algorithm,
+ * and concurrent. It is backed by a {@link ConcurrentHashMap} and can use a
non-blocking eviction
+ * thread, providing constant-time {@link #cacheBlock(BlockCacheKey,
Cacheable, boolean)} and
+ * {@link #getBlock(BlockCacheKey, boolean, boolean, boolean)} operations.
+ * </p>
+ * <p>
+ * The cache maintains three block-priority levels to provide scan resistance
and support in-memory
+ * column families:
+ * </p>
+ * <ul>
+ * <li>single-access blocks</li>
+ * <li>multiple-access blocks</li>
+ * <li>in-memory blocks</li>
+ * </ul>
+ * <p>
+ * Each priority is assigned a portion of the total cache capacity. During
eviction the cache tries
+ * to preserve the configured relative sizes while allowing unused capacity in
one priority to be
+ * consumed by another.
+ * </p>
+ * <p>
+ * This class is a storage engine only. It does not perform L1/L2
orchestration, victim-cache
+ * delegation, tier placement, admission control, promotion, or demotion.
Those responsibilities
+ * belong to the cache topology and policy layers.
+ * </p>
+ */
[email protected]
+public class LruCacheEngine implements CacheEngine, HeapSize,
Iterable<CachedBlock> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(LruCacheEngine.class);
+
+ /**
+ * Percentage of total size that eviction will evict until.
+ */
+ private static final String LRU_MIN_FACTOR_CONFIG_NAME =
"hbase.lru.blockcache.min.factor";
+
+ /**
+ * Acceptable cache size above which eviction is triggered.
+ */
+ private static final String LRU_ACCEPTABLE_FACTOR_CONFIG_NAME =
+ "hbase.lru.blockcache.acceptable.factor";
+
+ /**
+ * 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 =
+ "hbase.lru.blockcache.hard.capacity.limit.factor";
+
+ private static final String LRU_SINGLE_PERCENTAGE_CONFIG_NAME =
+ "hbase.lru.blockcache.single.percentage";
+
+ private static final String LRU_MULTI_PERCENTAGE_CONFIG_NAME =
+ "hbase.lru.blockcache.multi.percentage";
+
+ private static final String LRU_MEMORY_PERCENTAGE_CONFIG_NAME =
+ "hbase.lru.blockcache.memory.percentage";
+
+ /**
+ * Configuration key that gives data blocks from in-memory HFiles higher
eviction priority.
+ */
+ private static final String LRU_IN_MEMORY_FORCE_MODE_CONFIG_NAME =
+ "hbase.lru.rs.inmemoryforcemode";
+
+ static final float DEFAULT_LOAD_FACTOR = 0.75f;
+ static final int DEFAULT_CONCURRENCY_LEVEL = 16;
+
+ private static final float DEFAULT_MIN_FACTOR = 0.95f;
+ static final float DEFAULT_ACCEPTABLE_FACTOR = 0.99f;
+
+ private static final float DEFAULT_SINGLE_FACTOR = 0.25f;
+ private static final float DEFAULT_MULTI_FACTOR = 0.50f;
+ private static final float DEFAULT_MEMORY_FACTOR = 0.25f;
+
+ private static final float DEFAULT_HARD_CAPACITY_LIMIT_FACTOR = 1.2f;
+
+ private static final boolean DEFAULT_IN_MEMORY_FORCE_MODE = false;
+
+ private static final int STAT_THREAD_PERIOD = 60 * 5;
+
+ private static final String LRU_MAX_BLOCK_SIZE = "hbase.lru.max.block.size";
+
+ private static final long DEFAULT_MAX_BLOCK_SIZE = 16L * 1024L * 1024L;
+
+ /**
+ * Fixed heap overhead of an LRU cache engine instance.
+ */
+ public static final long CACHE_FIXED_OVERHEAD =
+ ClassSize.estimateBase(LruCacheEngine.class, false);
+
+ /**
+ * Cached blocks keyed by their HFile block cache key.
+ * <p>
+ * A {@link ConcurrentHashMap} is required because {@link #getBlock} and
eviction depend on the
+ * atomicity guarantees of {@code computeIfPresent}.
+ * </p>
+ */
+ private transient final ConcurrentHashMap<BlockCacheKey, LruCachedBlock> map;
+
+ /** Lock protecting the eviction process. */
+ private transient final ReentrantLock evictionLock = new ReentrantLock(true);
+
+ /** Maximum size of an individual block accepted by this cache. */
+ private final long maxBlockSize;
+
+ /** Whether an eviction pass is currently running. */
+ private volatile boolean evictionInProgress;
+
+ /** Optional background eviction thread. */
+ private transient final EvictionThread evictionThread;
+
+ /**
+ * Listener notified about capacity-driven block evictions.
+ */
+ private volatile CacheEvictionListener evictionListener;
+
+ /** Executor used to periodically report cache statistics. */
+ private transient final ScheduledExecutorService scheduleThreadPool =
+ Executors.newScheduledThreadPool(1, new ThreadFactoryBuilder()
+ .setNameFormat("LruCacheEngineStatsExecutor").setDaemon(true).build());
+
+ /** Current total heap size used by the cache. */
+ private final AtomicLong size;
+
+ /** Current heap size of data blocks. */
+ private final LongAdder dataBlockSize = new LongAdder();
+
+ /** Current heap size of index blocks. */
+ private final LongAdder indexBlockSize = new LongAdder();
+
+ /** Current heap size of bloom blocks. */
+ private final LongAdder bloomBlockSize = new LongAdder();
+
+ /** Current number of cached blocks. */
+ private final AtomicLong elements;
+
+ /** Current number of cached data blocks. */
+ private final LongAdder dataBlockElements = new LongAdder();
+
+ /** Current number of cached index blocks. */
+ private final LongAdder indexBlockElements = new LongAdder();
+
+ /** Current number of cached bloom blocks. */
+ private final LongAdder bloomBlockElements = new LongAdder();
+
+ /** Sequential cache access identifier. */
+ private final AtomicLong count;
+
+ /** Hard cache capacity limit factor. */
+ private float hardCapacityLimitFactor;
+
+ /** Cache statistics. */
+ private final CacheStats stats;
+
+ /** Maximum cache size in bytes. */
+ private long maxSize;
+
+ /** Expected average block size. */
+ private long blockSize;
+
+ /** Cache size factor at which eviction is triggered. */
+ private float acceptableFactor;
+
+ /** Cache size factor to which an eviction pass should reduce the cache. */
+ private float minFactor;
+
+ /** Fraction of capacity assigned to single-access blocks. */
+ private float singleFactor;
+
+ /** Fraction of capacity assigned to multiple-access blocks. */
+ private float multiFactor;
+
+ /** Fraction of capacity assigned to in-memory blocks. */
+ private float memoryFactor;
+
+ /** Heap overhead of the cache structure itself. */
+ private long overhead;
+
+ /** Whether data blocks from in-memory HFiles receive stronger retention
priority. */
+ private boolean forceInMemory;
+
+ /**
+ * Creates an LRU cache engine using default configuration values.
+ * @param maxSize maximum size of the cache, in bytes
+ * @param blockSize expected average block size, in bytes
+ */
+ public LruCacheEngine(long maxSize, long blockSize) {
+ this(maxSize, blockSize, true);
+ }
+
+ /**
+ * Creates an LRU cache engine and optionally enables the background
eviction thread.
+ * @param maxSize maximum size of the cache, in bytes
+ * @param blockSize expected average block size, in bytes
+ * @param evictionThread whether background eviction should be enabled
+ */
+ public LruCacheEngine(long maxSize, long blockSize, boolean evictionThread) {
+ this(maxSize, blockSize, evictionThread, (int) Math.ceil(1.2 * maxSize /
blockSize),
+ DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL, DEFAULT_MIN_FACTOR,
DEFAULT_ACCEPTABLE_FACTOR,
+ DEFAULT_SINGLE_FACTOR, DEFAULT_MULTI_FACTOR, DEFAULT_MEMORY_FACTOR,
+ DEFAULT_HARD_CAPACITY_LIMIT_FACTOR, false, DEFAULT_MAX_BLOCK_SIZE);
+ }
+
+ /**
+ * Creates an LRU cache engine using values from the supplied configuration.
+ * @param maxSize maximum size of the cache, in bytes
+ * @param blockSize expected average block size, in bytes
+ * @param evictionThread whether background eviction should be enabled
+ * @param conf cache configuration
+ */
+ public LruCacheEngine(long maxSize, long blockSize, boolean evictionThread,
Configuration conf) {
+ this(maxSize, blockSize, evictionThread, (int) Math.ceil(1.2 * maxSize /
blockSize),
+ DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL,
+ conf.getFloat(LRU_MIN_FACTOR_CONFIG_NAME, DEFAULT_MIN_FACTOR),
+ conf.getFloat(LRU_ACCEPTABLE_FACTOR_CONFIG_NAME,
DEFAULT_ACCEPTABLE_FACTOR),
+ conf.getFloat(LRU_SINGLE_PERCENTAGE_CONFIG_NAME, DEFAULT_SINGLE_FACTOR),
+ conf.getFloat(LRU_MULTI_PERCENTAGE_CONFIG_NAME, DEFAULT_MULTI_FACTOR),
+ conf.getFloat(LRU_MEMORY_PERCENTAGE_CONFIG_NAME, DEFAULT_MEMORY_FACTOR),
+ conf.getFloat(LRU_HARD_CAPACITY_LIMIT_FACTOR_CONFIG_NAME,
DEFAULT_HARD_CAPACITY_LIMIT_FACTOR),
+ conf.getBoolean(LRU_IN_MEMORY_FORCE_MODE_CONFIG_NAME,
DEFAULT_IN_MEMORY_FORCE_MODE),
+ conf.getLong(LRU_MAX_BLOCK_SIZE, DEFAULT_MAX_BLOCK_SIZE));
+ }
+
+ /**
+ * Creates an LRU cache engine using the supplied configuration and
background eviction.
+ * @param maxSize maximum size of the cache, in bytes
+ * @param blockSize expected average block size, in bytes
+ * @param conf cache configuration
+ */
+ public LruCacheEngine(long maxSize, long blockSize, Configuration conf) {
+ this(maxSize, blockSize, true, conf);
+ }
+
+ /**
+ * Creates a fully configured LRU cache engine.
+ * @param maxSize maximum size of this cache, in bytes
+ * @param blockSize expected average size of blocks, in bytes
+ * @param evictionThread whether to run eviction in a background thread
+ * @param mapInitialSize initial size of the backing map
+ * @param mapLoadFactor load factor of the backing map
+ * @param mapConcurrencyLevel concurrency level of the backing map
+ * @param minFactor fraction of maximum size retained after
eviction
+ * @param acceptableFactor fraction of maximum size that triggers eviction
+ * @param singleFactor fraction assigned to single-access blocks
+ * @param multiFactor fraction assigned to multiple-access blocks
+ * @param memoryFactor fraction assigned to in-memory blocks
+ * @param hardLimitFactor hard capacity limit factor
+ * @param forceInMemory whether in-memory HFile blocks receive
stronger retention priority
+ * @param maxBlockSize largest individual block accepted by this cache
+ */
+ public LruCacheEngine(long maxSize, long blockSize, boolean evictionThread,
int mapInitialSize,
+ float mapLoadFactor, int mapConcurrencyLevel, float minFactor, float
acceptableFactor,
+ float singleFactor, float multiFactor, float memoryFactor, float
hardLimitFactor,
+ boolean forceInMemory, long maxBlockSize) {
+ this.maxBlockSize = maxBlockSize;
+
+ if (
+ singleFactor + multiFactor + memoryFactor != 1 || singleFactor < 0 ||
multiFactor < 0
+ || memoryFactor < 0
+ ) {
+ throw new IllegalArgumentException(
+ "Single, multi, and memory factors should be non-negative and total
1.0");
+ }
+
+ if (minFactor >= acceptableFactor) {
+ throw new IllegalArgumentException("minFactor must be smaller than
acceptableFactor");
+ }
+
+ if (minFactor >= 1.0f || acceptableFactor >= 1.0f) {
+ throw new IllegalArgumentException("all factors must be < 1");
+ }
+
+ this.maxSize = maxSize;
+ this.blockSize = blockSize;
+ this.forceInMemory = forceInMemory;
+ this.map = new ConcurrentHashMap<>(mapInitialSize, mapLoadFactor,
mapConcurrencyLevel);
+ this.minFactor = minFactor;
+ this.acceptableFactor = acceptableFactor;
+ this.singleFactor = singleFactor;
+ this.multiFactor = multiFactor;
+ this.memoryFactor = memoryFactor;
+ this.stats = new CacheStats(getClass().getSimpleName());
+ this.count = new AtomicLong(0);
+ this.elements = new AtomicLong(0);
+ this.overhead = calculateOverhead(maxSize, blockSize, mapConcurrencyLevel);
+ this.size = new AtomicLong(this.overhead);
+ this.hardCapacityLimitFactor = hardLimitFactor;
+
+ if (evictionThread) {
+ this.evictionThread = new EvictionThread(this);
+ this.evictionThread.start();
+ } else {
+ this.evictionThread = null;
+ }
+
+ this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this),
STAT_THREAD_PERIOD,
+ STAT_THREAD_PERIOD, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Returns the human-readable name of this cache engine.
+ * @return cache engine name
+ */
+ @Override
+ public String getName() {
+ return getClass().getSimpleName();
+ }
+
+ /**
+ * Updates the maximum size of this cache.
+ * <p>
+ * If the cache is already larger than the new acceptable size, an eviction
pass is started.
+ * </p>
+ * @param maxSize new maximum size, in bytes
+ */
+ public void setMaxSize(long maxSize) {
+ this.maxSize = maxSize;
+ if (size.get() > acceptableSize() && !evictionInProgress) {
+ runEviction();
+ }
+ }
+
+ /**
+ * Returns a heap-backed reference suitable for storage in this cache.
+ * <p>
+ * Shared-memory {@link HFileBlock}s are cloned onto the heap. Other blocks
are retained before
+ * being referenced by the cache.
+ * </p>
+ * @param buf block to convert
+ * @return heap-backed retained block
+ */
+ private Cacheable asReferencedHeapBlock(Cacheable buf) {
+ if (buf instanceof HFileBlock) {
+ HFileBlock block = (HFileBlock) buf;
+ if (block.isSharedMem()) {
+ return HFileBlock.deepCloneOnHeap(block);
+ }
+ }
+
+ return buf.retain();
+ }
+
+ /**
+ * Caches the specified block.
+ * @param cacheKey block cache key
+ * @param buf block contents
+ * @param inMemory whether the block should receive in-memory priority
+ */
+ @Override
+ public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean
inMemory) {
+ if (buf.heapSize() > maxBlockSize) {
+ if (stats.failInsert() % 50 == 0) {
+ LOG.warn("Trying to cache too large a block " +
cacheKey.getHfileName() + " @ "
+ + cacheKey.getOffset() + " is " + buf.heapSize() + " which is larger
than "
+ + maxBlockSize);
+ }
+ return;
+ }
+
+ LruCachedBlock cachedBlock = map.get(cacheKey);
+ if (
+ cachedBlock != null &&
!BlockCacheUtil.shouldReplaceExistingCacheBlock(this, cacheKey, buf)
+ ) {
+ return;
+ }
+
+ long currentSize = size.get();
+ long currentAcceptableSize = acceptableSize();
+ long hardLimitSize = (long) (hardCapacityLimitFactor *
currentAcceptableSize);
+
+ if (currentSize >= hardLimitSize) {
+ stats.failInsert();
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("LruCacheEngine current size " +
StringUtils.byteDesc(currentSize)
+ + " has exceeded acceptable size " +
StringUtils.byteDesc(currentAcceptableSize) + "."
+ + " The hard limit size is " + StringUtils.byteDesc(hardLimitSize)
+ + ", failed to put cacheKey:" + cacheKey + " into LruCacheEngine.");
+ }
+ if (!evictionInProgress) {
+ runEviction();
+ }
+ return;
+ }
+
+ Cacheable referencedBlock = asReferencedHeapBlock(buf);
+ cachedBlock = new LruCachedBlock(cacheKey, referencedBlock,
count.incrementAndGet(), inMemory);
+
+ long newSize = updateSizeMetrics(cachedBlock, false);
+ map.put(cacheKey, cachedBlock);
+
+ long elementCount = elements.incrementAndGet();
Review Comment:
When `shouldReplaceExistingCacheBlock` accepts the HBASE-20447 replacement
case, this `map.put` overwrites the previous `LruCachedBlock` without releasing
its retained buffer or subtracting its size, element, and block-type counters.
Repeated replacements leak references and inflate occupancy, which can trigger
incorrect capacity decisions. Replace the mapping atomically and apply removal
accounting to the actual previous entry before publishing/counting the
replacement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
