| /* |
| * 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.bucket; |
| |
| import static org.apache.hadoop.hbase.io.hfile.CacheConfig.BUCKETCACHE_PERSIST_INTERVAL_KEY; |
| |
| import java.io.File; |
| import java.io.FileInputStream; |
| import java.io.FileOutputStream; |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.NavigableSet; |
| import java.util.Optional; |
| import java.util.PriorityQueue; |
| import java.util.Set; |
| import java.util.concurrent.ArrayBlockingQueue; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.ConcurrentSkipListSet; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicLong; |
| import java.util.concurrent.atomic.LongAdder; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.ReentrantLock; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| import java.util.function.Consumer; |
| import java.util.function.Function; |
| import org.apache.commons.lang3.mutable.MutableInt; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.hbase.HBaseConfiguration; |
| import org.apache.hadoop.hbase.HBaseIOException; |
| import org.apache.hadoop.hbase.TableName; |
| import org.apache.hadoop.hbase.client.Admin; |
| import org.apache.hadoop.hbase.io.ByteBuffAllocator; |
| import org.apache.hadoop.hbase.io.ByteBuffAllocator.Recycler; |
| import org.apache.hadoop.hbase.io.HeapSize; |
| import org.apache.hadoop.hbase.io.hfile.BlockCache; |
| 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.CacheConfig; |
| 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.CombinedBlockCache; |
| import org.apache.hadoop.hbase.io.hfile.HFileBlock; |
| import org.apache.hadoop.hbase.io.hfile.HFileContext; |
| import org.apache.hadoop.hbase.nio.ByteBuff; |
| import org.apache.hadoop.hbase.nio.RefCnt; |
| import org.apache.hadoop.hbase.protobuf.ProtobufMagic; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; |
| import org.apache.hadoop.hbase.util.IdReadWriteLock; |
| import org.apache.hadoop.hbase.util.IdReadWriteLockStrongRef; |
| import org.apache.hadoop.hbase.util.IdReadWriteLockWithObjectPool; |
| import org.apache.hadoop.hbase.util.IdReadWriteLockWithObjectPool.ReferenceType; |
| import org.apache.hadoop.hbase.util.Pair; |
| 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.Preconditions; |
| import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder; |
| |
| import org.apache.hadoop.hbase.shaded.protobuf.generated.BucketCacheProtos; |
| |
| /** |
| * BucketCache uses {@link BucketAllocator} to allocate/free blocks, and uses BucketCache#ramCache |
| * and BucketCache#backingMap in order to determine if a given element is in the cache. The bucket |
| * cache can use off-heap memory {@link ByteBufferIOEngine} or mmap |
| * {@link ExclusiveMemoryMmapIOEngine} or pmem {@link SharedMemoryMmapIOEngine} or local files |
| * {@link FileIOEngine} to store/read the block data. |
| * <p> |
| * Eviction is via a similar algorithm as used in |
| * {@link org.apache.hadoop.hbase.io.hfile.LruBlockCache} |
| * <p> |
| * BucketCache can be used as mainly a block cache (see |
| * {@link org.apache.hadoop.hbase.io.hfile.CombinedBlockCache}), combined with a BlockCache to |
| * decrease CMS GC and heap fragmentation. |
| * <p> |
| * It also can be used as a secondary cache (e.g. using a file on ssd/fusionio to store blocks) to |
| * enlarge cache space via a victim cache. |
| */ |
| @InterfaceAudience.Private |
| public class BucketCache implements BlockCache, HeapSize { |
| private static final Logger LOG = LoggerFactory.getLogger(BucketCache.class); |
| |
| /** Priority buckets config */ |
| static final String SINGLE_FACTOR_CONFIG_NAME = "hbase.bucketcache.single.factor"; |
| static final String MULTI_FACTOR_CONFIG_NAME = "hbase.bucketcache.multi.factor"; |
| static final String MEMORY_FACTOR_CONFIG_NAME = "hbase.bucketcache.memory.factor"; |
| static final String EXTRA_FREE_FACTOR_CONFIG_NAME = "hbase.bucketcache.extrafreefactor"; |
| static final String ACCEPT_FACTOR_CONFIG_NAME = "hbase.bucketcache.acceptfactor"; |
| static final String MIN_FACTOR_CONFIG_NAME = "hbase.bucketcache.minfactor"; |
| |
| /** Use strong reference for offsetLock or not */ |
| private static final String STRONG_REF_KEY = "hbase.bucketcache.offsetlock.usestrongref"; |
| private static final boolean STRONG_REF_DEFAULT = false; |
| |
| /** Priority buckets */ |
| static final float DEFAULT_SINGLE_FACTOR = 0.25f; |
| static final float DEFAULT_MULTI_FACTOR = 0.50f; |
| static final float DEFAULT_MEMORY_FACTOR = 0.25f; |
| static final float DEFAULT_MIN_FACTOR = 0.85f; |
| |
| private static final float DEFAULT_EXTRA_FREE_FACTOR = 0.10f; |
| private static final float DEFAULT_ACCEPT_FACTOR = 0.95f; |
| |
| // Number of blocks to clear for each of the bucket size that is full |
| private static final int DEFAULT_FREE_ENTIRE_BLOCK_FACTOR = 2; |
| |
| /** Statistics thread */ |
| private static final int statThreadPeriod = 5 * 60; |
| |
| final static int DEFAULT_WRITER_THREADS = 3; |
| final static int DEFAULT_WRITER_QUEUE_ITEMS = 64; |
| |
| // Store/read block data |
| transient final IOEngine ioEngine; |
| |
| // Store the block in this map before writing it to cache |
| transient final RAMCache ramCache; |
| |
| // In this map, store the block's meta data like offset, length |
| transient Map<BlockCacheKey, BucketEntry> backingMap; |
| |
| private AtomicBoolean backingMapValidated = new AtomicBoolean(false); |
| |
| /** |
| * Map of hFile -> Region -> File size. This map is used to track all files completed prefetch, |
| * together with the region those belong to and the total cached size for the |
| * region.TestBlockEvictionOnRegionMovement |
| */ |
| final Map<String, Pair<String, Long>> fullyCachedFiles = new ConcurrentHashMap<>(); |
| /** |
| * Map of region -> total size of the region prefetched on this region server. This is the total |
| * size of hFiles for this region prefetched on this region server |
| */ |
| final Map<String, Long> regionCachedSize = new ConcurrentHashMap<>(); |
| |
| private BucketCachePersister cachePersister; |
| |
| /** |
| * Flag if the cache is enabled or not... We shut it off if there are IO errors for some time, so |
| * that Bucket IO exceptions/errors don't bring down the HBase server. |
| */ |
| private volatile boolean cacheEnabled; |
| |
| /** |
| * A list of writer queues. We have a queue per {@link WriterThread} we have running. In other |
| * words, the work adding blocks to the BucketCache is divided up amongst the running |
| * WriterThreads. Its done by taking hash of the cache key modulo queue count. WriterThread when |
| * it runs takes whatever has been recently added and 'drains' the entries to the BucketCache. It |
| * then updates the ramCache and backingMap accordingly. |
| */ |
| transient final ArrayList<BlockingQueue<RAMQueueEntry>> writerQueues = new ArrayList<>(); |
| transient final WriterThread[] writerThreads; |
| |
| /** Volatile boolean to track if free space is in process or not */ |
| private volatile boolean freeInProgress = false; |
| private transient final Lock freeSpaceLock = new ReentrantLock(); |
| |
| private final LongAdder realCacheSize = new LongAdder(); |
| private final LongAdder heapSize = new LongAdder(); |
| /** Current number of cached elements */ |
| private final LongAdder blockNumber = new LongAdder(); |
| |
| /** Cache access count (sequential ID) */ |
| private final AtomicLong accessCount = new AtomicLong(); |
| |
| private static final int DEFAULT_CACHE_WAIT_TIME = 50; |
| |
| private final BucketCacheStats cacheStats = new BucketCacheStats(); |
| private final String persistencePath; |
| static AtomicBoolean isCacheInconsistent = new AtomicBoolean(false); |
| private final long cacheCapacity; |
| /** Approximate block size */ |
| private final long blockSize; |
| |
| /** Duration of IO errors tolerated before we disable cache, 1 min as default */ |
| private final int ioErrorsTolerationDuration; |
| // 1 min |
| public static final int DEFAULT_ERROR_TOLERATION_DURATION = 60 * 1000; |
| |
| // Start time of first IO error when reading or writing IO Engine, it will be |
| // reset after a successful read/write. |
| private volatile long ioErrorStartTime = -1; |
| |
| /** |
| * A ReentrantReadWriteLock to lock on a particular block identified by offset. The purpose of |
| * this is to avoid freeing the block which is being read. |
| * <p> |
| */ |
| transient final IdReadWriteLock<Long> offsetLock; |
| |
| NavigableSet<BlockCacheKey> blocksByHFile = new ConcurrentSkipListSet<>( |
| Comparator.comparing(BlockCacheKey::getHfileName).thenComparingLong(BlockCacheKey::getOffset)); |
| |
| /** Statistics thread schedule pool (for heavy debugging, could remove) */ |
| private transient final ScheduledExecutorService scheduleThreadPool = |
| Executors.newScheduledThreadPool(1, |
| new ThreadFactoryBuilder().setNameFormat("BucketCacheStatsExecutor").setDaemon(true).build()); |
| |
| // Allocate or free space for the block |
| private transient BucketAllocator bucketAllocator; |
| |
| /** Acceptable size of cache (no evictions if size < acceptable) */ |
| private float acceptableFactor; |
| |
| /** Minimum threshold of cache (when evicting, evict until size < min) */ |
| private float minFactor; |
| |
| /** |
| * Free this floating point factor of extra blocks when evicting. For example free the number of |
| * blocks requested * (1 + extraFreeFactor) |
| */ |
| private float extraFreeFactor; |
| |
| /** Single access bucket size */ |
| private float singleFactor; |
| |
| /** Multiple access bucket size */ |
| private float multiFactor; |
| |
| /** In-memory bucket size */ |
| private float memoryFactor; |
| |
| private long bucketcachePersistInterval; |
| |
| private static final String FILE_VERIFY_ALGORITHM = |
| "hbase.bucketcache.persistent.file.integrity.check.algorithm"; |
| private static final String DEFAULT_FILE_VERIFY_ALGORITHM = "MD5"; |
| |
| private static final String QUEUE_ADDITION_WAIT_TIME = |
| "hbase.bucketcache.queue.addition.waittime"; |
| private static final long DEFAULT_QUEUE_ADDITION_WAIT_TIME = 0; |
| private long queueAdditionWaitTime; |
| /** |
| * Use {@link java.security.MessageDigest} class's encryption algorithms to check persistent file |
| * integrity, default algorithm is MD5 |
| */ |
| private String algorithm; |
| |
| /* Tracing failed Bucket Cache allocations. */ |
| private long allocFailLogPrevTs; // time of previous log event for allocation failure. |
| private static final int ALLOCATION_FAIL_LOG_TIME_PERIOD = 60000; // Default 1 minute. |
| |
| public BucketCache(String ioEngineName, long capacity, int blockSize, int[] bucketSizes, |
| int writerThreadNum, int writerQLen, String persistencePath) throws IOException { |
| this(ioEngineName, capacity, blockSize, bucketSizes, writerThreadNum, writerQLen, |
| persistencePath, DEFAULT_ERROR_TOLERATION_DURATION, HBaseConfiguration.create()); |
| } |
| |
| public BucketCache(String ioEngineName, long capacity, int blockSize, int[] bucketSizes, |
| int writerThreadNum, int writerQLen, String persistencePath, int ioErrorsTolerationDuration, |
| Configuration conf) throws IOException { |
| boolean useStrongRef = conf.getBoolean(STRONG_REF_KEY, STRONG_REF_DEFAULT); |
| if (useStrongRef) { |
| this.offsetLock = new IdReadWriteLockStrongRef<>(); |
| } else { |
| this.offsetLock = new IdReadWriteLockWithObjectPool<>(ReferenceType.SOFT); |
| } |
| this.algorithm = conf.get(FILE_VERIFY_ALGORITHM, DEFAULT_FILE_VERIFY_ALGORITHM); |
| this.ioEngine = getIOEngineFromName(ioEngineName, capacity, persistencePath); |
| this.writerThreads = new WriterThread[writerThreadNum]; |
| long blockNumCapacity = capacity / blockSize; |
| if (blockNumCapacity >= Integer.MAX_VALUE) { |
| // Enough for about 32TB of cache! |
| throw new IllegalArgumentException("Cache capacity is too large, only support 32TB now"); |
| } |
| |
| this.acceptableFactor = conf.getFloat(ACCEPT_FACTOR_CONFIG_NAME, DEFAULT_ACCEPT_FACTOR); |
| this.minFactor = conf.getFloat(MIN_FACTOR_CONFIG_NAME, DEFAULT_MIN_FACTOR); |
| this.extraFreeFactor = conf.getFloat(EXTRA_FREE_FACTOR_CONFIG_NAME, DEFAULT_EXTRA_FREE_FACTOR); |
| this.singleFactor = conf.getFloat(SINGLE_FACTOR_CONFIG_NAME, DEFAULT_SINGLE_FACTOR); |
| this.multiFactor = conf.getFloat(MULTI_FACTOR_CONFIG_NAME, DEFAULT_MULTI_FACTOR); |
| this.memoryFactor = conf.getFloat(MEMORY_FACTOR_CONFIG_NAME, DEFAULT_MEMORY_FACTOR); |
| this.queueAdditionWaitTime = |
| conf.getLong(QUEUE_ADDITION_WAIT_TIME, DEFAULT_QUEUE_ADDITION_WAIT_TIME); |
| this.bucketcachePersistInterval = conf.getLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, 1000); |
| |
| sanityCheckConfigs(); |
| |
| LOG.info("Instantiating BucketCache with acceptableFactor: " + acceptableFactor |
| + ", minFactor: " + minFactor + ", extraFreeFactor: " + extraFreeFactor + ", singleFactor: " |
| + singleFactor + ", multiFactor: " + multiFactor + ", memoryFactor: " + memoryFactor |
| + ", useStrongRef: " + useStrongRef); |
| |
| this.cacheCapacity = capacity; |
| this.persistencePath = persistencePath; |
| this.blockSize = blockSize; |
| this.ioErrorsTolerationDuration = ioErrorsTolerationDuration; |
| |
| this.allocFailLogPrevTs = 0; |
| |
| for (int i = 0; i < writerThreads.length; ++i) { |
| writerQueues.add(new ArrayBlockingQueue<>(writerQLen)); |
| } |
| |
| assert writerQueues.size() == writerThreads.length; |
| this.ramCache = new RAMCache(); |
| |
| this.backingMap = new ConcurrentHashMap<>((int) blockNumCapacity); |
| |
| if (isCachePersistent()) { |
| if (ioEngine instanceof FileIOEngine) { |
| startBucketCachePersisterThread(); |
| } |
| try { |
| retrieveFromFile(bucketSizes); |
| } catch (IOException ioex) { |
| LOG.error("Can't restore from file[{}] because of ", persistencePath, ioex); |
| backingMap.clear(); |
| fullyCachedFiles.clear(); |
| backingMapValidated.set(true); |
| bucketAllocator = new BucketAllocator(capacity, bucketSizes); |
| regionCachedSize.clear(); |
| } |
| } else { |
| bucketAllocator = new BucketAllocator(capacity, bucketSizes); |
| } |
| final String threadName = Thread.currentThread().getName(); |
| this.cacheEnabled = true; |
| for (int i = 0; i < writerThreads.length; ++i) { |
| writerThreads[i] = new WriterThread(writerQueues.get(i)); |
| writerThreads[i].setName(threadName + "-BucketCacheWriter-" + i); |
| writerThreads[i].setDaemon(true); |
| } |
| startWriterThreads(); |
| |
| // Run the statistics thread periodically to print the cache statistics log |
| // TODO: Add means of turning this off. Bit obnoxious running thread just to make a log |
| // every five minutes. |
| this.scheduleThreadPool.scheduleAtFixedRate(new StatisticsThread(this), statThreadPeriod, |
| statThreadPeriod, TimeUnit.SECONDS); |
| LOG.info("Started bucket cache; ioengine=" + ioEngineName + ", capacity=" |
| + StringUtils.byteDesc(capacity) + ", blockSize=" + StringUtils.byteDesc(blockSize) |
| + ", writerThreadNum=" + writerThreadNum + ", writerQLen=" + writerQLen + ", persistencePath=" |
| + persistencePath + ", bucketAllocator=" + this.bucketAllocator.getClass().getName()); |
| } |
| |
| private void sanityCheckConfigs() { |
| Preconditions.checkArgument(acceptableFactor <= 1 && acceptableFactor >= 0, |
| ACCEPT_FACTOR_CONFIG_NAME + " must be between 0.0 and 1.0"); |
| Preconditions.checkArgument(minFactor <= 1 && minFactor >= 0, |
| MIN_FACTOR_CONFIG_NAME + " must be between 0.0 and 1.0"); |
| Preconditions.checkArgument(minFactor <= acceptableFactor, |
| MIN_FACTOR_CONFIG_NAME + " must be <= " + ACCEPT_FACTOR_CONFIG_NAME); |
| Preconditions.checkArgument(extraFreeFactor >= 0, |
| EXTRA_FREE_FACTOR_CONFIG_NAME + " must be greater than 0.0"); |
| Preconditions.checkArgument(singleFactor <= 1 && singleFactor >= 0, |
| SINGLE_FACTOR_CONFIG_NAME + " must be between 0.0 and 1.0"); |
| Preconditions.checkArgument(multiFactor <= 1 && multiFactor >= 0, |
| MULTI_FACTOR_CONFIG_NAME + " must be between 0.0 and 1.0"); |
| Preconditions.checkArgument(memoryFactor <= 1 && memoryFactor >= 0, |
| MEMORY_FACTOR_CONFIG_NAME + " must be between 0.0 and 1.0"); |
| Preconditions.checkArgument((singleFactor + multiFactor + memoryFactor) == 1, |
| SINGLE_FACTOR_CONFIG_NAME + ", " + MULTI_FACTOR_CONFIG_NAME + ", and " |
| + MEMORY_FACTOR_CONFIG_NAME + " segments must add up to 1.0"); |
| } |
| |
| /** |
| * Called by the constructor to start the writer threads. Used by tests that need to override |
| * starting the threads. |
| */ |
| protected void startWriterThreads() { |
| for (WriterThread thread : writerThreads) { |
| thread.start(); |
| } |
| } |
| |
| void startBucketCachePersisterThread() { |
| LOG.info("Starting BucketCachePersisterThread"); |
| cachePersister = new BucketCachePersister(this, bucketcachePersistInterval); |
| cachePersister.setDaemon(true); |
| cachePersister.start(); |
| } |
| |
| boolean isCacheEnabled() { |
| return this.cacheEnabled; |
| } |
| |
| @Override |
| public long getMaxSize() { |
| return this.cacheCapacity; |
| } |
| |
| public String getIoEngine() { |
| return ioEngine.toString(); |
| } |
| |
| /** |
| * Get the IOEngine from the IO engine name |
| * @return the IOEngine |
| */ |
| private IOEngine getIOEngineFromName(String ioEngineName, long capacity, String persistencePath) |
| throws IOException { |
| if (ioEngineName.startsWith("file:") || ioEngineName.startsWith("files:")) { |
| // In order to make the usage simple, we only need the prefix 'files:' in |
| // document whether one or multiple file(s), but also support 'file:' for |
| // the compatibility |
| String[] filePaths = |
| ioEngineName.substring(ioEngineName.indexOf(":") + 1).split(FileIOEngine.FILE_DELIMITER); |
| return new FileIOEngine(capacity, persistencePath != null, filePaths); |
| } else if (ioEngineName.startsWith("offheap")) { |
| return new ByteBufferIOEngine(capacity); |
| } else if (ioEngineName.startsWith("mmap:")) { |
| return new ExclusiveMemoryMmapIOEngine(ioEngineName.substring(5), capacity); |
| } else if (ioEngineName.startsWith("pmem:")) { |
| // This mode of bucket cache creates an IOEngine over a file on the persistent memory |
| // device. Since the persistent memory device has its own address space the contents |
| // mapped to this address space does not get swapped out like in the case of mmapping |
| // on to DRAM. Hence the cells created out of the hfile blocks in the pmem bucket cache |
| // can be directly referred to without having to copy them onheap. Once the RPC is done, |
| // the blocks can be returned back as in case of ByteBufferIOEngine. |
| return new SharedMemoryMmapIOEngine(ioEngineName.substring(5), capacity); |
| } else { |
| throw new IllegalArgumentException( |
| "Don't understand io engine name for cache- prefix with file:, files:, mmap: or offheap"); |
| } |
| } |
| |
| public boolean isCachePersistenceEnabled() { |
| return persistencePath != null; |
| } |
| |
| /** |
| * Cache the block with the specified name and buffer. |
| * @param cacheKey block's cache key |
| * @param buf block buffer |
| */ |
| @Override |
| public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) { |
| cacheBlock(cacheKey, buf, false); |
| } |
| |
| /** |
| * Cache the block with the specified name and buffer. |
| * @param cacheKey block's cache key |
| * @param cachedItem block buffer |
| * @param inMemory if block is in-memory |
| */ |
| @Override |
| public void cacheBlock(BlockCacheKey cacheKey, Cacheable cachedItem, boolean inMemory) { |
| cacheBlockWithWait(cacheKey, cachedItem, inMemory, false); |
| } |
| |
| /** |
| * Cache the block with the specified name and buffer. |
| * @param cacheKey block's cache key |
| * @param cachedItem block buffer |
| * @param inMemory if block is in-memory |
| */ |
| @Override |
| public void cacheBlock(BlockCacheKey cacheKey, Cacheable cachedItem, boolean inMemory, |
| boolean waitWhenCache) { |
| cacheBlockWithWait(cacheKey, cachedItem, inMemory, waitWhenCache && queueAdditionWaitTime > 0); |
| } |
| |
| /** |
| * Cache the block to ramCache |
| * @param cacheKey block's cache key |
| * @param cachedItem block buffer |
| * @param inMemory if block is in-memory |
| * @param wait if true, blocking wait when queue is full |
| */ |
| public void cacheBlockWithWait(BlockCacheKey cacheKey, Cacheable cachedItem, boolean inMemory, |
| boolean wait) { |
| if (cacheEnabled) { |
| if (backingMap.containsKey(cacheKey) || ramCache.containsKey(cacheKey)) { |
| if (shouldReplaceExistingCacheBlock(cacheKey, cachedItem)) { |
| BucketEntry bucketEntry = backingMap.get(cacheKey); |
| if (bucketEntry != null && bucketEntry.isRpcRef()) { |
| // avoid replace when there are RPC refs for the bucket entry in bucket cache |
| return; |
| } |
| cacheBlockWithWaitInternal(cacheKey, cachedItem, inMemory, wait); |
| } |
| } else { |
| cacheBlockWithWaitInternal(cacheKey, cachedItem, inMemory, wait); |
| } |
| } |
| } |
| |
| protected boolean shouldReplaceExistingCacheBlock(BlockCacheKey cacheKey, Cacheable newBlock) { |
| return BlockCacheUtil.shouldReplaceExistingCacheBlock(this, cacheKey, newBlock); |
| } |
| |
| protected void cacheBlockWithWaitInternal(BlockCacheKey cacheKey, Cacheable cachedItem, |
| boolean inMemory, boolean wait) { |
| if (!cacheEnabled) { |
| return; |
| } |
| if (cacheKey.getBlockType() == null && cachedItem.getBlockType() != null) { |
| cacheKey.setBlockType(cachedItem.getBlockType()); |
| } |
| LOG.trace("Caching key={}, item={}", cacheKey, cachedItem); |
| // Stuff the entry into the RAM cache so it can get drained to the persistent store |
| RAMQueueEntry re = new RAMQueueEntry(cacheKey, cachedItem, accessCount.incrementAndGet(), |
| inMemory, isCachePersistent() && ioEngine instanceof FileIOEngine); |
| /** |
| * Don't use ramCache.put(cacheKey, re) here. because there may be a existing entry with same |
| * key in ramCache, the heap size of bucket cache need to update if replacing entry from |
| * ramCache. But WriterThread will also remove entry from ramCache and update heap size, if |
| * using ramCache.put(), It's possible that the removed entry in WriterThread is not the correct |
| * one, then the heap size will mess up (HBASE-20789) |
| */ |
| if (ramCache.putIfAbsent(cacheKey, re) != null) { |
| return; |
| } |
| int queueNum = (cacheKey.hashCode() & 0x7FFFFFFF) % writerQueues.size(); |
| BlockingQueue<RAMQueueEntry> bq = writerQueues.get(queueNum); |
| boolean successfulAddition = false; |
| if (wait) { |
| try { |
| successfulAddition = bq.offer(re, queueAdditionWaitTime, TimeUnit.MILLISECONDS); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| } else { |
| successfulAddition = bq.offer(re); |
| } |
| if (!successfulAddition) { |
| ramCache.remove(cacheKey); |
| cacheStats.failInsert(); |
| } else { |
| this.blockNumber.increment(); |
| this.heapSize.add(cachedItem.heapSize()); |
| } |
| } |
| |
| /** |
| * Get the buffer of the block with the specified key. |
| * @param key block's cache key |
| * @param caching true if the caller caches blocks on cache misses |
| * @param repeat Whether this is a repeat lookup for the same block |
| * @param updateCacheMetrics Whether we should update cache metrics or not |
| * @return buffer of specified cache key, or null if not in cache |
| */ |
| @Override |
| public Cacheable getBlock(BlockCacheKey key, boolean caching, boolean repeat, |
| boolean updateCacheMetrics) { |
| if (!cacheEnabled) { |
| return null; |
| } |
| RAMQueueEntry re = ramCache.get(key); |
| if (re != null) { |
| if (updateCacheMetrics) { |
| cacheStats.hit(caching, key.isPrimary(), key.getBlockType()); |
| } |
| re.access(accessCount.incrementAndGet()); |
| return re.getData(); |
| } |
| BucketEntry bucketEntry = backingMap.get(key); |
| if (bucketEntry != null) { |
| long start = System.nanoTime(); |
| ReentrantReadWriteLock lock = offsetLock.getLock(bucketEntry.offset()); |
| try { |
| lock.readLock().lock(); |
| // We can not read here even if backingMap does contain the given key because its offset |
| // maybe changed. If we lock BlockCacheKey instead of offset, then we can only check |
| // existence here. |
| if (bucketEntry.equals(backingMap.get(key))) { |
| // Read the block from IOEngine based on the bucketEntry's offset and length, NOTICE: the |
| // block will use the refCnt of bucketEntry, which means if two HFileBlock mapping to |
| // the same BucketEntry, then all of the three will share the same refCnt. |
| Cacheable cachedBlock = ioEngine.read(bucketEntry); |
| if (ioEngine.usesSharedMemory()) { |
| // If IOEngine use shared memory, cachedBlock and BucketEntry will share the |
| // same RefCnt, do retain here, in order to count the number of RPC references |
| cachedBlock.retain(); |
| } |
| // Update the cache statistics. |
| if (updateCacheMetrics) { |
| cacheStats.hit(caching, key.isPrimary(), key.getBlockType()); |
| cacheStats.ioHit(System.nanoTime() - start); |
| } |
| bucketEntry.access(accessCount.incrementAndGet()); |
| if (this.ioErrorStartTime > 0) { |
| ioErrorStartTime = -1; |
| } |
| return cachedBlock; |
| } |
| } catch (HBaseIOException hioex) { |
| // When using file io engine persistent cache, |
| // the cache map state might differ from the actual cache. If we reach this block, |
| // we should remove the cache key entry from the backing map |
| backingMap.remove(key); |
| fullyCachedFiles.remove(key.getHfileName()); |
| LOG.debug("Failed to fetch block for cache key: {}.", key, hioex); |
| } catch (IOException ioex) { |
| LOG.error("Failed reading block " + key + " from bucket cache", ioex); |
| checkIOErrorIsTolerated(); |
| } finally { |
| lock.readLock().unlock(); |
| } |
| } |
| if (!repeat && updateCacheMetrics) { |
| cacheStats.miss(caching, key.isPrimary(), key.getBlockType()); |
| } |
| return null; |
| } |
| |
| /** |
| * This method is invoked after the bucketEntry is removed from {@link BucketCache#backingMap} |
| */ |
| void blockEvicted(BlockCacheKey cacheKey, BucketEntry bucketEntry, boolean decrementBlockNumber, |
| boolean evictedByEvictionProcess) { |
| bucketEntry.markAsEvicted(); |
| blocksByHFile.remove(cacheKey); |
| if (decrementBlockNumber) { |
| this.blockNumber.decrement(); |
| if (ioEngine.isPersistent()) { |
| fileNotFullyCached(cacheKey.getHfileName()); |
| } |
| } |
| if (evictedByEvictionProcess) { |
| cacheStats.evicted(bucketEntry.getCachedTime(), cacheKey.isPrimary()); |
| } |
| if (ioEngine.isPersistent()) { |
| setCacheInconsistent(true); |
| } |
| } |
| |
| private void fileNotFullyCached(String hfileName) { |
| // Update the regionPrefetchedSizeMap before removing the file from prefetchCompleted |
| if (fullyCachedFiles.containsKey(hfileName)) { |
| Pair<String, Long> regionEntry = fullyCachedFiles.get(hfileName); |
| String regionEncodedName = regionEntry.getFirst(); |
| long filePrefetchSize = regionEntry.getSecond(); |
| LOG.debug("Removing file {} for region {}", hfileName, regionEncodedName); |
| regionCachedSize.computeIfPresent(regionEncodedName, (rn, pf) -> pf - filePrefetchSize); |
| // If all the blocks for a region are evicted from the cache, remove the entry for that region |
| if ( |
| regionCachedSize.containsKey(regionEncodedName) |
| && regionCachedSize.get(regionEncodedName) == 0 |
| ) { |
| regionCachedSize.remove(regionEncodedName); |
| } |
| } |
| fullyCachedFiles.remove(hfileName); |
| } |
| |
| public void fileCacheCompleted(Path filePath, long size) { |
| Pair<String, Long> pair = new Pair<>(); |
| // sets the region name |
| String regionName = filePath.getParent().getParent().getName(); |
| pair.setFirst(regionName); |
| pair.setSecond(size); |
| fullyCachedFiles.put(filePath.getName(), pair); |
| } |
| |
| private void updateRegionCachedSize(Path filePath, long cachedSize) { |
| if (filePath != null) { |
| String regionName = filePath.getParent().getParent().getName(); |
| regionCachedSize.merge(regionName, cachedSize, |
| (previousSize, newBlockSize) -> previousSize + newBlockSize); |
| } |
| } |
| |
| /** |
| * Free the {{@link BucketEntry} actually,which could only be invoked when the |
| * {@link BucketEntry#refCnt} becoming 0. |
| */ |
| void freeBucketEntry(BucketEntry bucketEntry) { |
| bucketAllocator.freeBlock(bucketEntry.offset(), bucketEntry.getLength()); |
| realCacheSize.add(-1 * bucketEntry.getLength()); |
| } |
| |
| /** |
| * Try to evict the block from {@link BlockCache} by force. We'll call this in few cases:<br> |
| * 1. Close an HFile, and clear all cached blocks. <br> |
| * 2. Call {@link Admin#clearBlockCache(TableName)} to clear all blocks for a given table.<br> |
| * <p> |
| * Firstly, we'll try to remove the block from RAMCache,and then try to evict from backingMap. |
| * Here we evict the block from backingMap immediately, but only free the reference from bucket |
| * cache by calling {@link BucketEntry#markedAsEvicted}. If there're still some RPC referring this |
| * block, block can only be de-allocated when all of them release the block. |
| * <p> |
| * NOTICE: we need to grab the write offset lock firstly before releasing the reference from |
| * bucket cache. if we don't, we may read an {@link BucketEntry} with refCnt = 0 when |
| * {@link BucketCache#getBlock(BlockCacheKey, boolean, boolean, boolean)}, it's a memory leak. |
| * @param cacheKey Block to evict |
| * @return true to indicate whether we've evicted successfully or not. |
| */ |
| @Override |
| public boolean evictBlock(BlockCacheKey cacheKey) { |
| return doEvictBlock(cacheKey, null, false); |
| } |
| |
| /** |
| * Evict the {@link BlockCacheKey} and {@link BucketEntry} from {@link BucketCache#backingMap} and |
| * {@link BucketCache#ramCache}. <br/> |
| * NOTE:When Evict from {@link BucketCache#backingMap},only the matched {@link BlockCacheKey} and |
| * {@link BucketEntry} could be removed. |
| * @param cacheKey {@link BlockCacheKey} to evict. |
| * @param bucketEntry {@link BucketEntry} matched {@link BlockCacheKey} to evict. |
| * @return true to indicate whether we've evicted successfully or not. |
| */ |
| private boolean doEvictBlock(BlockCacheKey cacheKey, BucketEntry bucketEntry, |
| boolean evictedByEvictionProcess) { |
| if (!cacheEnabled) { |
| return false; |
| } |
| boolean existedInRamCache = removeFromRamCache(cacheKey); |
| if (bucketEntry == null) { |
| bucketEntry = backingMap.get(cacheKey); |
| } |
| final BucketEntry bucketEntryToUse = bucketEntry; |
| |
| if (bucketEntryToUse == null) { |
| if (existedInRamCache && evictedByEvictionProcess) { |
| cacheStats.evicted(0, cacheKey.isPrimary()); |
| } |
| return existedInRamCache; |
| } else { |
| return bucketEntryToUse.withWriteLock(offsetLock, () -> { |
| if (backingMap.remove(cacheKey, bucketEntryToUse)) { |
| LOG.debug("removed key {} from back map in the evict process", cacheKey); |
| blockEvicted(cacheKey, bucketEntryToUse, !existedInRamCache, evictedByEvictionProcess); |
| return true; |
| } |
| return false; |
| }); |
| } |
| } |
| |
| /** |
| * <pre> |
| * Create the {@link Recycler} for {@link BucketEntry#refCnt},which would be used as |
| * {@link RefCnt#recycler} of {@link HFileBlock#buf} returned from {@link BucketCache#getBlock}. |
| * NOTE: for {@link BucketCache#getBlock},the {@link RefCnt#recycler} of {@link HFileBlock#buf} |
| * from {@link BucketCache#backingMap} and {@link BucketCache#ramCache} are different: |
| * 1.For {@link RefCnt#recycler} of {@link HFileBlock#buf} from {@link BucketCache#backingMap}, |
| * it is the return value of current {@link BucketCache#createRecycler} method. |
| * |
| * 2.For {@link RefCnt#recycler} of {@link HFileBlock#buf} from {@link BucketCache#ramCache}, |
| * it is {@link ByteBuffAllocator#putbackBuffer}. |
| * </pre> |
| */ |
| private Recycler createRecycler(final BucketEntry bucketEntry) { |
| return () -> { |
| freeBucketEntry(bucketEntry); |
| return; |
| }; |
| } |
| |
| /** |
| * NOTE: This method is only for test. |
| */ |
| public boolean evictBlockIfNoRpcReferenced(BlockCacheKey blockCacheKey) { |
| BucketEntry bucketEntry = backingMap.get(blockCacheKey); |
| if (bucketEntry == null) { |
| return false; |
| } |
| return evictBucketEntryIfNoRpcReferenced(blockCacheKey, bucketEntry); |
| } |
| |
| /** |
| * Evict {@link BlockCacheKey} and its corresponding {@link BucketEntry} only if |
| * {@link BucketEntry#isRpcRef} is false. <br/> |
| * NOTE:When evict from {@link BucketCache#backingMap},only the matched {@link BlockCacheKey} and |
| * {@link BucketEntry} could be removed. |
| * @param blockCacheKey {@link BlockCacheKey} to evict. |
| * @param bucketEntry {@link BucketEntry} matched {@link BlockCacheKey} to evict. |
| * @return true to indicate whether we've evicted successfully or not. |
| */ |
| boolean evictBucketEntryIfNoRpcReferenced(BlockCacheKey blockCacheKey, BucketEntry bucketEntry) { |
| if (!bucketEntry.isRpcRef()) { |
| return doEvictBlock(blockCacheKey, bucketEntry, true); |
| } |
| return false; |
| } |
| |
| protected boolean removeFromRamCache(BlockCacheKey cacheKey) { |
| return ramCache.remove(cacheKey, re -> { |
| if (re != null) { |
| this.blockNumber.decrement(); |
| this.heapSize.add(-1 * re.getData().heapSize()); |
| } |
| }); |
| } |
| |
| public boolean isCacheInconsistent() { |
| return isCacheInconsistent.get(); |
| } |
| |
| public void setCacheInconsistent(boolean setCacheInconsistent) { |
| isCacheInconsistent.set(setCacheInconsistent); |
| } |
| |
| /* |
| * Statistics thread. Periodically output cache statistics to the log. |
| */ |
| private static class StatisticsThread extends Thread { |
| private final BucketCache bucketCache; |
| |
| public StatisticsThread(BucketCache bucketCache) { |
| super("BucketCacheStatsThread"); |
| setDaemon(true); |
| this.bucketCache = bucketCache; |
| } |
| |
| @Override |
| public void run() { |
| bucketCache.logStats(); |
| } |
| } |
| |
| public void logStats() { |
| long totalSize = bucketAllocator.getTotalSize(); |
| long usedSize = bucketAllocator.getUsedSize(); |
| long freeSize = totalSize - usedSize; |
| long cacheSize = getRealCacheSize(); |
| LOG.info("failedBlockAdditions=" + cacheStats.getFailedInserts() + ", " + "totalSize=" |
| + StringUtils.byteDesc(totalSize) + ", " + "freeSize=" + StringUtils.byteDesc(freeSize) + ", " |
| + "usedSize=" + StringUtils.byteDesc(usedSize) + ", " + "cacheSize=" |
| + StringUtils.byteDesc(cacheSize) + ", " + "accesses=" + cacheStats.getRequestCount() + ", " |
| + "hits=" + cacheStats.getHitCount() + ", " + "IOhitsPerSecond=" |
| + cacheStats.getIOHitsPerSecond() + ", " + "IOTimePerHit=" |
| + String.format("%.2f", cacheStats.getIOTimePerHit()) + ", " + "hitRatio=" |
| + (cacheStats.getHitCount() == 0 |
| ? "0," |
| : (StringUtils.formatPercent(cacheStats.getHitRatio(), 2) + ", ")) |
| + "cachingAccesses=" + cacheStats.getRequestCachingCount() + ", " + "cachingHits=" |
| + cacheStats.getHitCachingCount() + ", " + "cachingHitsRatio=" |
| + (cacheStats.getHitCachingCount() == 0 |
| ? "0," |
| : (StringUtils.formatPercent(cacheStats.getHitCachingRatio(), 2) + ", ")) |
| + "evictions=" + cacheStats.getEvictionCount() + ", " + "evicted=" |
| + cacheStats.getEvictedCount() + ", " + "evictedPerRun=" + cacheStats.evictedPerEviction() |
| + ", " + "allocationFailCount=" + cacheStats.getAllocationFailCount()); |
| cacheStats.reset(); |
| |
| bucketAllocator.logDebugStatistics(); |
| } |
| |
| public long getRealCacheSize() { |
| return this.realCacheSize.sum(); |
| } |
| |
| public long acceptableSize() { |
| return (long) Math.floor(bucketAllocator.getTotalSize() * acceptableFactor); |
| } |
| |
| long getPartitionSize(float partitionFactor) { |
| return (long) Math.floor(bucketAllocator.getTotalSize() * partitionFactor * minFactor); |
| } |
| |
| /** |
| * Return the count of bucketSizeinfos still need free space |
| */ |
| private int bucketSizesAboveThresholdCount(float minFactor) { |
| BucketAllocator.IndexStatistics[] stats = bucketAllocator.getIndexStatistics(); |
| int fullCount = 0; |
| for (int i = 0; i < stats.length; i++) { |
| long freeGoal = (long) Math.floor(stats[i].totalCount() * (1 - minFactor)); |
| freeGoal = Math.max(freeGoal, 1); |
| if (stats[i].freeCount() < freeGoal) { |
| fullCount++; |
| } |
| } |
| return fullCount; |
| } |
| |
| /** |
| * This method will find the buckets that are minimally occupied and are not reference counted and |
| * will free them completely without any constraint on the access times of the elements, and as a |
| * process will completely free at most the number of buckets passed, sometimes it might not due |
| * to changing refCounts |
| * @param completelyFreeBucketsNeeded number of buckets to free |
| **/ |
| private void freeEntireBuckets(int completelyFreeBucketsNeeded) { |
| if (completelyFreeBucketsNeeded != 0) { |
| // First we will build a set where the offsets are reference counted, usually |
| // this set is small around O(Handler Count) unless something else is wrong |
| Set<Integer> inUseBuckets = new HashSet<>(); |
| backingMap.forEach((k, be) -> { |
| if (be.isRpcRef()) { |
| inUseBuckets.add(bucketAllocator.getBucketIndex(be.offset())); |
| } |
| }); |
| Set<Integer> candidateBuckets = |
| bucketAllocator.getLeastFilledBuckets(inUseBuckets, completelyFreeBucketsNeeded); |
| for (Map.Entry<BlockCacheKey, BucketEntry> entry : backingMap.entrySet()) { |
| if (candidateBuckets.contains(bucketAllocator.getBucketIndex(entry.getValue().offset()))) { |
| evictBucketEntryIfNoRpcReferenced(entry.getKey(), entry.getValue()); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Free the space if the used size reaches acceptableSize() or one size block couldn't be |
| * allocated. When freeing the space, we use the LRU algorithm and ensure there must be some |
| * blocks evicted |
| * @param why Why we are being called |
| */ |
| void freeSpace(final String why) { |
| // Ensure only one freeSpace progress at a time |
| if (!freeSpaceLock.tryLock()) { |
| return; |
| } |
| try { |
| freeInProgress = true; |
| long bytesToFreeWithoutExtra = 0; |
| // Calculate free byte for each bucketSizeinfo |
| StringBuilder msgBuffer = LOG.isDebugEnabled() ? new StringBuilder() : null; |
| BucketAllocator.IndexStatistics[] stats = bucketAllocator.getIndexStatistics(); |
| long[] bytesToFreeForBucket = new long[stats.length]; |
| for (int i = 0; i < stats.length; i++) { |
| bytesToFreeForBucket[i] = 0; |
| long freeGoal = (long) Math.floor(stats[i].totalCount() * (1 - minFactor)); |
| freeGoal = Math.max(freeGoal, 1); |
| if (stats[i].freeCount() < freeGoal) { |
| bytesToFreeForBucket[i] = stats[i].itemSize() * (freeGoal - stats[i].freeCount()); |
| bytesToFreeWithoutExtra += bytesToFreeForBucket[i]; |
| if (msgBuffer != null) { |
| msgBuffer.append("Free for bucketSize(" + stats[i].itemSize() + ")=" |
| + StringUtils.byteDesc(bytesToFreeForBucket[i]) + ", "); |
| } |
| } |
| } |
| if (msgBuffer != null) { |
| msgBuffer.append("Free for total=" + StringUtils.byteDesc(bytesToFreeWithoutExtra) + ", "); |
| } |
| |
| if (bytesToFreeWithoutExtra <= 0) { |
| return; |
| } |
| long currentSize = bucketAllocator.getUsedSize(); |
| long totalSize = bucketAllocator.getTotalSize(); |
| if (LOG.isDebugEnabled() && msgBuffer != null) { |
| LOG.debug("Free started because \"" + why + "\"; " + msgBuffer.toString() |
| + " of current used=" + StringUtils.byteDesc(currentSize) + ", actual cacheSize=" |
| + StringUtils.byteDesc(realCacheSize.sum()) + ", total=" |
| + StringUtils.byteDesc(totalSize)); |
| } |
| |
| long bytesToFreeWithExtra = |
| (long) Math.floor(bytesToFreeWithoutExtra * (1 + extraFreeFactor)); |
| |
| // Instantiate priority buckets |
| BucketEntryGroup bucketSingle = |
| new BucketEntryGroup(bytesToFreeWithExtra, blockSize, getPartitionSize(singleFactor)); |
| BucketEntryGroup bucketMulti = |
| new BucketEntryGroup(bytesToFreeWithExtra, blockSize, getPartitionSize(multiFactor)); |
| BucketEntryGroup bucketMemory = |
| new BucketEntryGroup(bytesToFreeWithExtra, blockSize, getPartitionSize(memoryFactor)); |
| |
| // Scan entire map putting bucket entry into appropriate bucket entry |
| // group |
| for (Map.Entry<BlockCacheKey, BucketEntry> bucketEntryWithKey : backingMap.entrySet()) { |
| switch (bucketEntryWithKey.getValue().getPriority()) { |
| case SINGLE: { |
| bucketSingle.add(bucketEntryWithKey); |
| break; |
| } |
| case MULTI: { |
| bucketMulti.add(bucketEntryWithKey); |
| break; |
| } |
| case MEMORY: { |
| bucketMemory.add(bucketEntryWithKey); |
| break; |
| } |
| } |
| } |
| |
| PriorityQueue<BucketEntryGroup> bucketQueue = |
| new PriorityQueue<>(3, Comparator.comparingLong(BucketEntryGroup::overflow)); |
| |
| bucketQueue.add(bucketSingle); |
| bucketQueue.add(bucketMulti); |
| bucketQueue.add(bucketMemory); |
| |
| int remainingBuckets = bucketQueue.size(); |
| long bytesFreed = 0; |
| |
| BucketEntryGroup bucketGroup; |
| while ((bucketGroup = bucketQueue.poll()) != null) { |
| long overflow = bucketGroup.overflow(); |
| if (overflow > 0) { |
| long bucketBytesToFree = |
| Math.min(overflow, (bytesToFreeWithoutExtra - bytesFreed) / remainingBuckets); |
| bytesFreed += bucketGroup.free(bucketBytesToFree); |
| } |
| remainingBuckets--; |
| } |
| |
| // Check and free if there are buckets that still need freeing of space |
| if (bucketSizesAboveThresholdCount(minFactor) > 0) { |
| bucketQueue.clear(); |
| remainingBuckets = 3; |
| |
| bucketQueue.add(bucketSingle); |
| bucketQueue.add(bucketMulti); |
| bucketQueue.add(bucketMemory); |
| |
| while ((bucketGroup = bucketQueue.poll()) != null) { |
| long bucketBytesToFree = (bytesToFreeWithExtra - bytesFreed) / remainingBuckets; |
| bytesFreed += bucketGroup.free(bucketBytesToFree); |
| remainingBuckets--; |
| } |
| } |
| |
| // Even after the above free we might still need freeing because of the |
| // De-fragmentation of the buckets (also called Slab Calcification problem), i.e |
| // there might be some buckets where the occupancy is very sparse and thus are not |
| // yielding the free for the other bucket sizes, the fix for this to evict some |
| // of the buckets, we do this by evicting the buckets that are least fulled |
| freeEntireBuckets(DEFAULT_FREE_ENTIRE_BLOCK_FACTOR * bucketSizesAboveThresholdCount(1.0f)); |
| |
| if (LOG.isDebugEnabled()) { |
| long single = bucketSingle.totalSize(); |
| long multi = bucketMulti.totalSize(); |
| long memory = bucketMemory.totalSize(); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Bucket cache free space completed; " + "freed=" |
| + StringUtils.byteDesc(bytesFreed) + ", " + "total=" + StringUtils.byteDesc(totalSize) |
| + ", " + "single=" + StringUtils.byteDesc(single) + ", " + "multi=" |
| + StringUtils.byteDesc(multi) + ", " + "memory=" + StringUtils.byteDesc(memory)); |
| } |
| } |
| |
| } catch (Throwable t) { |
| LOG.warn("Failed freeing space", t); |
| } finally { |
| cacheStats.evict(); |
| freeInProgress = false; |
| freeSpaceLock.unlock(); |
| } |
| } |
| |
| // This handles flushing the RAM cache to IOEngine. |
| class WriterThread extends Thread { |
| private final BlockingQueue<RAMQueueEntry> inputQueue; |
| private volatile boolean writerEnabled = true; |
| private final ByteBuffer metaBuff = ByteBuffer.allocate(HFileBlock.BLOCK_METADATA_SPACE); |
| |
| WriterThread(BlockingQueue<RAMQueueEntry> queue) { |
| super("BucketCacheWriterThread"); |
| this.inputQueue = queue; |
| } |
| |
| // Used for test |
| void disableWriter() { |
| this.writerEnabled = false; |
| } |
| |
| @Override |
| public void run() { |
| List<RAMQueueEntry> entries = new ArrayList<>(); |
| try { |
| while (cacheEnabled && writerEnabled) { |
| try { |
| try { |
| // Blocks |
| entries = getRAMQueueEntries(inputQueue, entries); |
| } catch (InterruptedException ie) { |
| if (!cacheEnabled || !writerEnabled) { |
| break; |
| } |
| } |
| doDrain(entries, metaBuff); |
| } catch (Exception ioe) { |
| LOG.error("WriterThread encountered error", ioe); |
| } |
| } |
| } catch (Throwable t) { |
| LOG.warn("Failed doing drain", t); |
| } |
| LOG.info(this.getName() + " exiting, cacheEnabled=" + cacheEnabled); |
| } |
| } |
| |
| /** |
| * Put the new bucket entry into backingMap. Notice that we are allowed to replace the existing |
| * cache with a new block for the same cache key. there's a corner case: one thread cache a block |
| * in ramCache, copy to io-engine and add a bucket entry to backingMap. Caching another new block |
| * with the same cache key do the same thing for the same cache key, so if not evict the previous |
| * bucket entry, then memory leak happen because the previous bucketEntry is gone but the |
| * bucketAllocator do not free its memory. |
| * @see BlockCacheUtil#shouldReplaceExistingCacheBlock(BlockCache blockCache,BlockCacheKey |
| * cacheKey, Cacheable newBlock) |
| * @param key Block cache key |
| * @param bucketEntry Bucket entry to put into backingMap. |
| */ |
| protected void putIntoBackingMap(BlockCacheKey key, BucketEntry bucketEntry) { |
| BucketEntry previousEntry = backingMap.put(key, bucketEntry); |
| blocksByHFile.add(key); |
| updateRegionCachedSize(key.getFilePath(), bucketEntry.getLength()); |
| if (previousEntry != null && previousEntry != bucketEntry) { |
| previousEntry.withWriteLock(offsetLock, () -> { |
| blockEvicted(key, previousEntry, false, false); |
| return null; |
| }); |
| } |
| } |
| |
| /** |
| * Prepare and return a warning message for Bucket Allocator Exception |
| * @param fle The exception |
| * @param re The RAMQueueEntry for which the exception was thrown. |
| * @return A warning message created from the input RAMQueueEntry object. |
| */ |
| private static String getAllocationFailWarningMessage(final BucketAllocatorException fle, |
| final RAMQueueEntry re) { |
| final StringBuilder sb = new StringBuilder(); |
| sb.append("Most recent failed allocation after "); |
| sb.append(ALLOCATION_FAIL_LOG_TIME_PERIOD); |
| sb.append(" ms;"); |
| if (re != null) { |
| if (re.getData() instanceof HFileBlock) { |
| final HFileContext fileContext = ((HFileBlock) re.getData()).getHFileContext(); |
| final String columnFamily = Bytes.toString(fileContext.getColumnFamily()); |
| final String tableName = Bytes.toString(fileContext.getTableName()); |
| if (tableName != null) { |
| sb.append(" Table: "); |
| sb.append(tableName); |
| } |
| if (columnFamily != null) { |
| sb.append(" CF: "); |
| sb.append(columnFamily); |
| } |
| sb.append(" HFile: "); |
| if (fileContext.getHFileName() != null) { |
| sb.append(fileContext.getHFileName()); |
| } else { |
| sb.append(re.getKey()); |
| } |
| } else { |
| sb.append(" HFile: "); |
| sb.append(re.getKey()); |
| } |
| } |
| sb.append(" Message: "); |
| sb.append(fle.getMessage()); |
| return sb.toString(); |
| } |
| |
| /** |
| * Flush the entries in ramCache to IOEngine and add bucket entry to backingMap. Process all that |
| * are passed in even if failure being sure to remove from ramCache else we'll never undo the |
| * references and we'll OOME. |
| * @param entries Presumes list passed in here will be processed by this invocation only. No |
| * interference expected. |
| */ |
| void doDrain(final List<RAMQueueEntry> entries, ByteBuffer metaBuff) throws InterruptedException { |
| if (entries.isEmpty()) { |
| return; |
| } |
| // This method is a little hard to follow. We run through the passed in entries and for each |
| // successful add, we add a non-null BucketEntry to the below bucketEntries. Later we must |
| // do cleanup making sure we've cleared ramCache of all entries regardless of whether we |
| // successfully added the item to the bucketcache; if we don't do the cleanup, we'll OOME by |
| // filling ramCache. We do the clean up by again running through the passed in entries |
| // doing extra work when we find a non-null bucketEntries corresponding entry. |
| final int size = entries.size(); |
| BucketEntry[] bucketEntries = new BucketEntry[size]; |
| // Index updated inside loop if success or if we can't succeed. We retry if cache is full |
| // when we go to add an entry by going around the loop again without upping the index. |
| int index = 0; |
| while (cacheEnabled && index < size) { |
| RAMQueueEntry re = null; |
| try { |
| re = entries.get(index); |
| if (re == null) { |
| LOG.warn("Couldn't get entry or changed on us; who else is messing with it?"); |
| index++; |
| continue; |
| } |
| // Reset the position for reuse. |
| // It should be guaranteed that the data in the metaBuff has been transferred to the |
| // ioEngine safely. Otherwise, this reuse is problematic. Fortunately, the data is already |
| // transferred with our current IOEngines. Should take care, when we have new kinds of |
| // IOEngine in the future. |
| metaBuff.clear(); |
| BucketEntry bucketEntry = |
| re.writeToCache(ioEngine, bucketAllocator, realCacheSize, this::createRecycler, metaBuff); |
| // Successfully added. Up index and add bucketEntry. Clear io exceptions. |
| bucketEntries[index] = bucketEntry; |
| if (ioErrorStartTime > 0) { |
| ioErrorStartTime = -1; |
| } |
| index++; |
| } catch (BucketAllocatorException fle) { |
| long currTs = EnvironmentEdgeManager.currentTime(); |
| cacheStats.allocationFailed(); // Record the warning. |
| if ( |
| allocFailLogPrevTs == 0 || (currTs - allocFailLogPrevTs) > ALLOCATION_FAIL_LOG_TIME_PERIOD |
| ) { |
| LOG.warn(getAllocationFailWarningMessage(fle, re)); |
| allocFailLogPrevTs = currTs; |
| } |
| // Presume can't add. Too big? Move index on. Entry will be cleared from ramCache below. |
| bucketEntries[index] = null; |
| index++; |
| } catch (CacheFullException cfe) { |
| // Cache full when we tried to add. Try freeing space and then retrying (don't up index) |
| if (!freeInProgress) { |
| freeSpace("Full!"); |
| } else { |
| Thread.sleep(50); |
| } |
| } catch (IOException ioex) { |
| // Hopefully transient. Retry. checkIOErrorIsTolerated disables cache if problem. |
| LOG.error("Failed writing to bucket cache", ioex); |
| checkIOErrorIsTolerated(); |
| } |
| } |
| |
| // Make sure data pages are written on media before we update maps. |
| try { |
| ioEngine.sync(); |
| } catch (IOException ioex) { |
| LOG.error("Failed syncing IO engine", ioex); |
| checkIOErrorIsTolerated(); |
| // Since we failed sync, free the blocks in bucket allocator |
| for (int i = 0; i < entries.size(); ++i) { |
| BucketEntry bucketEntry = bucketEntries[i]; |
| if (bucketEntry != null) { |
| bucketAllocator.freeBlock(bucketEntry.offset(), bucketEntry.getLength()); |
| bucketEntries[i] = null; |
| } |
| } |
| } |
| |
| // Now add to backingMap if successfully added to bucket cache. Remove from ramCache if |
| // success or error. |
| for (int i = 0; i < size; ++i) { |
| BlockCacheKey key = entries.get(i).getKey(); |
| // Only add if non-null entry. |
| if (bucketEntries[i] != null) { |
| putIntoBackingMap(key, bucketEntries[i]); |
| if (ioEngine.isPersistent()) { |
| setCacheInconsistent(true); |
| } |
| } |
| // Always remove from ramCache even if we failed adding it to the block cache above. |
| boolean existed = ramCache.remove(key, re -> { |
| if (re != null) { |
| heapSize.add(-1 * re.getData().heapSize()); |
| } |
| }); |
| if (!existed && bucketEntries[i] != null) { |
| // Block should have already been evicted. Remove it and free space. |
| final BucketEntry bucketEntry = bucketEntries[i]; |
| bucketEntry.withWriteLock(offsetLock, () -> { |
| if (backingMap.remove(key, bucketEntry)) { |
| blockEvicted(key, bucketEntry, false, false); |
| } |
| return null; |
| }); |
| } |
| } |
| |
| long used = bucketAllocator.getUsedSize(); |
| if (used > acceptableSize()) { |
| freeSpace("Used=" + used + " > acceptable=" + acceptableSize()); |
| } |
| return; |
| } |
| |
| /** |
| * Blocks until elements available in {@code q} then tries to grab as many as possible before |
| * returning. |
| * @param receptacle Where to stash the elements taken from queue. We clear before we use it just |
| * in case. |
| * @param q The queue to take from. |
| * @return {@code receptacle} laden with elements taken from the queue or empty if none found. |
| */ |
| static List<RAMQueueEntry> getRAMQueueEntries(BlockingQueue<RAMQueueEntry> q, |
| List<RAMQueueEntry> receptacle) throws InterruptedException { |
| // Clear sets all entries to null and sets size to 0. We retain allocations. Presume it |
| // ok even if list grew to accommodate thousands. |
| receptacle.clear(); |
| receptacle.add(q.take()); |
| q.drainTo(receptacle); |
| return receptacle; |
| } |
| |
| /** |
| * @see #retrieveFromFile(int[]) |
| */ |
| @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "OBL_UNSATISFIED_OBLIGATION", |
| justification = "false positive, try-with-resources ensures close is called.") |
| void persistToFile() throws IOException { |
| LOG.debug("Thread {} started persisting bucket cache to file", |
| Thread.currentThread().getName()); |
| if (!isCachePersistent()) { |
| throw new IOException("Attempt to persist non-persistent cache mappings!"); |
| } |
| File tempPersistencePath = new File(persistencePath + EnvironmentEdgeManager.currentTime()); |
| try (FileOutputStream fos = new FileOutputStream(tempPersistencePath, false)) { |
| fos.write(ProtobufMagic.PB_MAGIC); |
| BucketProtoUtils.toPB(this).writeDelimitedTo(fos); |
| } catch (IOException e) { |
| LOG.error("Failed to persist bucket cache to file", e); |
| throw e; |
| } |
| LOG.debug("Thread {} finished persisting bucket cache to file, renaming", |
| Thread.currentThread().getName()); |
| if (!tempPersistencePath.renameTo(new File(persistencePath))) { |
| LOG.warn("Failed to commit cache persistent file. We might lose cached blocks if " |
| + "RS crashes/restarts before we successfully checkpoint again."); |
| } |
| } |
| |
| public boolean isCachePersistent() { |
| return ioEngine.isPersistent() && persistencePath != null; |
| } |
| |
| @Override |
| public Optional<Map<String, Long>> getRegionCachedInfo() { |
| return Optional.of(Collections.unmodifiableMap(regionCachedSize)); |
| } |
| |
| /** |
| * @see #persistToFile() |
| */ |
| private void retrieveFromFile(int[] bucketSizes) throws IOException { |
| LOG.info("Started retrieving bucket cache from file"); |
| File persistenceFile = new File(persistencePath); |
| if (!persistenceFile.exists()) { |
| LOG.warn("Persistence file missing! " |
| + "It's ok if it's first run after enabling persistent cache."); |
| bucketAllocator = new BucketAllocator(cacheCapacity, bucketSizes, backingMap, realCacheSize); |
| blockNumber.add(backingMap.size()); |
| backingMapValidated.set(true); |
| return; |
| } |
| assert !cacheEnabled; |
| |
| try (FileInputStream in = new FileInputStream(persistenceFile)) { |
| int pblen = ProtobufMagic.lengthOfPBMagic(); |
| byte[] pbuf = new byte[pblen]; |
| int read = in.read(pbuf); |
| if (read != pblen) { |
| throw new IOException("Incorrect number of bytes read while checking for protobuf magic " |
| + "number. Requested=" + pblen + ", Received= " + read + ", File=" + persistencePath); |
| } |
| if (!ProtobufMagic.isPBMagicPrefix(pbuf)) { |
| // In 3.0 we have enough flexibility to dump the old cache data. |
| // TODO: In 2.x line, this might need to be filled in to support reading the old format |
| throw new IOException( |
| "Persistence file does not start with protobuf magic number. " + persistencePath); |
| } |
| parsePB(BucketCacheProtos.BucketCacheEntry.parseDelimitedFrom(in)); |
| bucketAllocator = new BucketAllocator(cacheCapacity, bucketSizes, backingMap, realCacheSize); |
| blockNumber.add(backingMap.size()); |
| LOG.info("Bucket cache retrieved from file successfully"); |
| } |
| } |
| |
| private void updateRegionSizeMapWhileRetrievingFromFile() { |
| // Update the regionCachedSize with the region size while restarting the region server |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Updating region size map after retrieving cached file list"); |
| dumpPrefetchList(); |
| } |
| regionCachedSize.clear(); |
| fullyCachedFiles.forEach((hFileName, hFileSize) -> { |
| // Get the region name for each file |
| String regionEncodedName = hFileSize.getFirst(); |
| long cachedFileSize = hFileSize.getSecond(); |
| regionCachedSize.merge(regionEncodedName, cachedFileSize, |
| (oldpf, fileSize) -> oldpf + fileSize); |
| }); |
| } |
| |
| private void dumpPrefetchList() { |
| for (Map.Entry<String, Pair<String, Long>> outerEntry : fullyCachedFiles.entrySet()) { |
| LOG.debug("Cached File Entry:<{},<{},{}>>", outerEntry.getKey(), |
| outerEntry.getValue().getFirst(), outerEntry.getValue().getSecond()); |
| } |
| } |
| |
| /** |
| * Create an input stream that deletes the file after reading it. Use in try-with-resources to |
| * avoid this pattern where an exception thrown from a finally block may mask earlier exceptions: |
| * |
| * <pre> |
| * File f = ... |
| * try (FileInputStream fis = new FileInputStream(f)) { |
| * // use the input stream |
| * } finally { |
| * if (!f.delete()) throw new IOException("failed to delete"); |
| * } |
| * </pre> |
| * |
| * @param file the file to read and delete |
| * @return a FileInputStream for the given file |
| * @throws IOException if there is a problem creating the stream |
| */ |
| private FileInputStream deleteFileOnClose(final File file) throws IOException { |
| return new FileInputStream(file) { |
| private File myFile; |
| |
| private FileInputStream init(File file) { |
| myFile = file; |
| return this; |
| } |
| |
| @Override |
| public void close() throws IOException { |
| // close() will be called during try-with-resources and it will be |
| // called by finalizer thread during GC. To avoid double-free resource, |
| // set myFile to null after the first call. |
| if (myFile == null) { |
| return; |
| } |
| |
| super.close(); |
| if (!myFile.delete()) { |
| throw new IOException("Failed deleting persistence file " + myFile.getAbsolutePath()); |
| } |
| myFile = null; |
| } |
| }.init(file); |
| } |
| |
| private void verifyCapacityAndClasses(long capacitySize, String ioclass, String mapclass) |
| throws IOException { |
| if (capacitySize != cacheCapacity) { |
| throw new IOException("Mismatched cache capacity:" + StringUtils.byteDesc(capacitySize) |
| + ", expected: " + StringUtils.byteDesc(cacheCapacity)); |
| } |
| if (!ioEngine.getClass().getName().equals(ioclass)) { |
| throw new IOException("Class name for IO engine mismatch: " + ioclass + ", expected:" |
| + ioEngine.getClass().getName()); |
| } |
| if (!backingMap.getClass().getName().equals(mapclass)) { |
| throw new IOException("Class name for cache map mismatch: " + mapclass + ", expected:" |
| + backingMap.getClass().getName()); |
| } |
| } |
| |
| private void parsePB(BucketCacheProtos.BucketCacheEntry proto) throws IOException { |
| Pair<ConcurrentHashMap<BlockCacheKey, BucketEntry>, NavigableSet<BlockCacheKey>> pair = |
| BucketProtoUtils.fromPB(proto.getDeserializersMap(), proto.getBackingMap(), |
| this::createRecycler); |
| backingMap = pair.getFirst(); |
| blocksByHFile = pair.getSecond(); |
| fullyCachedFiles.clear(); |
| fullyCachedFiles.putAll(BucketProtoUtils.fromPB(proto.getCachedFilesMap())); |
| if (proto.hasChecksum()) { |
| try { |
| ((PersistentIOEngine) ioEngine).verifyFileIntegrity(proto.getChecksum().toByteArray(), |
| algorithm); |
| backingMapValidated.set(true); |
| } catch (IOException e) { |
| LOG.warn("Checksum for cache file failed. " |
| + "We need to validate each cache key in the backing map. " |
| + "This may take some time, so we'll do it in a background thread,"); |
| Runnable cacheValidator = () -> { |
| while (bucketAllocator == null) { |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException ex) { |
| throw new RuntimeException(ex); |
| } |
| } |
| long startTime = EnvironmentEdgeManager.currentTime(); |
| int totalKeysOriginally = backingMap.size(); |
| for (Map.Entry<BlockCacheKey, BucketEntry> keyEntry : backingMap.entrySet()) { |
| try { |
| ((FileIOEngine) ioEngine).checkCacheTime(keyEntry.getValue()); |
| } catch (IOException e1) { |
| LOG.debug("Check for key {} failed. Evicting.", keyEntry.getKey()); |
| evictBlock(keyEntry.getKey()); |
| fullyCachedFiles.remove(keyEntry.getKey().getHfileName()); |
| } |
| } |
| backingMapValidated.set(true); |
| LOG.info("Finished validating {} keys in the backing map. Recovered: {}. This took {}ms.", |
| totalKeysOriginally, backingMap.size(), |
| (EnvironmentEdgeManager.currentTime() - startTime)); |
| }; |
| Thread t = new Thread(cacheValidator); |
| t.setDaemon(true); |
| t.start(); |
| } |
| } else { |
| // if has not checksum, it means the persistence file is old format |
| LOG.info("Persistent file is old format, it does not support verifying file integrity!"); |
| backingMapValidated.set(true); |
| } |
| updateRegionSizeMapWhileRetrievingFromFile(); |
| verifyCapacityAndClasses(proto.getCacheCapacity(), proto.getIoClass(), proto.getMapClass()); |
| } |
| |
| /** |
| * Check whether we tolerate IO error this time. If the duration of IOEngine throwing errors |
| * exceeds ioErrorsDurationTimeTolerated, we will disable the cache |
| */ |
| private void checkIOErrorIsTolerated() { |
| long now = EnvironmentEdgeManager.currentTime(); |
| // Do a single read to a local variable to avoid timing issue - HBASE-24454 |
| long ioErrorStartTimeTmp = this.ioErrorStartTime; |
| if (ioErrorStartTimeTmp > 0) { |
| if (cacheEnabled && (now - ioErrorStartTimeTmp) > this.ioErrorsTolerationDuration) { |
| LOG.error("IO errors duration time has exceeded " + ioErrorsTolerationDuration |
| + "ms, disabling cache, please check your IOEngine"); |
| disableCache(); |
| } |
| } else { |
| this.ioErrorStartTime = now; |
| } |
| } |
| |
| /** |
| * Used to shut down the cache -or- turn it off in the case of something broken. |
| */ |
| private void disableCache() { |
| if (!cacheEnabled) return; |
| LOG.info("Disabling cache"); |
| cacheEnabled = false; |
| ioEngine.shutdown(); |
| this.scheduleThreadPool.shutdown(); |
| for (int i = 0; i < writerThreads.length; ++i) |
| writerThreads[i].interrupt(); |
| this.ramCache.clear(); |
| if (!ioEngine.isPersistent() || persistencePath == null) { |
| // If persistent ioengine and a path, we will serialize out the backingMap. |
| this.backingMap.clear(); |
| this.blocksByHFile.clear(); |
| this.fullyCachedFiles.clear(); |
| this.regionCachedSize.clear(); |
| } |
| } |
| |
| private void join() throws InterruptedException { |
| for (int i = 0; i < writerThreads.length; ++i) |
| writerThreads[i].join(); |
| } |
| |
| @Override |
| public void shutdown() { |
| disableCache(); |
| LOG.info("Shutdown bucket cache: IO persistent=" + ioEngine.isPersistent() + "; path to write=" |
| + persistencePath); |
| if (ioEngine.isPersistent() && persistencePath != null) { |
| try { |
| join(); |
| if (cachePersister != null) { |
| LOG.info("Shutting down cache persister thread."); |
| cachePersister.shutdown(); |
| while (cachePersister.isAlive()) { |
| Thread.sleep(10); |
| } |
| } |
| persistToFile(); |
| } catch (IOException ex) { |
| LOG.error("Unable to persist data on exit: " + ex.toString(), ex); |
| } catch (InterruptedException e) { |
| LOG.warn("Failed to persist data on exit", e); |
| } |
| } |
| } |
| |
| /** |
| * Needed mostly for UTs that might run in the same VM and create different BucketCache instances |
| * on different UT methods. |
| */ |
| @Override |
| protected void finalize() { |
| if (cachePersister != null && !cachePersister.isInterrupted()) { |
| cachePersister.interrupt(); |
| } |
| } |
| |
| @Override |
| public CacheStats getStats() { |
| return cacheStats; |
| } |
| |
| public BucketAllocator getAllocator() { |
| return this.bucketAllocator; |
| } |
| |
| @Override |
| public long heapSize() { |
| return this.heapSize.sum(); |
| } |
| |
| @Override |
| public long size() { |
| return this.realCacheSize.sum(); |
| } |
| |
| @Override |
| public long getCurrentDataSize() { |
| return size(); |
| } |
| |
| @Override |
| public long getFreeSize() { |
| return this.bucketAllocator.getFreeSize(); |
| } |
| |
| @Override |
| public long getBlockCount() { |
| return this.blockNumber.sum(); |
| } |
| |
| @Override |
| public long getDataBlockCount() { |
| return getBlockCount(); |
| } |
| |
| @Override |
| public long getCurrentSize() { |
| return this.bucketAllocator.getUsedSize(); |
| } |
| |
| protected String getAlgorithm() { |
| return algorithm; |
| } |
| |
| /** |
| * Evicts all blocks for a specific HFile. |
| * <p> |
| * This is used for evict-on-close to remove all blocks of a specific HFile. |
| * @return the number of blocks evicted |
| */ |
| @Override |
| public int evictBlocksByHfileName(String hfileName) { |
| fileNotFullyCached(hfileName); |
| Set<BlockCacheKey> keySet = blocksByHFile.subSet(new BlockCacheKey(hfileName, Long.MIN_VALUE), |
| true, new BlockCacheKey(hfileName, Long.MAX_VALUE), true); |
| |
| int numEvicted = 0; |
| for (BlockCacheKey key : keySet) { |
| if (evictBlock(key)) { |
| ++numEvicted; |
| } |
| } |
| |
| return numEvicted; |
| } |
| |
| /** |
| * Used to group bucket entries into priority buckets. There will be a BucketEntryGroup for each |
| * priority (single, multi, memory). Once bucketed, the eviction algorithm takes the appropriate |
| * number of elements out of each according to configuration parameters and their relative sizes. |
| */ |
| private class BucketEntryGroup { |
| |
| private CachedEntryQueue queue; |
| private long totalSize = 0; |
| private long bucketSize; |
| |
| public BucketEntryGroup(long bytesToFree, long blockSize, long bucketSize) { |
| this.bucketSize = bucketSize; |
| queue = new CachedEntryQueue(bytesToFree, blockSize); |
| totalSize = 0; |
| } |
| |
| public void add(Map.Entry<BlockCacheKey, BucketEntry> block) { |
| totalSize += block.getValue().getLength(); |
| queue.add(block); |
| } |
| |
| public long free(long toFree) { |
| Map.Entry<BlockCacheKey, BucketEntry> entry; |
| long freedBytes = 0; |
| // TODO avoid a cycling siutation. We find no block which is not in use and so no way to free |
| // What to do then? Caching attempt fail? Need some changes in cacheBlock API? |
| while ((entry = queue.pollLast()) != null) { |
| BlockCacheKey blockCacheKey = entry.getKey(); |
| BucketEntry be = entry.getValue(); |
| if (evictBucketEntryIfNoRpcReferenced(blockCacheKey, be)) { |
| freedBytes += be.getLength(); |
| } |
| if (freedBytes >= toFree) { |
| return freedBytes; |
| } |
| } |
| return freedBytes; |
| } |
| |
| public long overflow() { |
| return totalSize - bucketSize; |
| } |
| |
| public long totalSize() { |
| return totalSize; |
| } |
| } |
| |
| /** |
| * Block Entry stored in the memory with key,data and so on |
| */ |
| static class RAMQueueEntry { |
| private final BlockCacheKey key; |
| private final Cacheable data; |
| private long accessCounter; |
| private boolean inMemory; |
| private boolean isCachePersistent; |
| |
| RAMQueueEntry(BlockCacheKey bck, Cacheable data, long accessCounter, boolean inMemory, |
| boolean isCachePersistent) { |
| this.key = bck; |
| this.data = data; |
| this.accessCounter = accessCounter; |
| this.inMemory = inMemory; |
| this.isCachePersistent = isCachePersistent; |
| } |
| |
| public Cacheable getData() { |
| return data; |
| } |
| |
| public BlockCacheKey getKey() { |
| return key; |
| } |
| |
| public void access(long accessCounter) { |
| this.accessCounter = accessCounter; |
| } |
| |
| private ByteBuffAllocator getByteBuffAllocator() { |
| if (data instanceof HFileBlock) { |
| return ((HFileBlock) data).getByteBuffAllocator(); |
| } |
| return ByteBuffAllocator.HEAP; |
| } |
| |
| public BucketEntry writeToCache(final IOEngine ioEngine, final BucketAllocator alloc, |
| final LongAdder realCacheSize, Function<BucketEntry, Recycler> createRecycler, |
| ByteBuffer metaBuff) throws IOException { |
| int len = data.getSerializedLength(); |
| // This cacheable thing can't be serialized |
| if (len == 0) { |
| return null; |
| } |
| if (isCachePersistent && data instanceof HFileBlock) { |
| len += Long.BYTES; // we need to record the cache time for consistency check in case of |
| // recovery |
| } |
| long offset = alloc.allocateBlock(len); |
| boolean succ = false; |
| BucketEntry bucketEntry = null; |
| try { |
| int diskSizeWithHeader = (data instanceof HFileBlock) |
| ? ((HFileBlock) data).getOnDiskSizeWithHeader() |
| : data.getSerializedLength(); |
| bucketEntry = new BucketEntry(offset, len, diskSizeWithHeader, accessCounter, inMemory, |
| createRecycler, getByteBuffAllocator()); |
| bucketEntry.setDeserializerReference(data.getDeserializer()); |
| if (data instanceof HFileBlock) { |
| // If an instance of HFileBlock, save on some allocations. |
| HFileBlock block = (HFileBlock) data; |
| ByteBuff sliceBuf = block.getBufferReadOnly(); |
| block.getMetaData(metaBuff); |
| // adds the cache time prior to the block and metadata part |
| if (isCachePersistent) { |
| ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES); |
| buffer.putLong(bucketEntry.getCachedTime()); |
| buffer.rewind(); |
| ioEngine.write(buffer, offset); |
| ioEngine.write(sliceBuf, (offset + Long.BYTES)); |
| } else { |
| ioEngine.write(sliceBuf, offset); |
| } |
| ioEngine.write(metaBuff, offset + len - metaBuff.limit()); |
| } else { |
| // Only used for testing. |
| ByteBuffer bb = ByteBuffer.allocate(len); |
| data.serialize(bb, true); |
| ioEngine.write(bb, offset); |
| } |
| succ = true; |
| } finally { |
| if (!succ) { |
| alloc.freeBlock(offset, len); |
| } |
| } |
| realCacheSize.add(len); |
| return bucketEntry; |
| } |
| } |
| |
| /** |
| * Only used in test |
| */ |
| void stopWriterThreads() throws InterruptedException { |
| for (WriterThread writerThread : writerThreads) { |
| writerThread.disableWriter(); |
| writerThread.interrupt(); |
| writerThread.join(); |
| } |
| } |
| |
| @Override |
| public Iterator<CachedBlock> iterator() { |
| // Don't bother with ramcache since stuff is in here only a little while. |
| final Iterator<Map.Entry<BlockCacheKey, BucketEntry>> i = this.backingMap.entrySet().iterator(); |
| return new Iterator<CachedBlock>() { |
| private final long now = System.nanoTime(); |
| |
| @Override |
| public boolean hasNext() { |
| return i.hasNext(); |
| } |
| |
| @Override |
| public CachedBlock next() { |
| final Map.Entry<BlockCacheKey, BucketEntry> e = i.next(); |
| return new CachedBlock() { |
| @Override |
| public String toString() { |
| return BlockCacheUtil.toString(this, now); |
| } |
| |
| @Override |
| public BlockPriority getBlockPriority() { |
| return e.getValue().getPriority(); |
| } |
| |
| @Override |
| public BlockType getBlockType() { |
| // Not held by BucketEntry. Could add it if wanted on BucketEntry creation. |
| return null; |
| } |
| |
| @Override |
| public long getOffset() { |
| return e.getKey().getOffset(); |
| } |
| |
| @Override |
| public long getSize() { |
| return e.getValue().getLength(); |
| } |
| |
| @Override |
| public long getCachedTime() { |
| return e.getValue().getCachedTime(); |
| } |
| |
| @Override |
| public String getFilename() { |
| return e.getKey().getHfileName(); |
| } |
| |
| @Override |
| public int compareTo(CachedBlock other) { |
| int diff = this.getFilename().compareTo(other.getFilename()); |
| if (diff != 0) return diff; |
| |
| diff = Long.compare(this.getOffset(), other.getOffset()); |
| if (diff != 0) return diff; |
| if (other.getCachedTime() < 0 || this.getCachedTime() < 0) { |
| throw new IllegalStateException( |
| "" + this.getCachedTime() + ", " + other.getCachedTime()); |
| } |
| return Long.compare(other.getCachedTime(), this.getCachedTime()); |
| } |
| |
| @Override |
| public int hashCode() { |
| return e.getKey().hashCode(); |
| } |
| |
| @Override |
| public boolean equals(Object obj) { |
| if (obj instanceof CachedBlock) { |
| CachedBlock cb = (CachedBlock) obj; |
| return compareTo(cb) == 0; |
| } else { |
| return false; |
| } |
| } |
| }; |
| } |
| |
| @Override |
| public void remove() { |
| throw new UnsupportedOperationException(); |
| } |
| }; |
| } |
| |
| @Override |
| public BlockCache[] getBlockCaches() { |
| return null; |
| } |
| |
| public int getRpcRefCount(BlockCacheKey cacheKey) { |
| BucketEntry bucketEntry = backingMap.get(cacheKey); |
| if (bucketEntry != null) { |
| return bucketEntry.refCnt() - (bucketEntry.markedAsEvicted.get() ? 0 : 1); |
| } |
| return 0; |
| } |
| |
| float getAcceptableFactor() { |
| return acceptableFactor; |
| } |
| |
| float getMinFactor() { |
| return minFactor; |
| } |
| |
| float getExtraFreeFactor() { |
| return extraFreeFactor; |
| } |
| |
| float getSingleFactor() { |
| return singleFactor; |
| } |
| |
| float getMultiFactor() { |
| return multiFactor; |
| } |
| |
| float getMemoryFactor() { |
| return memoryFactor; |
| } |
| |
| public String getPersistencePath() { |
| return persistencePath; |
| } |
| |
| /** |
| * Wrapped the delegate ConcurrentMap with maintaining its block's reference count. |
| */ |
| static class RAMCache { |
| /** |
| * Defined the map as {@link ConcurrentHashMap} explicitly here, because in |
| * {@link RAMCache#get(BlockCacheKey)} and |
| * {@link RAMCache#putIfAbsent(BlockCacheKey, BucketCache.RAMQueueEntry)} , we need to guarantee |
| * the atomicity of map#computeIfPresent(key, func) and map#putIfAbsent(key, func). Besides, the |
| * func method can execute exactly once only when the key is present(or absent) and under the |
| * lock context. Otherwise, the reference count of block will be messed up. Notice that the |
| * {@link java.util.concurrent.ConcurrentSkipListMap} can not guarantee that. |
| */ |
| final ConcurrentHashMap<BlockCacheKey, RAMQueueEntry> delegate = new ConcurrentHashMap<>(); |
| |
| public boolean containsKey(BlockCacheKey key) { |
| return delegate.containsKey(key); |
| } |
| |
| public RAMQueueEntry get(BlockCacheKey key) { |
| return delegate.computeIfPresent(key, (k, re) -> { |
| // It'll be referenced by RPC, so retain atomically here. if the get and retain is not |
| // atomic, another thread may remove and release the block, when retaining in this thread we |
| // may retain a block with refCnt=0 which is disallowed. (see HBASE-22422) |
| re.getData().retain(); |
| return re; |
| }); |
| } |
| |
| /** |
| * Return the previous associated value, or null if absent. It has the same meaning as |
| * {@link ConcurrentMap#putIfAbsent(Object, Object)} |
| */ |
| public RAMQueueEntry putIfAbsent(BlockCacheKey key, RAMQueueEntry entry) { |
| AtomicBoolean absent = new AtomicBoolean(false); |
| RAMQueueEntry re = delegate.computeIfAbsent(key, k -> { |
| // The RAMCache reference to this entry, so reference count should be increment. |
| entry.getData().retain(); |
| absent.set(true); |
| return entry; |
| }); |
| return absent.get() ? null : re; |
| } |
| |
| public boolean remove(BlockCacheKey key) { |
| return remove(key, re -> { |
| }); |
| } |
| |
| /** |
| * Defined an {@link Consumer} here, because once the removed entry release its reference count, |
| * then it's ByteBuffers may be recycled and accessing it outside this method will be thrown an |
| * exception. the consumer will access entry to remove before release its reference count. |
| * Notice, don't change its reference count in the {@link Consumer} |
| */ |
| public boolean remove(BlockCacheKey key, Consumer<RAMQueueEntry> action) { |
| RAMQueueEntry previous = delegate.remove(key); |
| action.accept(previous); |
| if (previous != null) { |
| previous.getData().release(); |
| } |
| return previous != null; |
| } |
| |
| public boolean isEmpty() { |
| return delegate.isEmpty(); |
| } |
| |
| public void clear() { |
| Iterator<Map.Entry<BlockCacheKey, RAMQueueEntry>> it = delegate.entrySet().iterator(); |
| while (it.hasNext()) { |
| RAMQueueEntry re = it.next().getValue(); |
| it.remove(); |
| re.getData().release(); |
| } |
| } |
| |
| public boolean hasBlocksForFile(String fileName) { |
| return delegate.keySet().stream().filter(key -> key.getHfileName().equals(fileName)) |
| .findFirst().isPresent(); |
| } |
| } |
| |
| public Map<BlockCacheKey, BucketEntry> getBackingMap() { |
| return backingMap; |
| } |
| |
| public AtomicBoolean getBackingMapValidated() { |
| return backingMapValidated; |
| } |
| |
| @Override |
| public Optional<Map<String, Pair<String, Long>>> getFullyCachedFiles() { |
| return Optional.of(fullyCachedFiles); |
| } |
| |
| public static Optional<BucketCache> getBucketCacheFromCacheConfig(CacheConfig cacheConf) { |
| if (cacheConf.getBlockCache().isPresent()) { |
| BlockCache bc = cacheConf.getBlockCache().get(); |
| if (bc instanceof CombinedBlockCache) { |
| BlockCache l2 = ((CombinedBlockCache) bc).getSecondLevelCache(); |
| if (l2 instanceof BucketCache) { |
| return Optional.of((BucketCache) l2); |
| } |
| } else if (bc instanceof BucketCache) { |
| return Optional.of((BucketCache) bc); |
| } |
| } |
| return Optional.empty(); |
| } |
| |
| @Override |
| public void notifyFileCachingCompleted(Path fileName, int totalBlockCount, int dataBlockCount, |
| long size) { |
| // block eviction may be happening in the background as prefetch runs, |
| // so we need to count all blocks for this file in the backing map under |
| // a read lock for the block offset |
| final List<ReentrantReadWriteLock> locks = new ArrayList<>(); |
| LOG.debug("Notifying caching completed for file {}, with total blocks {}", fileName, |
| dataBlockCount); |
| try { |
| final MutableInt count = new MutableInt(); |
| LOG.debug("iterating over {} entries in the backing map", backingMap.size()); |
| backingMap.entrySet().stream().forEach(entry -> { |
| if (entry.getKey().getHfileName().equals(fileName.getName())) { |
| LOG.debug("found block for file {} in the backing map. Acquiring read lock for offset {}", |
| fileName, entry.getKey().getOffset()); |
| ReentrantReadWriteLock lock = offsetLock.getLock(entry.getKey().getOffset()); |
| lock.readLock().lock(); |
| locks.add(lock); |
| if (backingMap.containsKey(entry.getKey())) { |
| count.increment(); |
| } |
| } |
| }); |
| // We may either place only data blocks on the BucketCache or all type of blocks |
| if (dataBlockCount == count.getValue() || totalBlockCount == count.getValue()) { |
| LOG.debug("File {} has now been fully cached.", fileName); |
| fileCacheCompleted(fileName, size); |
| } else { |
| LOG.debug( |
| "Prefetch executor completed for {}, but only {} blocks were cached. " |
| + "Total blocks for file: {}. Checking for blocks pending cache in cache writer queue.", |
| fileName, count.getValue(), dataBlockCount); |
| if (ramCache.hasBlocksForFile(fileName.getName())) { |
| LOG.debug("There are still blocks pending caching for file {}. Will sleep 100ms " |
| + "and try the verification again.", fileName); |
| Thread.sleep(100); |
| notifyFileCachingCompleted(fileName, totalBlockCount, dataBlockCount, size); |
| } else { |
| LOG.info( |
| "We found only {} blocks cached from a total of {} for file {}, " |
| + "but no blocks pending caching. Maybe cache is full?", |
| count, dataBlockCount, fileName); |
| } |
| } |
| } catch (InterruptedException e) { |
| throw new RuntimeException(e); |
| } finally { |
| for (ReentrantReadWriteLock lock : locks) { |
| lock.readLock().unlock(); |
| } |
| } |
| } |
| |
| @Override |
| public Optional<Boolean> blockFitsIntoTheCache(HFileBlock block) { |
| long currentUsed = bucketAllocator.getUsedSize(); |
| boolean result = (currentUsed + block.getOnDiskSizeWithHeader()) < acceptableSize(); |
| return Optional.of(result); |
| } |
| |
| @Override |
| public Optional<Boolean> shouldCacheFile(String fileName) { |
| // if we don't have the file in fullyCachedFiles, we should cache it |
| return Optional.of(!fullyCachedFiles.containsKey(fileName)); |
| } |
| |
| @Override |
| public Optional<Boolean> isAlreadyCached(BlockCacheKey key) { |
| return Optional.of(getBackingMap().containsKey(key)); |
| } |
| |
| @Override |
| public Optional<Integer> getBlockSize(BlockCacheKey key) { |
| BucketEntry entry = backingMap.get(key); |
| if (entry == null) { |
| return Optional.empty(); |
| } else { |
| return Optional.of(entry.getOnDiskSizeWithHeader()); |
| } |
| |
| } |
| } |