<commit_before>/**
 * Copyright The Apache Software Foundation
 *
 * 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 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.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
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 org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
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.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.nio.ByteBuff;
import org.apache.hadoop.hbase.nio.RefCnt;
import org.apache.hadoop.hbase.protobuf.ProtobufMagic;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.IdReadWriteLock;
import org.apache.hadoop.hbase.util.IdReadWriteLock.ReferenceType;
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 on-heap or
 * off-heap memory {@link ByteBufferIOEngine} or in a file {@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";

  /** 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 ConcurrentHashMap<BlockCacheKey, BucketEntry> backingMap;

  /**
   * 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;

  /**
   * Used in tests. If this flag is false and the cache speed is very fast,
   * bucket cache will skip some blocks when caching. If the flag is true, we
   * will wait until blocks are flushed to IOEngine.
   */
  boolean wait_when_cache = false;

  private final BucketCacheStats cacheStats = new BucketCacheStats();

  private final String persistencePath;
  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>
   * Key set of offsets in BucketCache is limited so soft reference is the best choice here.
   */
  transient final IdReadWriteLock<Long> offsetLock = new IdReadWriteLock<>(ReferenceType.SOFT);

  private final NavigableSet<BlockCacheKey> blocksByHFile = new ConcurrentSkipListSet<>((a, b) -> {
    int nameComparison = a.getHfileName().compareTo(b.getHfileName());
    if (nameComparison != 0) {
      return nameComparison;
    }
    return Long.compare(a.getOffset(), b.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 static final String FILE_VERIFY_ALGORITHM =
    "hbase.bucketcache.persistent.file.integrity.check.algorithm";
  private static final String DEFAULT_FILE_VERIFY_ALGORITHM = "MD5";

  /**
   * Use {@link java.security.MessageDigest} class's encryption algorithms to check
   * persistent file integrity, default algorithm is MD5
   * */
  private String algorithm;

  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 {
    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);

    sanityCheckConfigs();

    LOG.info("Instantiating BucketCache with acceptableFactor: " + acceptableFactor + ", minFactor: " + minFactor +
        ", extraFreeFactor: " + extraFreeFactor + ", singleFactor: " + singleFactor + ", multiFactor: " + multiFactor +
        ", memoryFactor: " + memoryFactor);

    this.cacheCapacity = capacity;
    this.persistencePath = persistencePath;
    this.blockSize = blockSize;
    this.ioErrorsTolerationDuration = ioErrorsTolerationDuration;

    bucketAllocator = new BucketAllocator(capacity, bucketSizes);
    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 (ioEngine.isPersistent() && persistencePath != null) {
      try {
        retrieveFromFile(bucketSizes);
      } catch (IOException ioex) {
        LOG.error("Can't restore from file[" + persistencePath + "] because of ", ioex);
      }
    }
    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();
    }
  }

  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
   * @param ioEngineName
   * @param capacity
   * @param persistencePath
   * @return the IOEngine
   * @throws IOException
   */
  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");
    }
  }

  /**
   * 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, wait_when_cache);
  }

  /**
   * 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 (BlockCacheUtil.shouldReplaceExistingCacheBlock(this, cacheKey, cachedItem)) {
          cacheBlockWithWaitInternal(cacheKey, cachedItem, inMemory, wait);
        }
      } else {
        cacheBlockWithWaitInternal(cacheKey, cachedItem, inMemory, wait);
      }
    }
  }

  private void cacheBlockWithWaitInternal(BlockCacheKey cacheKey, Cacheable cachedItem,
      boolean inMemory, boolean wait) {
    if (!cacheEnabled) {
      return;
    }
    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,
              createRecycler(cacheKey));<commit_msg>HBASE-26155 JVM crash when scan (#3553)<commit_after>/**
 * Copyright The Apache Software Foundation
 *
 * 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 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.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
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 org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
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.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.nio.ByteBuff;
import org.apache.hadoop.hbase.nio.RefCnt;
import org.apache.hadoop.hbase.protobuf.ProtobufMagic;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.IdReadWriteLock;
import org.apache.hadoop.hbase.util.IdReadWriteLock.ReferenceType;
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 on-heap or
 * off-heap memory {@link ByteBufferIOEngine} or in a file {@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";

  /** 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 ConcurrentHashMap<BlockCacheKey, BucketEntry> backingMap;

  /**
   * 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;

  /**
   * Used in tests. If this flag is false and the cache speed is very fast,
   * bucket cache will skip some blocks when caching. If the flag is true, we
   * will wait until blocks are flushed to IOEngine.
   */
  boolean wait_when_cache = false;

  private final BucketCacheStats cacheStats = new BucketCacheStats();

  private final String persistencePath;
  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>
   * Key set of offsets in BucketCache is limited so soft reference is the best choice here.
   */
  transient final IdReadWriteLock<Long> offsetLock = new IdReadWriteLock<>(ReferenceType.SOFT);

  private final NavigableSet<BlockCacheKey> blocksByHFile = new ConcurrentSkipListSet<>((a, b) -> {
    int nameComparison = a.getHfileName().compareTo(b.getHfileName());
    if (nameComparison != 0) {
      return nameComparison;
    }
    return Long.compare(a.getOffset(), b.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 static final String FILE_VERIFY_ALGORITHM =
    "hbase.bucketcache.persistent.file.integrity.check.algorithm";
  private static final String DEFAULT_FILE_VERIFY_ALGORITHM = "MD5";

  /**
   * Use {@link java.security.MessageDigest} class's encryption algorithms to check
   * persistent file integrity, default algorithm is MD5
   * */
  private String algorithm;

  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 {
    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);

    sanityCheckConfigs();

    LOG.info("Instantiating BucketCache with acceptableFactor: " + acceptableFactor + ", minFactor: " + minFactor +
        ", extraFreeFactor: " + extraFreeFactor + ", singleFactor: " + singleFactor + ", multiFactor: " + multiFactor +
        ", memoryFactor: " + memoryFactor);

    this.cacheCapacity = capacity;
    this.persistencePath = persistencePath;
    this.blockSize = blockSize;
    this.ioErrorsTolerationDuration = ioErrorsTolerationDuration;

    bucketAllocator = new BucketAllocator(capacity, bucketSizes);
    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 (ioEngine.isPersistent() && persistencePath != null) {
      try {
        retrieveFromFile(bucketSizes);
      } catch (IOException ioex) {
        LOG.error("Can't restore from file[" + persistencePath + "] because of ", ioex);
      }
    }
    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();
    }
  }

  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
   * @param ioEngineName
   * @param capacity
   * @param persistencePath
   * @return the IOEngine
   * @throws IOException
   */
  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");
    }
  }

  /**
   * 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, wait_when_cache);
  }

  /**
   * 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 (BlockCacheUtil.shouldReplaceExistingCacheBlock(this, 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);
      }
    }
  }

  private void cacheBlockWithWaitInternal(BlockCacheKey cacheKey, Cacheable cachedItem,
      boolean inMemory, boolean wait) {
    if (!cacheEnabled) {
      return;
    }
    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,
              createRecycler(cacheKey));
{'fraction_non_alphanumeric': 0.08881789137380192, 'fraction_numerical': 0.004932907348242812, 'mean_word_length': 4.132624950806769, 'pattern_counts': {'":': 2, '<': 53, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2280779', 'n_tokens_mistral': 11316, 'n_tokens_neox': 10999, 'n_words': 3440}