LevelDB源码分析之十一：cache

Cache为接口类。ShardedLRUCache继承自Cache，实现了Cache中的缓存操作方法。
ShardedLRUCache封装了16个LRUCache缓存片，每次对缓存的读取、插入、查找、删除操作都是调用某个缓存片LRUCache中的相应方法完成。
LRUCache为一个循环双向链表，与标准实现一致。其“头结点”lru_的prev始终指向最新结点，next始终指向最久未用节点，其对象容器为HashTable（哈希表），用于为LRUCache提供快速的查找操作。
LRUHandle为结点类。其next_hash指针用于HashTable中的bucket单向链表 。next和prev指针用于LRUCache循环双向链表的后继和前驱。

他们之间的关系如下图所示：



关于HashTable可以参考：哈希表（Hash Table）

关于LRUCache可以参考：LRU Cache原理和实现

一.Cache接口类

class Cache;

// 创建Cache的全局方法，capacity为容量大小
extern Cache* NewLRUCache(size_t capacity);

class Cache {
public:
Cache() { }

virtual ~Cache();

// 表示结点的结构体
struct Handle { };

// 插入键值对，并指定占用的缓存大小（charge），返回被插入的结点。
// 如果插入的键值对用不到了，传给deleter函数。
// 如果返回值用不到了，记得调用this->Release(handle)释放。
virtual Handle* Insert(const Slice& key, void* value, size_t charge,
void (*deleter)(const Slice& key, void* value)) = 0;

// 如果没找到结点，返回NULL。否则返回找到的结点。
// 如果返回值用不到了，记得调用this->Release(handle)释放。
virtual Handle* Lookup(const Slice& key) = 0;

// 释放结点，注意不要重复释放。
virtual void Release(Handle* handle) = 0;

// 返回结点handle中的Value值，注意判断handle的有效性。
virtual void* Value(Handle* handle) = 0;

// 删除包含key的结点
virtual void Erase(const Slice& key) = 0;

// 返回数字ID，用于处理多线程同时访问缓存时的同步
virtual uint64_t NewId() = 0;

private:
void LRU_Remove(Handle* e);
void LRU_Append(Handle* e);
void Unref(Handle* e);

struct Rep;
Rep* rep_;

// No copying allowed
Cache(const Cache&);
void operator=(const Cache&);
};

二.LRUHandle

struct LRUHandle {
void* value;
void (*deleter)(const Slice&, void* value);//删除器。当refs == 0时，调用deleter完成value对象释放。
LRUHandle* next_hash; // 作为HashTable中的节点，指向hash值相同的节点（解决hash冲突采用链地址法）
LRUHandle* next;      // 作为LRUCache中的节点，指向后继
LRUHandle* prev;      // 作为LRUCache中的节点，指向前驱
size_t charge;        // 用户指定占用缓存的大小
size_t key_length;    // key长度
uint32_t refs;        // 引用计数
uint32_t hash;        // 哈希值
char key_data[1];     // key内容的起始位置

Slice key() const {
// For cheaper lookups, we allow a temporary Handle object
// to store a pointer to a key in "value".
if (next == this) {
return *(reinterpret_cast<Slice*>(value));
} else {
return Slice(key_data, key_length);
}
}
};

一个LRUHandle就是一个结点，这个结构体设计的巧妙之处在于，它既可以作为HashTable中的结点，也可以作为LRUCache中的结点。

关于char key_data[1]，在LevelDB源码分析之五：skiplist（1）这篇博客中又类似用法。key_data位于结构体的最后面，可在申请内存时，申请足够多的空间。 

往下面看会看到这句：

LRUHandle* e = reinterpret_cast<LRUHandle*>(
malloc(sizeof(LRUHandle)-1 + key.size()));

注意在使用malloc申请空间时，sizeof(LRUHandle)-1，其中减去的1就是key_data[1]，然后根据key.size()动态申请空间。最后，key_data还是指向这块空间的。

三.HashTable

LRUCache的实现并为用C++标准库内置的哈希表，用的是自己实现的哈希表。

class HandleTable {
public:
HandleTable() : length_(0), elems_(0), list_(NULL) { Resize(); }
~HandleTable() { delete[] list_; }

LRUHandle* Lookup(const Slice& key, uint32_t hash) {
return *FindPointer(key, hash);
}

LRUHandle* Insert(LRUHandle* h) {
LRUHandle** ptr = FindPointer(h->key(), h->hash);
LRUHandle* old = *ptr;
h->next_hash = (old == NULL ? NULL : old->next_hash);
// 不管找没找到h结点，都是可以直接将h替换到*ptr的。
// 1.如果找到了，因为key相同，直接替换相当于替换结点中的value
// 2.如果没找到，直接替换是理所当然的了
*ptr = h;
//如果没找到，相当于要添加了一个新的结点，此时结点总数elems_加1
if (old == NULL) {
++elems_;
if (elems_ > length_) {
// Since each cache entry is fairly large, we aim for a small
// average linked list length (<= 1).
// 当elems_ > length_时进行扩容，这样可以保证在大部分情况下，所有以哈希地址为头的
// 链表平均最多只有一个结点。因为结点比较大，扩容后能保证查找的时间复制度为O(1)。
Resize();
}
}
// 将old返回很重要，因为这个被摘到的handle需要在函数外面销毁。
return old;
}
// 删除结点，比较简单
LRUHandle* Remove(const Slice& key, uint32_t hash) {
LRUHandle** ptr = FindPointer(key, hash);
LRUHandle* result = *ptr;
if (result != NULL) {
*ptr = result->next_hash;
--elems_;
}
return result;
}

private:
// The table consists of an array of buckets where each bucket is
// a linked list of cache entries that hash into the bucket.
uint32_t length_;  //哈希地址数组的长度
uint32_t elems_;   //哈希表中所有结点的总数
LRUHandle** list_; //哈希地址数组的二级指针

// Return a pointer to slot that points to a cache entry that
// matches key/hash.  If there is no such cache entry, return a
// pointer to the trailing slot in the corresponding linked list.
// 如果找到了结点，返回该结点的二级指针。
// 如果没找到结点，分两种情况:
// 1.根据哈希值求出的哈希地址是空的，也就是说以该哈希地址（哈希桶）为头的单链表
//   还没有结点。hash & (length_ - 1)的取值范围是0—（length_-1）。此时返回
//   哈希地址的二级指针，*ptr=NULL 。
// 2.查找到了以某哈希地址为头的单链表的尾部，也没找到该结点。此时返回
//   尾部结点next_hash域的二级指针，*ptr=NULL 。
LRUHandle** FindPointer(const Slice& key, uint32_t hash) {
LRUHandle** ptr = &list_[hash & (length_ - 1)];
while (*ptr != NULL &&
((*ptr)->hash != hash || key != (*ptr)->key())) {
ptr = &(*ptr)->next_hash;
}
return ptr;
}
// 哈希表扩容
// HandleTable内部维护了一个LRUHandle*的数组，默认大小为4。随着插入数据的增多，
// 该数组会自动增长一倍，并将原数组中的数据重新分配到新的数组中。
void Resize() {
uint32_t new_length = 4;
while (new_length < elems_) {
new_length *= 2;
}
LRUHandle** new_list = new LRUHandle*[new_length];
memset(new_list, 0, sizeof(new_list[0]) * new_length);
uint32_t count = 0;
// 需要注意的是，从新分配结点的时候，结点都往每个链表的头部插入的。
// 而正常的Insert操作，相同hash值的结点是插入到链表的尾部
for (uint32_t i = 0; i < length_; i++) {
LRUHandle* h = list_[i];
while (h != NULL) {
LRUHandle* next = h->next_hash;
Slice key = h->key();
uint32_t hash = h->hash;
LRUHandle** ptr = &new_list[hash & (new_length - 1)];
h->next_hash = *ptr;
*ptr = h;
h = next;
count++;
}
}
assert(elems_ == count);
delete[] list_;
list_ = new_list;
length_ = new_length;
}
};

HandleTable是哈希表，但比较奇怪的是，查找、插入、删除动作除了传入key外，还要自备hash值。这样做是因为，hash值除了HandleTable中使用，在外部做多级缓存时也需要，后面会提到。
四.LRUCache

LRUCache::LRUCache()
: usage_(0),
last_id_(0) {
// Make empty circular linked list
// 创建空的循环链表
lru_.next = &lru_;
lru_.prev = &lru_;
}

LRUCache::~LRUCache() {
for (LRUHandle* e = lru_.next; e != &lru_; ) {
LRUHandle* next = e->next;
// 如果不为1，说明LRUCache的使用者并未主动调用Release或Erase方法。
// 因为初始的引用计数为2，调用Release或Erase时，引用计数会减一。
assert(e->refs == 1);
Unref(e);
e = next;
}
}
// 引用计数减一。引用计数变为0时，调用删除器deleter。
void LRUCache::Unref(LRUHandle* e) {
assert(e->refs > 0);
e->refs--;
if (e->refs <= 0) {
usage_ -= e->charge;
(*e->deleter)(e->key(), e->value);
free(e);
}
}

void LRUCache::LRU_Remove(LRUHandle* e) {
e->next->prev = e->prev;
e->prev->next = e->next;
}

void LRUCache::LRU_Append(LRUHandle* e) {
// Make "e" newest entry by inserting just before lru_
e->next = &lru_;
e->prev = lru_.prev;
e->prev->next = e;
e->next->prev = e;
}
// 根据LRUCache规则，被访问的结点要移动到双向链表的lru_结点之前
// 移动只是改变了结点前驱指针和后继指针的指向，结点的存储位置并没变化
// 返回被找到的结点，如果没找到，返回NULL
Cache::Handle* LRUCache::Lookup(const Slice& key, uint32_t hash) {
  MutexLock l(&mutex_);
  LRUHandle* e = table_.Lookup(key, hash);
  if (e != NULL) {
    e->refs++;// 这里为何要加一，后面会说到
    LRU_Remove(e);
    LRU_Append(e);
  }
  // 如果返回值不为NULL且用不到了，记得调用Relese或Erase释放。
  return reinterpret_cast<Cache::Handle*>(e);
}
// 释放结点
void LRUCache::Release(Cache::Handle* handle) {
MutexLock l(&mutex_);
Unref(reinterpret_cast<LRUHandle*>(handle));
}

Cache::Handle* LRUCache::Insert(
const Slice& key, uint32_t hash, void* value, size_t charge,
void (*deleter)(const Slice& key, void* value)) {
MutexLock l(&mutex_);

LRUHandle* e = reinterpret_cast<LRUHandle*>(
malloc(sizeof(LRUHandle)-1 + key.size()));
e->value = value;
e->deleter = deleter;
e->charge = charge;
e->key_length = key.size();
e->hash = hash;
// 初始设为2，一个是给LRUCache自身析构时用，一个是给外部调用Release或Erase时用。
e->refs = 2;
memcpy(e->key_data, key.data(), key.size());
LRU_Append(e);
usage_ += charge;
// 当哈希表中已存在hash值相同的结点时，将原有的结点从双向链表中移除，
// 并释放该结点。
// 这里不需要调用哈希表的Remove方法将该结点从哈希表中移除，因为Insert
// 的时候实际上已经移除了。
// 这段代码也可以看出为何哈希表的Insert方法要返回旧结点？因为不返回旧
// 结点，后续就无法对该结点进行LRU_Remove操作了。
LRUHandle* old = table_.Insert(e);
if (old != NULL) {
LRU_Remove(old);
Unref(old);
}
// 如果已用容量超过了总容量且头结点lru_还有后继。
// 删除lru_的后继结点，根据LRUCache规则，这个结点最近用的最少。
// 该结点既要从哈希表中移除，也要从双向链表中移除，然后再释放。
while (usage_ > capacity_ && lru_.next != &lru_) {
LRUHandle* old = lru_.next;
LRU_Remove(old);
table_.Remove(old->key(), old->hash);
Unref(old);
}
// 返回新插入的结点，LRUCache的使用者需要主动调用该结点的Relese或Erase方法来释放结点。
return reinterpret_cast<Cache::Handle*>(e);
}
// 删除结点，注意和Release方法的不同。删除结点先将结点从哈希表和
// 双向链表中移除，然后再释放该结点。
void LRUCache::Erase(const Slice& key, uint32_t hash) {
MutexLock l(&mutex_);
LRUHandle* e = table_.Remove(key, hash);
if (e != NULL) {
LRU_Remove(e);
Unref(e);
}
}

1.为何要维护引用计数？

在Insert时，引用计数初始化为2，一个是给LRUCache自身析构时用，一个是给外部调用Release或Erase时用。Insert时，返回的是新插入的结点，插入完成后，需要调用该结点Release或Erase方法将引用计数减1，那么此时该结点的引用计数就是1了。在LRUCache析构时，会先将结点的引用计数再减1，如果此时引用计数为0，则调用deleter，并将该结点彻底从内存中free。

在Lookup时，如果查找接结点存在，此时引用计数会加1，也即是变成了3。此时，在用户持有该结点的期间，该缓存可能被删除（多种原因，如：超过缓存容量触发回收、具有相同key的新缓存插入、整个缓存被析构等），导致用户访问到非法内存，程序崩溃。因此，需要使用引用计数要加1来维护结点的生命周期。因为Lookup返回的是找到的结点，用户在查找完成后，要主动调用该结点的Release或Erase来使引用计数从新变成2。

2.LRUHandle为什么会被同时置于哈希表和双向链表之中？

注意看LookUp的实现，如果单纯使用链表，则仅能提供O(n)的查询效率，所以在查询时，利用了哈希表实现O(1)的查询。

那么，如果单纯使用哈希表呢？虽然可以实现O(1)的查询，但却无法更新缓存节点的访问时间。这是因为链表可以按照固定的顺序被遍历，而哈希表中的节点无法提供固定的遍历顺序（考虑Resize前后）。

那么，可不可以将访问时间记录在Handle中，然后仅用哈希表，这样既可以实现O(1)的查询，又可以方便地更新缓存记录的访问时间，岂不美哉？但是，如果没有按访问时间排序的链表，当触发缓存回收时，我们如何快速定位到哪些缓存记录要被回收呢？

链表O(n)的查询效率、哈希表不支持排序，两种数据结构单独都无法满足这里的需求。作者大神巧妙地将二者结合，取长补短，利用哈希表实现O(1)的查询，利用链表维持对缓存记录按访问时间排序

	查询	插入	删除	排序
链表	O(n)	O(1)	O(1)	支持
哈希表	O(1)	O(1)	O(1)	不支持

注1：哈希表实现O(1)操作的前提是：平均每哈希桶元素数 <= 1

注2：为了保持平均哈希桶元素数，必要时必须Resize，而Resize后，原有顺序将被打破
五.ShardedLRUCache

static const int kNumShardBits = 4;
static const int kNumShards = 1 << kNumShardBits;

class ShardedLRUCache : public Cache {
 private:
  // 16片LRUCache缓存
  LRUCache shard_[kNumShards];
  port::Mutex id_mutex_;
  uint64_t last_id_;
  // 使用哈希函数求出哈希值
  static inline uint32_t HashSlice(const Slice& s) {
    return Hash(s.data(), s.size(), 0);
  }
  // 取哈希值的高四位来定位使用那片LRUCache缓存
  static uint32_t Shard(uint32_t hash) {
    return hash >> (32 - kNumShardBits);
  }

 public:
  // capacity是Cache大小，其单位可以自行指定（如table cache，一个sstable文件的索引信息是一个单位，
  // 而block cache，一个byte是一个单位）
  explicit ShardedLRUCache(size_t capacity)
      : last_id_(0) {
    const size_t per_shard = (capacity + (kNumShards - 1)) / kNumShards;
    for (int s = 0; s < kNumShards; s++) {
      shard_[s].SetCapacity(per_shard);
    }
  }
  virtual ~ShardedLRUCache() { }
  virtual Handle* Insert(const Slice& key, void* value, size_t charge,
                         void (*deleter)(const Slice& key, void* value)) {
    const uint32_t hash = HashSlice(key);
    return shard_[Shard(hash)].Insert(key, hash, value, charge, deleter);
  }
  virtual Handle* Lookup(const Slice& key) {
    const uint32_t hash = HashSlice(key);
    return shard_[Shard(hash)].Lookup(key, hash);
  }
  virtual void Release(Handle* handle) {
    LRUHandle* h = reinterpret_cast<LRUHandle*>(handle);
    shard_[Shard(h->hash)].Release(handle);
  }
  virtual void Erase(const Slice& key) {
    const uint32_t hash = HashSlice(key);
    shard_[Shard(hash)].Erase(key, hash);
  }
  // 返回handle结点的value值
  virtual void* Value(Handle* handle) {
    return reinterpret_cast<LRUHandle*>(handle)->value;
  }
  // 返回数字ID，用于处理多线程同时访问缓存时的同步
  virtual uint64_t NewId() {
    MutexLock l(&id_mutex_);
    return ++(last_id_);
  }
};

SharedLRUCache到底是什么呢？我们为什么需要它？这是因为levelDB是多线程的，每个线程访问缓冲区的时候都会将缓冲区锁住，为了多线程访问，尽可能快速，减少锁开销，ShardedLRUCache内部有16个LRUCache，查找Key时首先计算key属于哪一个分片，分片的计算方法是取32位hash值的高4位，然后在相应的LRUCache中进行查找，这样就大大减少了多线程的访问锁的开销。这种设计思路，非常值得我辈学习，致敬大神。

六.cache的使用

为了加快查找速度，LevelDB在内存中采用Cache的方式，在table中采用bloom filter的方式，尽最大可能加快随机读操作。LevelDB的Cache分为两种，分别是table cache和block cache。

1.table cache

table cache缓存的是table的索引数据，类似于文件系统中对inode的缓存。table cache默认大小是1000，注意此处缓存的是1000个table文件的索引信息，而不是1000个字节。table cache的大小由options.max_open_files确定，其最小值为20-10，最大值为50000－10。

2.block cache

block cache是缓存的block数据，block是table文件内组织数据的单位，也是从持久化存储中读取和写入的单位。由于table是按照key有序分布的，因此一个block内的数据也是按照key紧邻排布的（有序依照使用者传入的比较函数，默认按照字典序），类似于Linux中的page cache。

block默认大小为4k，由LevelDB调用open函数打开数据库时时传入的options.block_size参数指定。LevelDB的代码中限制的block最小大小为1k，最大大小为4M。对于频繁做scan操作的应用，可适当调大此参数，对大量小value随机读取的应用，也可尝试调小该参数。

block cache默认实现是一个8M大小的ShardedLRUCache，此参数由options.block_cache设定。当然也可根据应用需求，提供自定义的缓存策略。注意，此处的大小是未压缩的block大小。

参考链接：https://www.jianshu.com/p/9e7773432772

参考链接：http://www.360doc.com/content/14/0325/16/15064667_363619144.shtml