[TOC]

参考

leveldb中的LRUCache设计

LevelDB-LruCache源码学习

1. 前言

leveldb是Google大神Jeff Dean的作品，只有2万行出头的代码量，非常精彩，非常值得解剖学习。本文介绍leveldb的LRUCache设计。 我们知道，相对于块设备，内存永远都是奢侈品。局部性原理应该算是计算机科学中数一数二的重要原理，无数精巧的设计，不过是让有限的内存提供更高的用户数据访问命中，茫茫海量的数据，我需要你的时候，你恰巧在内存，这才是工程师不不懈的追求。

内存是稀缺资源，缓存替换算法是计算机科学的重要Topic。LRU （Last Recent Used）是一个重要的缓存替换算法。Leveldb采用的就是这种算法。但是仅仅知道LRU这三个字母是没啥用的，我们一起学习下leveldb的缓存替换算法，体会下大神的精巧设计。

2. 内部实现

没有文档的情况下，沉入代码细节，容易只见树木不见森林，需要花费好久才能体会出作者的设计思路。代码告诉你How，而设计文档才告诉你Why。

LRUCache设计有4个数据结构，是依次递进的关系，分别是：

• LRUHandle
• HandleTable
• LRUCache
• ShardedLRUCache

事实上到了第三个数据结构LRUCache，LRU的缓存管理数据结构已经实现了，之所以引入第四个数据结构，就是因为减少竞争。因为多线程访问需要加锁，为了减少竞争，提升效率，ShardedLRUCache内部有16个LRUCache，查找key的时候，先计算属于哪一个LRUCache，然后在相应的LRUCache中上锁查找。

class ShardedLRUCache : public Cache {  
 private:  
  LRUCache shard_[kNumShards];  
  ...
}

这不是什么高深的思路，这种减少竞争的策略非常常见。因此，读懂缓存管理策略的关键在前三个数据结构。

LevelDB的Cache管理，维护有2个双向链表和一个哈希表。哈希表是非常容易理解的。如何确定一个key值到底存不存在，如果存在如何快速获取key值对应的value值。我们都学过数据结构，这活，哈希表是比较适合的。

注意，我们都知道，hash表存在一个重要的问题，就是碰撞，有可能多个不同的键值hash之后值相同，解决碰撞的一个重要思路是链表，将hash之后计算的key相同的元素链入同一个表头对应的链表。

可是我们并不满意这种速度，LevelDB做了进一步的优化，即及时扩大hash桶的个数，尽可能地不会发生碰撞。因此LevelDB自己实现了一个hash表，即HandleTable数据结构。

说句题外话，我不太喜欢数据结构的命名方式，比如HandleTable，命名就是个HashTable，如果出现Hash会好理解很多。这个名字还自罢了，LRUHandle这个名字更是让人摸不到头脑，明明就是一个数据节点，如果名字中出现Node，整个代码都会好理解很多。好了吐槽结束，看下HandleTable的数据结构：

class HandleTable {
 public:
  
 ...
 
 private:

  uint32_t length_;
  uint32_t elems_;
  LRUHandle** list_;
};

第一个元素length_ 纪录的就是当前hash桶的个数，第二个元素 elems_维护在整个hash表中一共存放了多少个元素。第三个就更好理解了，二维指针，每一个指针指向一个桶的表头位置。

为了提升查找效率，提早增加桶的个数来尽可能地保持一个桶后面只有一个元素，在插入的算法中有如下内容：

  LRUHandle* Insert(LRUHandle* h) {
    //这里特别注意的是返回二级指针的原因。
    //list_[hash & (length_ - 1)]是一个分配在堆上的指针
    //情况1、 list_[hash & (length_ - 1)]==NULL时，新添加的节点需添加在其后，
    //因此list_[hash & (length_ - 1)]= 新添加的节点内存地址
    //情况2、list_[hash & (length_ - 1)]!=NULL时，通常链表插入时需修改前一个
    //节点的next_hash,因此需要前一个节点的地址，函数返回为二级指针，
    //就是前一个节点next_hash指针的自身地址因此一行 *ptr = h;就完成了旧指针的踢出
    //及新节点的加入
    LRUHandle** ptr = FindPointer(h->key(), h->hash);
    LRUHandle* old = *ptr;  //老的元素返回，LRUCache会将相同key的老元素释放，详情看LRUCache的Insert函数。
    h->next_hash = (old == NULL ? NULL : old->next_hash);
    *ptr = h;
    if (old == NULL) {
      ++elems_;
      if (elems_ > length_) {
        // Since each cache entry is fairly large, we aim for a small
        // average linked list length (<= 1).
        Resize();
      }
    }
    return old;
  }

注意，当整个hash表中元素的个数超过 hash表桶的的个数的时候，调用Resize函数，该函数会将桶的个数增加一倍，同时将现有的元素搬迁到合适的桶的后面。正是这种提早扩大桶的个数，良好的hash函数会保证每个桶对应的链表中尽可能的只有1个元素，从这个角度讲，LevelDB使用这种优化后的哈希表，查找的效率为O（1）。

  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;
    for (uint32_t i = 0; i < length_; i++) {
      LRUHandle* h = list_[i];
      while (h != NULL) {
        LRUHandle* next = h->next_hash;
        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;
  }

另外有个很巧妙的地方，FindPointer 是返回的二级指针，以便获取老的节点。

  //list_[hash & (length_ - 1)]是一个分配在堆上的指针
  //情况1、 list_[hash & (length_ - 1)]==NULL时，新添加的节点需添加在其后，
  //因此list_[hash & (length_ - 1)]= 新添加的节点内存地址
  //情况2、list_[hash & (length_ - 1)]!=NULL时，通常链表插入时需修改前一个
  //节点的next_hash,因此需要前一个节点的地址，函数返回为二级指针，
  //就是前一个节点next_hash指针的自身地址因此一行 *ptr = h;就完成了旧指针的踢出
  //及新节点的加入
  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;
  }

设计缓存，快速找到位置是一个重要指标，但是毫无疑问，不仅仅只有这一个设计指标。因为使用LRU，当空间不够的时候，需要踢出某些元素的时候，必需能够快速地找到，哪些元素Last Recent Used，作为替换的牺牲品。这个指标哈希表可就爱莫能助了。

当然，我们可以讲最近访问时间作为元素的一个字段保存起来，但是我们不得不扫描整个hash表，将访问时间排序才能知道哪个元素更应该被剔除。毫无疑问效率太低。

LRUCache不仅需要哈希表来快速查找，还需要链表能够快速插入和删除。LRUCache 维护有两条双向链表：

  LRUHandle lru_;      // lru_ 是冷链表，属于冷宫，

  LRUHandle in_use_;   // in_use_ 属于热链表，热数据在此链表

  HandleTable table_; // 哈希表部分已经讲过

2.1. LruCache的存储结构

此结构既用于双向链表也用于hash表。

struct LRUHandle {
  void* value;//缓存的内容
  void (*deleter)(const Slice&, void* value);//回调函数，自定义清理value、key
  LRUHandle* next_hash;//在HandleTable LRUCache::table_中当hash冲突时指向后续LRUHandle
  LRUHandle* next;//在LRUHandle LRUCache::lru_(双向链表)中指向后驱节点
  LRUHandle* prev;//在LRUHandle LRUCache::lru_(双向链表)中指向前驱节点
  size_t charge; // TODO(opt): Only allow uint32_t?
  size_t key_length;//key字节数
  uint32_t refs;//引用计数
  uint32_t hash; // Hash of key(); used for fast sharding and comparisons
  uint64_t timestamp;//改造lurcache支持时间淘汰
  char key_data[1]; // Beginning of 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);
    }
  }
};

2.2. 哈希表--HandleTable实现

核心成员
LRUHandle** list_; 即使用拉链法实现哈希表。

源码分析

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) {
    //判断h指向的内容在table中是不是已存在，存在返回，否则返回NULL。
    LRUHandle** ptr = FindPointer(h->key(), h->hash);
    LRUHandle* old = *ptr;
    h->next_hash = (old == NULL ? NULL : old->next_hash);
    //这里特别注意的是返回二级指针的原因。
    //list_[hash & (length_ - 1)]是一个分配在堆上的指针
    //情况1、 list_[hash & (length_ - 1)]==NULL时，新添加的节点需添加在其后，
    //因此list_[hash & (length_ - 1)]= 新添加的节点内存地址
    //情况2、list_[hash & (length_ - 1)]!=NULL时，通常链表插入时需修改前一个
    //节点的next_hash,因此需要前一个节点的地址，函数返回为二级指针，
    //就是前一个节点next_hash指针的自身地址因此一行 *ptr = h;就完成了旧指针的踢出
    //及新节点的加入
    *ptr = h;
    if (old == NULL) {
      ++elems_;
      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;
  }

  uint32_t Size() {
    return elems_;
  }

 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_;

  //返回二级指针很有技巧性
  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;
  }

  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;
    for (uint32_t i = 0; i < length_; i++) {
      LRUHandle* h = list_[i];
      while (h != NULL) {
        LRUHandle* next = h->next_hash;
        uint32_t hash = h->hash;
        //采用链表的表头插入法
        LRUHandle** ptr = &new_list[hash & (new_length - 1)];
        h->next_hash = *ptr;
        //h做为新表头节点
        *ptr = h;
        h = next;
        count++;
      }
    }
    assert(elems_ == count);
    delete[] list_;
    list_ = new_list;
    length_ = new_length;
  }
};

2.3. LRUCache

2.3.1. 核心结构

// Initialized before use.
  size_t capacity_;//缓存容量已字节为单位
  // mutex_ protects the following state.
  std::mutex mutex_;//锁
  size_t usage_;//当前缓存使用量，以字节为单位
  // Dummy head of LRU list.
  // lru.prev is newest entry, lru.next is oldest entry.
  LRUHandle lru_; //双向链表，lru_为表头，插入新节点时采用表头法
  HandleTable table_;//hash表

2.3.2. 源码分析

void LRUCache::LRU_Append(LRUHandle* e) {
  // 采用表头法插入新节点
  e->next = &lru_;
  e->prev = lru_.prev;
  e->prev->next = e;
  e->next->prev = e;
}
Cache::Handle* LRUCache::Insert(
    const Slice& key, uint32_t hash, void* value, size_t charge,
    void (*deleter)(const Slice& key, void* value)) {

  std::lock_guard<std::mutex> lock(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;
  e->timestamp = base::TimeUtil::CurrentTimeInSec();
  e->refs = 2; // One from LRUCache, one for the returned handle
  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);
  }
  //缓存淘汰。当到达容量上限后，淘汰访问最不频繁的节点即表头lru_的next。
  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);
}

2.3.2. hash分片的设置

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

  static uint32_t Shard(uint32_t hash) {
    return hash >> (32 - kNumShardBits);
  }

hash分片数是2的n次方的原因是将常规的%运算转化为位运算实现，因为位运算效率更高，差不多是%的3倍。