鸿 网 互 联 www.68idc.cn

当前位置 : 主页 > 数据库 > mysql > >

LevelDB : LRU Cache

来源:互联网 作者:佚名 时间:2016-06-21 11:35
关于LRU Cache 1. http://blog.csdn.net/huntinux/article/details/39290833 2. http://www.cnblogs.com/liuhao/archive/2012/11/29/2795455.html 3. http://blog.itpub.net/26239116/viewspace-1842049/ (重点参考) 4. http://mingxinglai.com/cn/2013/01

关于LRU Cache
1. http://blog.csdn.net/huntinux/article/details/39290833
2. http://www.cnblogs.com/liuhao/archive/2012/11/29/2795455.html
3. http://blog.itpub.net/26239116/viewspace-1842049/ (重点参考)
4. http://mingxinglai.com/cn/2013/01/leveldb-cache/ (重要参考,建议移步这里)

引文1中给出了LRU cache 的一种实现方式是: 双向链表 + hashtable。由于cache中的移动操作频繁,因此使用双向链表。为了弥补双向链表查找性能缺陷,引入hashtable。读者可以理解了引文1中的简单实现再继续往下看。

LRUHandle

用来表示哈希表中的元素

// An entry is a variable length heap-allocated structure.  Entries
// are kept in a circular doubly linked list ordered by access time.
struct LRUHandle {
  void* value;
  void (*deleter)(const Slice&, void* value);
  LRUHandle* next_hash; // 作为Hash表中的节点,指向hash值相同的节点(解决hash冲突采用链地址法)
  LRUHandle* next; // 作为Cache中的节点,指向后继
  LRUHandle* prev; // 作为Cache中的节点,指向前驱
  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
  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);
    }
  }
};

其中比较有意思的是char key_data[1];,会有这样的疑问:为什么不直接定义成指针呢?
定义成指针将会占用4字节(32位机器)或8字节(64位机器)的空间,而这样定义只占用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还是指向这块空间的。看来key_data[1]只是一个占位符。(个人理解)

哈希表 HandleTable

bucket:桶,哈希表当中有若干bucket,而每个bucket是一个链表,用来存放hash值相同的元素。(所以这里使用的解决“hash冲突”的方法是链表法)

HandleTable中:
length_ : hash表桶(bucket)的数量
elems_ : 放入哈希表中元素的个数
当元素个数大于桶的个数时(elems_ > length_ ),就重新hash(rehash),这样hash表平均查找效率为O(1)。(可以与Redis中哈希表的rehash对比一下。)
list_ : 是一个数组,数组中的每个元素是一个指针,构成链表存储hash值相同的元素。在这里,每个元素的类型是 LRUHandle *

构造函数

构造函数的初始化列表将桶个数和元素个数设置为0,然后在函数体中进行Resize(),Resize函数对哈希表大小进行调整,然后对所有元素进行rehash。经过resize,哈希表的初始大小为4。

HandleTable() : length_(0), elems_(0), list_(NULL) { Resize(); // 初始大小length_为4}

查找

查找一个元素是否在哈希表中。首先根据该元素的hash值定位到某个hash bucket(是一个链表),然后在链表中顺序查找。最后,如果找到了就返回一个指向该元素的指针。否则返回该桶的最后一个位置的指针。

注意该函数的返回值是一个二级指针,调用这可以使用该指针对其进行修改。

此外,hash值是如何与数组下标联系起来的呢? 通过 hash & (length -1) ,length为数组大小,并且length是4的倍数(见Resize函数)那么length-1相当于一个mask,与hash做与操作就计算出了该元素在哪个桶了。

  // 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定位到某个桶
    LRUHandle** ptr = &list_[hash & (length_ - 1)];
    // 在链表中顺序查找(比对key)
    while (*ptr != NULL &&
           ((*ptr)->hash != hash || key != (*ptr)->key())) {
      ptr = &(*ptr)->next_hash;
    }
    // 返回查找结果
    return ptr;
  }

插入操作

这是将一个元素插入到哈希表的接口。

  LRUHandle* Insert(LRUHandle* h) {
    // 在链表中查找是否有该元素
    LRUHandle** ptr = FindPointer(h->key(), h->hash);
    LRUHandle* old = *ptr;
    // old==NULL表示没有找到,执行插入操作
    // 否则表示找到了相同元素,那么也要用新的代替旧的
    h->next_hash = (old == NULL ? NULL : old->next_hash);
    *ptr = h;
    if (old == NULL) {
      ++elems_; // 元素个数加1
      if (elems_ > length_) {
        // 必要时,调整哈希表大小
        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;
  }

Resize (Rehash)

该函数保证桶的个数大于元素的个数。

  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; // 让list_指向新的桶数组
    length_ = new_length; // 更新length_
  }
};

LRUCache

一个LRUCache的实现,成员变量介绍:

  // Initialized before use.
  size_t capacity_; // cache的容量

  // mutex_ protects the following state.
  mutable port::Mutex mutex_;
  size_t usage_; // cache已经使用的容量

  // Dummy head of LRU list.
  // lru.prev is newest entry, lru.next is oldest entry.
  LRUHandle lru_; // LRU链表(双向循环链表),按照访问先后进行排序,表头的prev是最近访问的

  HandleTable table_; // 存放节点的哈希表,用于快读查找

可以看到,LRUCache就是是通过 双向链表 + hashtable 实现的(理由在最上面)。

查找

查找操作就是调用前面介绍的哈希表的查找函数。

Cache::Handle* LRUCache::Lookup(const Slice& key, uint32_t hash) {
  MutexLock l(&mutex_);
  // 在哈希表中查找
  LRUHandle* e = table_.Lookup(key, hash);
  if (e != NULL) {
    e->refs++; // 增加引用计数
    // 从cache中移动到最前面(现remove再append)
    LRU_Remove(e); 
    LRU_Append(e);
  }
  return reinterpret_cast<Cache::Handle*>(e);
}

// remove和append只是关于双向链表的操作,比较简单
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;
}

// 引用计数减一。引用计数变为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);
  }
} 

插入

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、hash、value等创建一个新的元素,将被插入到cache中
  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
  memcpy(e->key_data, key.data(), key.size());
  LRU_Append(e);
  usage_ += charge;

  // 哈希表的Insert函数在插入时如果发现有相同元素,则将旧的返回,将新的替换旧的
  // 然后将旧的进行释放
  LRUHandle* old = table_.Insert(e);
  if (old != NULL) {
    LRU_Remove(old);
    Unref(old);
  }

  // 当cache满了,需要移除oldest的元素
  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);
}

设置cache的容量

  // Separate from constructor so caller can easily make an array of LRUCache
  void SetCapacity(size_t capacity) { capacity_ = capacity; }

此外,Erase将清空cache,而Prune会移除引用计数为1的元素(即外部没有使用)

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);
  }
}

void LRUCache::Prune() {
  MutexLock l(&mutex_);
  for (LRUHandle* e = lru_.next; e != &lru_; ) {
    LRUHandle* next = e->next;
    if (e->refs == 1) {
      table_.Remove(e->key(), e->hash);
      LRU_Remove(e);
      Unref(e);
    }
    e = next;
  }
}

SharedLRUCache

不错的分析:http://mingxinglai.com/cn/2013/01/leveldb-cache/

SharedLRUCache中有一个LRUCache的数组,SharedLRUCache做的工作就是计算出元素的hash值,然后根据hash值的高4位确定使用哪一个LRUCache,这么做的理由(摘自上面的引文):

这是因为levelDB是多线程的,每个线程访问缓冲区的时候都会将缓冲区锁住,为了多线程访问,尽可能快速,减少锁开销,ShardedLRUCache内部有16个LRUCache,查找Key时首先计算key属于哪一个分片,分片的计算方法是取32位hash值的高4位,然后在相应的LRUCache中进行查找,这样就大大减少了多线程的访问锁的开销。

最后,引用引文4中的图作为总结。
这里写图片描述

网友评论