Hbase 内核剖析

public static class CachedBlocksByFile {    private int count;    private int dataBlockCount;    private long size;    private long dataSize;    private final long now = System.nanoTime();    /**     * How many blocks to look at before we give up.     * There could be many millions of blocks. We don't want the     * ui to freeze while we run through 1B blocks... users will     * think hbase dead. UI displays warning in red when stats     * are incomplete.     */    private final int max;    public static final int DEFAULT_MAX = 1000000;
    CachedBlocksByFile() {      this(null);    }
    CachedBlocksByFile(final Configuration c) {      this.max = c == null? DEFAULT_MAX: c.getInt("hbase.ui.blockcache.by.file.max", DEFAULT_MAX);    }
    /**     * Map by filename. use concurent utils because we want our Map and contained blocks sorted.     */    private transient NavigableMap<String, NavigableSet<CachedBlock>> cachedBlockByFile =      new ConcurrentSkipListMap<>();

public final class PrefetchExecutor {
  private static final Logger LOG = LoggerFactory.getLogger(PrefetchExecutor.class);
  /** Futures for tracking block prefetch activity */  private static final Map<Path,Future<?>> prefetchFutures = new ConcurrentSkipListMap<>();  /** Executor pool shared among all HFiles for block prefetch */  private static final ScheduledExecutorService prefetchExecutorPool;  /** Delay before beginning prefetch */  private static final int prefetchDelayMillis;  /** Variation in prefetch delay times, to mitigate stampedes */  private static final float prefetchDelayVariation;

public class ServerManager {  public static final String WAIT_ON_REGIONSERVERS_MAXTOSTART =      "hbase.master.wait.on.regionservers.maxtostart";
  public static final String WAIT_ON_REGIONSERVERS_MINTOSTART =      "hbase.master.wait.on.regionservers.mintostart";
  public static final String WAIT_ON_REGIONSERVERS_TIMEOUT =      "hbase.master.wait.on.regionservers.timeout";
  public static final String WAIT_ON_REGIONSERVERS_INTERVAL =      "hbase.master.wait.on.regionservers.interval";
  /**   * see HBASE-20727   * if set to true, flushedSequenceIdByRegion and storeFlushedSequenceIdsByRegion   * will be persisted to HDFS and loaded when master restart to speed up log split   */  public static final String PERSIST_FLUSHEDSEQUENCEID =      "hbase.master.persist.flushedsequenceid.enabled";
  public static final boolean PERSIST_FLUSHEDSEQUENCEID_DEFAULT = true;
  public static final String FLUSHEDSEQUENCEID_FLUSHER_INTERVAL =      "hbase.master.flushedsequenceid.flusher.interval";
  public static final int FLUSHEDSEQUENCEID_FLUSHER_INTERVAL_DEFAULT =      3 * 60 * 60 * 1000; // 3 hours
  public static final String MAX_CLOCK_SKEW_MS = "hbase.master.maxclockskew";
  private static final Logger LOG = LoggerFactory.getLogger(ServerManager.class);
  // Set if we are to shutdown the cluster.  private AtomicBoolean clusterShutdown = new AtomicBoolean(false);
  /**   * The last flushed sequence id for a region.   */  private final ConcurrentNavigableMap<byte[], Long> flushedSequenceIdByRegion =    new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR);

public class RegionStates {  private static final Logger LOG = LoggerFactory.getLogger(RegionStates.class);
  // This comparator sorts the RegionStates by time stamp then Region name.  // Comparing by timestamp alone can lead us to discard different RegionStates that happen  // to share a timestamp.  private static class RegionStateStampComparator implements Comparator<RegionState> {    @Override    public int compare(final RegionState l, final RegionState r) {      int stampCmp = Long.compare(l.getStamp(), r.getStamp());      return stampCmp != 0 ? stampCmp : RegionInfo.COMPARATOR.compare(l.getRegion(), r.getRegion());    }  }
  public final static RegionStateStampComparator REGION_STATE_STAMP_COMPARATOR =      new RegionStateStampComparator();
  private final Object regionsMapLock = new Object();
  // TODO: Replace the ConcurrentSkipListMaps  /**   * RegionName -- i.e. RegionInfo.getRegionName() -- as bytes to {@link RegionStateNode}   */  private final ConcurrentSkipListMap<byte[], RegionStateNode> regionsMap =      new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR);

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 interface BloomFilter extends BloomFilterBase {
  /**   * Check if the specified key is contained in the bloom filter.   * @param keyCell the key to check for the existence of   * @param bloom bloom filter data to search. This can be null if auto-loading   *        is supported.   * @param type The type of Bloom ROW/ ROW_COL   * @return true if matched by bloom, false if not   */  boolean contains(Cell keyCell, ByteBuff bloom, BloomType type);
  /**   * Check if the specified key is contained in the bloom filter.   * @param buf data to check for existence of   * @param offset offset into the data   * @param length length of the data   * @param bloom bloom filter data to search. This can be null if auto-loading   *        is supported.   * @return true if matched by bloom, false if not   */  boolean contains(byte[] buf, int offset, int length, ByteBuff bloom);
  /**   * @return true if this Bloom filter can automatically load its data   *         and thus allows a null byte buffer to be passed to contains()   */  boolean supportsAutoLoading();}

public class CompoundBloomFilter extends CompoundBloomFilterBase    implements BloomFilter {
  /** Used to load chunks on demand */  private HFile.Reader reader;
  private HFileBlockIndex.BlockIndexReader index;
  private int hashCount;  private Hash hash;
  private long[] numQueriesPerChunk;  private long[] numPositivesPerChunk;
  /**   * De-serialization for compound Bloom filter metadata. Must be consistent   * with what {@link CompoundBloomFilterWriter} does.   *   * @param meta serialized Bloom filter metadata without any magic blocks   * @throws IOException   */  public CompoundBloomFilter(DataInput meta, HFile.Reader reader)      throws IOException {    this.reader = reader;
    totalByteSize = meta.readLong();    hashCount = meta.readInt();    hashType = meta.readInt();    totalKeyCount = meta.readLong();    totalMaxKeys = meta.readLong();    numChunks = meta.readInt();    byte[] comparatorClassName = Bytes.readByteArray(meta);    // The writer would have return 0 as the vint length for the case of     // Bytes.BYTES_RAWCOMPARATOR.  In such cases do not initialize comparator, it can be    // null    if (comparatorClassName.length != 0) {      comparator = FixedFileTrailer.createComparator(Bytes.toString(comparatorClassName));    }
    hash = Hash.getInstance(hashType);    if (hash == null) {      throw new IllegalArgumentException("Invalid hash type: " + hashType);    }    // We will pass null for ROW block    if(comparator == null) {      index = new HFileBlockIndex.ByteArrayKeyBlockIndexReader(1);    } else {      index = new HFileBlockIndex.CellBasedKeyBlockIndexReader(comparator, 1);    }    index.readRootIndex(meta, numChunks);  }
  @Override  public boolean contains(byte[] key, int keyOffset, int keyLength, ByteBuff bloom) {    int block = index.rootBlockContainingKey(key, keyOffset, keyLength);    if (block < 0) {      return false; // This key is not in the file.    }    boolean result;    HFileBlock bloomBlock = getBloomBlock(block);    try {      ByteBuff bloomBuf = bloomBlock.getBufferReadOnly();      result = BloomFilterUtil.contains(key, keyOffset, keyLength, bloomBuf,          bloomBlock.headerSize(), bloomBlock.getUncompressedSizeWithoutHeader(), hash, hashCount);    } finally {      // After the use, should release the block to deallocate byte buffers.      bloomBlock.release();    }    if (numPositivesPerChunk != null && result) {      // Update statistics. Only used in unit tests.      ++numPositivesPerChunk[block];    }    return result;  }
  private HFileBlock getBloomBlock(int block) {    HFileBlock bloomBlock;    try {      // We cache the block and use a positional read.      bloomBlock = reader.readBlock(index.getRootBlockOffset(block),          index.getRootBlockDataSize(block), true, true, false, true, BlockType.BLOOM_CHUNK, null);    } catch (IOException ex) {      // The Bloom filter is broken, turn it off.      throw new IllegalArgumentException("Failed to load Bloom block", ex);    }
    if (numQueriesPerChunk != null) {      // Update statistics. Only used in unit tests.      ++numQueriesPerChunk[block];    }    return bloomBlock;  }
  @Override  public boolean contains(Cell keyCell, ByteBuff bloom, BloomType type) {    int block = index.rootBlockContainingKey(keyCell);    if (block < 0) {      return false; // This key is not in the file.    }    boolean result;    HFileBlock bloomBlock = getBloomBlock(block);    try {      ByteBuff bloomBuf = bloomBlock.getBufferReadOnly();      result = BloomFilterUtil.contains(keyCell, bloomBuf, bloomBlock.headerSize(),        bloomBlock.getUncompressedSizeWithoutHeader(), hash, hashCount, type);    } finally {      // After the use, should release the block to deallocate the byte buffers.      bloomBlock.release();    }    if (numPositivesPerChunk != null && result) {      // Update statistics. Only used in unit tests.      ++numPositivesPerChunk[block];    }    return result;  }

public enum BloomType {  /**   * Bloomfilters disabled   */  NONE,  /**   * Bloom enabled with Table row as Key   */  ROW,  /**   * Bloom enabled with Table row & column (family+qualifier) as Key   */  ROWCOL,  /**   * Bloom enabled with Table row prefix as Key, specify the length of the prefix   */  ROWPREFIX_FIXED_LENGTH}

//LRUBlockCachepublic class LruBlockCache implements FirstLevelBlockCache{...}//BucketCachepublic class BucketCache implements BlockCache, HeapSize{...}