LevelDB源码分析之十一：cache
时间：2019-10-09 13:44:05
手机看文章
扫描二维码
随时随地手机看文章
[导读]Cache为接口类。ShardedLRUCache继承自Cache，实现了Cache中的缓存操作方法。ShardedLRUCache封装了16个LRUCache缓存片，每次对缓存的读取、插入、查找、删
Cache为接口类。ShardedLRUCache继承自Cache，实现了Cache中的缓存操作方法。
ShardedLRUCache封装了16个LRUCache缓存片，每次对缓存的读取、插入、查找、删除操作都是调用某个缓存片LRUCache中的相应方法完成。
LRUCache为一个循环双向链表，与标准实现一致。其“头结点”lru_的prev始终指向最新结点，next始终指向最久未用节点，其对象容器为HashTable（哈希表），用于为LRUCache提供快速的查找操作。
关于char key_data[1]，在LevelDB源码分析之五：skiplist（1）这篇博客中又类似用法。key_data位于结构体的最后面，可在申请内存时，申请足够多的空间。
查询
插入
删除
排序
链表
O(n)
O(1)
O(1)
支持
哈希表
O(1)
O(1)
O(1)
不支持
注1：哈希表实现O(1)操作的前提是：平均每哈希桶元素数 <= 1
注2：为了保持平均哈希桶元素数，必要时必须Resize，而Resize后，原有顺序将被打破
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&);
};
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(value));
} else {
return Slice(key_data, key_length);
}
}
}; 关于char key_data[1]，在LevelDB源码分析之五：skiplist（1）这篇博客中又类似用法。key_data位于结构体的最后面，可在申请内存时，申请足够多的空间。
往下面看会看到这句：
LRUHandle* e = reinterpret_cast(
malloc(sizeof(LRUHandle)-1 + key.size())); 三.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;
}
};四.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(e);
}
// 释放结点
void LRUCache::Release(Cache::Handle* handle) {
MutexLock l(&mutex_);
Unref(reinterpret_cast(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(
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(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);
}
} 在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)的查询，利用链表维持对缓存记录按访问时间排序
注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(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(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
下载文档
