hbase源码系列（十三）缓存机制MemStore与Block Cache

protected final MemStore memstore;
private final CacheConfig cacheConf;
final StoreEngine<?, ?, ?, ?> storeEngine;

long add(final KeyValue kv) {
    this.lock.readLock().lock();
    try {
      KeyValue toAdd = maybeCloneWithAllocator(kv);
      return internalAdd(toAdd);
    } finally {
      this.lock.readLock().unlock();
    }
}

private KeyValue maybeCloneWithAllocator(KeyValue kv) {
    if (allocator == null) {
      return kv;
    }
    int len = kv.getLength();
    //从allocator当中分配出来len长度的非堆空间
    Allocation alloc = allocator.allocateBytes(len);
    if (alloc == null) {
      // 太大了，allocator决定不给它分配
      return kv;
    }
    //用allocator生成的空间，new一个kv出来
    assert alloc != null && alloc.getData() != null;
    System.arraycopy(kv.getBuffer(), kv.getOffset(), alloc.getData(), alloc.getOffset(), len);
    KeyValue newKv = new KeyValue(alloc.getData(), alloc.getOffset(), len);
    newKv.setMvccVersion(kv.getMvccVersion());
    return newKv;
 }

static class Chunk {
    /* 实际数据保存的地方，被不停地分配 */
    private byte[] data;
    private static final int UNINITIALIZED = -1;
    private static final int OOM = -2;
    /* 下一个chunk的起始位置，也是上一个chunk的结束位置 */
    private AtomicInteger nextFreeOffset = new AtomicInteger(UNINITIALIZED);

    /** 分配给了多少个kv */
    private AtomicInteger allocCount = new AtomicInteger();

    /** Chunk的大小 */
    private final int size;

/** 可分配的最大值，超过这个值就不给它分配了，默认值是256K */
hbase.hregion.memstore.mslab.max.allocation 默认值是256  * 1024
/** 每个Chunk的大小，默认是2M */
hbase.hregion.memstore.mslab.chunksize 默认值是2048 * 1024

hbase.regionserver.global.memstore.upperLimit 
hbase.regionserver.global.memstore.lowerLimit

private long internalAdd(final KeyValue toAdd) {
    long s = heapSizeChange(toAdd, addToKVSet(toAdd));
    //把时间戳范围加到内部去
    timeRangeTracker.includeTimestamp(toAdd);
    this.size.addAndGet(s);
    return s;
}

volatile KeyValueSkipListSet kvset;
volatile KeyValueSkipListSet snapshot;

　　   PriorityQueue<BlockBucket> bucketQueue =
        new PriorityQueue<BlockBucket>(3);

      bucketQueue.add(bucketSingle);
      bucketQueue.add(bucketMulti);
      bucketQueue.add(bucketMemory);

      int remainingBuckets = 3;
      long bytesFreed = 0;

      BlockBucket bucket;
      while((bucket = bucketQueue.poll()) != null) {
        long overflow = bucket.overflow();
        if(overflow > 0) {
          //把要释放的空间bytesToFree分给3个bucket，3个分完
          long bucketBytesToFree = Math.min(overflow,
            (bytesToFree - bytesFreed) / remainingBuckets);
          bytesFreed += bucket.free(bucketBytesToFree);
        }
        remainingBuckets--;
      }

public static final HTableDescriptor META_TABLEDESC = new HTableDescriptor(
      TableName.META_TABLE_NAME,
      new HColumnDescriptor[] {
          new HColumnDescriptor(HConstants.CATALOG_FAMILY)
              // 保持10个版本是为了帮助调试
              .setMaxVersions(10)
              .setInMemory(true)
              .setBlocksize(8 * 1024)
              .setScope(HConstants.REPLICATION_SCOPE_LOCAL)
              // 不使用BloomFilter
              .setBloomFilterType(BloomType.NONE)
});

protected long evictBlock(CachedBlock block, boolean evictedByEvictionProcess) {
　　//从map里面删除
    map.remove(block.getCacheKey());if (evictedByEvictionProcess && victimHandler != null) {
      boolean wait = getCurrentSize() < acceptableSize();
      boolean inMemory = block.getPriority() == BlockPriority.MEMORY;
　　　//保存到victimHandler里面
      victimHandler.cacheBlockWithWait(block.getCacheKey(), block.getBuffer(),
          inMemory, wait);
    }
    return block.heapSize()
}

public Cacheable getBlock(BlockCacheKey cacheKey, boolean caching, boolean repeat) {
    CachedBlock cb = map.get(cacheKey);
    if(cb == null) {if (victimHandler != null)
        return victimHandler.getBlock(cacheKey, caching, repeat);
      return null;
    }
    return cb.getBuffer();
}

// Store/read block data
IOEngine ioEngine;
// 内存map
private ConcurrentHashMap<BlockCacheKey, RAMQueueEntry> ramCache;
// 后备队列，质保存块的索引信息，比如offset, length
private ConcurrentHashMap<BlockCacheKey, BucketEntry> backingMap;

    //已经有就不加啦
    if (backingMap.containsKey(cacheKey) || ramCache.containsKey(cacheKey))
      return;
    //写入一级缓存
    RAMQueueEntry re = new RAMQueueEntry(cacheKey, cachedItem, accessCount.incrementAndGet(), inMemory);
    ramCache.put(cacheKey, re);
    //用哈希值给计算出一个随机的队列来
    int queueNum = (cacheKey.hashCode() & 0x7FFFFFFF) % writerQueues.size();
    //把实体也插入到写入队列
    BlockingQueue<RAMQueueEntry> bq = writerQueues.get(queueNum);

    　List<RAMQueueEntry> entries = new ArrayList<RAMQueueEntry>();
      try {
        while (cacheEnabled && writerEnabled) {
          try {
            //从inputQueue拿出来放到entries，然后再对entries操作
            entries.add(inputQueue.take());
            inputQueue.drainTo(entries);
          } catch (InterruptedException ie) {
            if (!cacheEnabled) break;
          }
          doDrain(entries);
        }

final private static class Bucket {
    //基准起始位置
    private long baseOffset;
    //每个item分配的大小
    private int itemAllocationSize; 
    //对应的在bucketSizeInfos中的位置
    private int sizeIndex;
    //总容量
    private int itemCount;
    private int freeList[];
    //空闲的数量
    private int freeCount;
    //已经使用的数量
    private int usedCount;
}

final class BucketSizeInfo {
    // Free bucket means it has space to allocate a block;
    // Completely free bucket means it has no block.
    private List<Bucket> bucketList, freeBuckets, completelyFreeBuckets;
    private int sizeIndex;
}

　　　 //序列化长度 = 数据长度 + 额外的序列化的长度16个字节
      int len = data.getSerializedLength();
      // This cacheable thing can‘t be serialized...
      if (len == 0) return null;
      //bucketAllocator给分配点空间
      long offset = bucketAllocator.allocateBlock(len);
      //生成一个实体
      BucketEntry bucketEntry = new BucketEntry(offset, len, accessTime, inMemory);
      //设置Deserializer，具体的实现在HFileBlock当中
      bucketEntry.setDeserialiserReference(data.getDeserializer(), deserialiserMap);
      try {
        if (data instanceof HFileBlock) {
          ByteBuffer sliceBuf = ((HFileBlock) data).getBufferReadOnlyWithHeader();
          sliceBuf.rewind();
          assert len == sliceBuf.limit() + HFileBlock.EXTRA_SERIALIZATION_SPACE;
          ByteBuffer extraInfoBuffer = ByteBuffer.allocate(HFileBlock.EXTRA_SERIALIZATION_SPACE);
          ((HFileBlock) data).serializeExtraInfo(extraInfoBuffer);
          //先写入数据信息，再写入头信息
          ioEngine.write(sliceBuf, offset);
          ioEngine.write(extraInfoBuffer, offset + len - HFileBlock.EXTRA_SERIALIZATION_SPACE);
        } else {
          //如果不是HFileBlock的话，把数据序列化到bb当中，然后写入到IOEngine
          ByteBuffer bb = ByteBuffer.allocate(len);
          data.serialize(bb);
          ioEngine.write(bb, offset);
        }
      } catch (IOException ioe) {
        // 出错了就释放掉这个这个块
        bucketAllocator.freeBlock(offset);
        throw ioe;
　　  }