LevelDB：Cache源码精读——缓存

一、原理分析：

这里讲的Cache缓存是指内存缓存，既然是内存缓存，因为内存有限，所以缓存肯定有一个容量大小capacity。

1、模拟实例化一个缓存时，LevelDB的Cache对象结构。

1.1、LevelDB可以创建一个容量大小capacity 的Cache，

1.2、Cache子类ShardedLRUCache将容量大小capacity的缓存分成了很多个小缓存LRUCache。

1.3、小缓存LRUCache里实现了一个双向链表lru_和一个二级指针数组table_用来缓存数据。双向链表lru_用来保证当缓存容量饱满时，清除比较旧的缓存数据；二级指针数组table_用来保证缓存数据的快速查找。

2、模拟缓存一个数据时，LevelDB的Cache工作流程。

2.1、调用Cache的insert方法插入缓存数据data，子类ShardedLRUCache将缓存数据data进行hash操作，得到的hash值定位得到属于哪个小缓存LRUCache，LRUCache将缓存数据封装成LRUHandle数据对象，进行存储。

2.2、先将缓存数据添加到双向链表lru_中，由于lru_.pre存储比较新的数据，lru_.next存储比较旧的数据，所以将缓存数据添加在lru_.pre上。

2.3、再存储到二级指针数组table_里，存储之前，先查找数据是否存在。查找根据缓存数据的hash值，定位缓存数据属于哪个一级指针，然后遍历这一级指针上存放的二级指针链表，查找缓存数据。

2.4、最后如果缓存数据的总大小大于缓存LRUCache的容量大小，则循环从双向链表lru_的next取缓存数据，将其从双向链表lru_和二级指针数组table_中移除，直到缓存数据的总大小小于缓存LRUCache的容量大小。

二、代码实现：

1、创建一个容量大小capacity 的Cache

/*****************************************************************************       类：Cache  *****************************************************************************/            Cache* NewLRUCache(size_t capacity)      {          return new ShardedLRUCache(capacity);      }

2、将容量大小capacity的缓存分成了很多个小缓存LRUCache；将缓存数据data进行hash操作，得到的hash值定位得到属于哪个小缓存LRUCache。

/*****************************************************************************           类：ShardedLRUCache  *****************************************************************************/                    static const int kNumShardBits = 4;          static const int kNumShards = 1 << kNumShardBits; // 2^4==16                    class ShardedLRUCache : public Cache          {          private:              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);              }                            // 得到shard_数组的下标              static uint32_t Shard(uint32_t hash)              {                  /*                   hash是4个字节，32位，向右移动28位，则剩下高4位有效位，                   即最小的是0000等于0，最大的是1111等于15                    则得到的数字在[0,15]范围内。                   */                  return hash >> (32 - kNumShardBits);              }                        public:              explicit ShardedLRUCache(size_t capacity) : last_id_(0)              {                  /*                   将容量平均分成kNumShards份，如果有剩余，将剩余的补全。为什么要补全呢？                   例如设置容量大小为10，则最多就能放下大小为10的数据，现在将容量分成3份，                   如果不补全，余量被丢弃，每份容量则为3，总容量为9，需要放大小为10的数据则放不下了。                   如果补全，剩余量1加上2，每份就多得1个容量，也就每份容量为4，总容量为12，能保证数据都放下。                   */                  /*                   //补全块，                   如果capacity除以kNumShards有余数，那么余数加上(kNumShards - 1)，                   除以kNumShards，就能多得到一块。                   如果如果capacity除以kNumShards无余数，那么0加上(kNumShards - 1)，                   除以kNumShards，还是0                   */                  const size_t per_shard = (capacity + (kNumShards - 1)) / kNumShards;                  for (int s = 0; s < kNumShards; s++)                  {                      shard_[s].SetCapacity(per_shard);                  }              }              virtual ~ShardedLRUCache() { }              // charge 数据大小              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);              }              virtual void* Value(Handle* handle)              {                  return reinterpret_cast<LRUHandle*>(handle)->value;              }              virtual uint64_t NewId()              {                  MutexLock l(&id_mutex_);                  return ++(last_id_);              }          };                }

3、小缓存LRUCache里实现了一个双向链表lru_和一个二级指针数组table_用来缓存数据。双向链表lru_用来保证当缓存容量饱满时，清除比较旧的缓存数据；二级指针数组table_用来保证缓存数据的快速查找；lru_.pre存储比较新的数据，lru_.next存储比较旧的数据；缓存数据的总大小大于缓存LRUCache的容量大小，则循环从双向链表lru_的next取缓存数据，将其从双向链表lru_和二级指针数组table_中移除，直到缓存数据的总大小小于缓存LRUCache的容量大小。

/*****************************************************************************           类：LRUCache  *****************************************************************************/                    // A single shard of sharded cache.          class LRUCache          {          public:              LRUCache();              ~LRUCache();                            // Separate from constructor so caller can easily make an array of LRUCache              void SetCapacity(size_t capacity) { capacity_ = capacity; }                            // Like Cache methods, but with an extra "hash" parameter.              Cache::Handle* Insert(const Slice& key, uint32_t hash, void* value,                                    size_t charge, void (*deleter)(const Slice& key, void* value));              Cache::Handle* Lookup(const Slice& key, uint32_t hash);              void Release(Cache::Handle* handle);              void Erase(const Slice& key, uint32_t hash);                        private:              void LRU_Remove(LRUHandle* e);              void LRU_Append(LRUHandle* e);              void Unref(LRUHandle* e);                            // Initialized before use.              // 缓存的总容量              size_t capacity_;                            // mutex_ protects the following state.              port::Mutex mutex_;              // 缓存数据的总大小              size_t usage_;                            // Dummy head of LRU list.              // lru.prev is newest entry, lru.next is oldest entry.              // 双向循环链表，有大小限制，保证数据的新旧，当缓存不够时，保证先清除旧的数据              LRUHandle lru_;              /*                二级指针数组，链表没有大小限制，动态扩展大小，保证数据快速查找，               hash定位一级指针，得到存放在一级指针上的二级指针链表，遍历查找数据               */              HandleTable table_;          };                    LRUCache::LRUCache(): usage_(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;                  assert(e->refs == 1);  // Error if caller has an unreleased handle                  Unref(e);                  e = next;              }          }                    void LRUCache::Unref(LRUHandle* e)          {              assert(e->refs > 0);              e->refs--;              if (e->refs <= 0) // 引用计数小于等于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_              // 新数据插到lru_的前面              e->next = &lru_;              e->prev = lru_.prev;              e->prev->next = e;              e->next->prev = e;          }                    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_的next处的数据，保证清除比较旧的数据。                   */                  LRU_Remove(e);                  LRU_Append(e);              }              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_);                            // 减去记录key的首地址大小(一个字节)，加上key实际大小              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;              e->refs = 2;  // One from LRUCache, one for the returned handle              // 记录key的首地址              memcpy(e->key_data, key.data(), key.size());              LRU_Append(e);              // 缓存数据的大小              usage_ += charge;                            LRUHandle* old = table_.Insert(e);              if (old != NULL)              {                  LRU_Remove(old);                  Unref(old);              }                            // 缓存不够，清除比较旧的数据              while (usage_ > capacity_ && lru_.next != &lru_)              {                  LRUHandle* old = lru_.next;                  LRU_Remove(old);                  table_.Remove(old->key(), old->hash);                  Unref(old);              }                            return reinterpret_cast<Cache::Handle*>(e);          }                    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);              }          }

4、根据缓存数据的hash值，定位缓存数据属于哪个一级指针，然后遍历这一级指针上存放的二级指针链表，查找缓存数据。

/*****************************************************************************           类：HandleTable  *****************************************************************************/                    // We provide our own simple hash table since it removes a whole bunch          // of porting hacks and is also faster than some of the built-in hash          // table implementations in some of the compiler/runtime combinations          // we have tested.  E.g., readrandom speeds up by ~5% over the g++          // 4.4.3's builtin hashtable.          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);                  *ptr = h;                  // 找到的节点的值为NULL，说明h是新节点                  if (old == NULL)                  {                      // 元素个数                      ++elems_;                      // 元素个数加1大于一级指针个数。如果每个节点h定位一级指针不存在哈希冲突，则每个一级指针存放一个节点                      if (elems_ > length_)                      {                          // Since each cache entry is fairly large, we aim for a small                          // average linked list length (<= 1).                          Resize();                      }                  }                  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.              LRUHandle** FindPointer(const Slice& key, uint32_t hash)              {                  /*                    hash & (length_ - 1)的运算结果是0到length-1;                   */                  LRUHandle** ptr = &list_[hash & (length_ - 1)];                  // 二级指针链表*ptr不为空，遍历二级指针链表找到hash相同且key也相同的节点                  while (*ptr != NULL && ((*ptr)->hash != hash || key != (*ptr)->key()))                  {                      ptr = &(*ptr)->next_hash;                  }                  // 返回匹配节点的地址                  return ptr;              }                            void Resize()              {                  uint32_t new_length = 4;                  while (new_length < elems_)                  {                      new_length *= 2;                  }                  // 下面的new方法，只表明给一级指针分配了内存块                  LRUHandle** new_list = new LRUHandle*[new_length];                  /*                   避免一级指针分配的内存块，存有野指针，所以需要使用memset对内存块进行清零处理。                   memset:作用是在一段内存块中存储某个给定的值，                          它对较大的结构体或数组进行清零操作的一种最快方法。                          存储0，就是置空。                   new_list和&new_list[i]是一级指针，                   *new_list和new_list[i]是二级指针，                   **new_list是二级指针存储的值。                   下面memset代码的意思是：                   即将一级指针内存块中存储0，就是new_list[i] = 0或new_list[i] = NULL;                   也就是将二级指针*new_list置空。                   */                  memset(new_list, 0, sizeof(new_list[0]) * new_length);                  uint32_t count = 0;                  for (uint32_t i = 0; i < length_; i++) // 遍历一级指针                  {                      /*                       由于每个h通过表达式hash&(new_length - 1)得到属于一级指针的位置，                       所以表达式计算结果相同（注：hash不相同，计算结果也可能相同）的h,会定位到相同的一级指针，                       并组成一个二级链表存放在一级指针上。                       一级指针上存放的二级指针链表，通过h的next_hash链接起来                       */                      LRUHandle* h = list_[i];                      while (h != NULL)                      {                          /*                           功能：下面遍历的逻辑是重新定位h属于的一级指针。并在新的一级指针上组成新的二级链表。                           */                          LRUHandle* next = h->next_hash;                          uint32_t hash = h->hash;                          // 定位新的一级指针 *ptr就是new_list[hash & (new_length - 1)]                          LRUHandle** ptr = &new_list[hash & (new_length - 1)];                          // 如果是第一次运行，则*ptr为NULL，其他则是取到上个循环h的地址                          h->next_hash = *ptr;                          // new_list[hash & (new_length - 1)] = h;                          *ptr = h;                          // 二级链表下一个节点                          h = next;                          count++;                      }                  }                  assert(elems_ == count);                  delete[] list_;                  list_ = new_list;                  length_ = new_length;              }          };

5、缓存数据封装成LRUHandle数据对象，进行存储；双向链表也是LRUHandle数据对象，使用了next和prev字段；二级指针数组中的二级指针链表也是LRUHandle数据对象，使用了next_hash字段。

        struct LRUHandle          {              void* value;              void (*deleter)(const Slice&, void* value);              LRUHandle* next_hash;  // 二级指针数组的二级指针链表              LRUHandle* next;       // 双向链表 指向比较旧的数据              LRUHandle* prev;       // 双向链表 指向比较新的数据              size_t charge;      // TODO(opt): Only allow uint32_t?              size_t key_length;              uint32_t refs;              uint32_t hash;      // Hash of key(); used for fast sharding and comparisons              char key_data[1];   // Beginning of key 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);                  }              }          };