• 技术贴 | Rocksdb 中 Memtable 源码解析

     

    一、什么是 Memtable？

    Memtable 是 Rocksdb 在内存中保存数据的一种数据结构，一个 Memtable 的容量是固定的，在 Memtable 写满后，会转换为  Immutable Memtable，Immutable Memtable 中的数据会 Flush 到 SST File 中。

    Memtable 和 Immutable Memtable 的唯一区别是 Memtable 可读可写，而 Immutable Memtable 是只读且不允许写入。Rocksdb 引入了 Column Family 的概念，在一个 Column Family 中只有一个 Memtable，但允许存在多个 Immutable Memtable。Rocksdb 支持创建多数据结构类型的 Memtable，默认的是 SkipList，即跳跃表。

    二、Memtable 的数据结构

    Rocksdb 中 Memtable 有多种实现方式 (SkipList / HashSkipList / HashLinkList / Vector)，其中默认的实现方式为 SkipList。

    一个 Memtable 中维护了两个 SkipList，其中范围删除插入 range_del_table_，其余的操作写入 table_。

    Memtable 定义的操作接口 Add () 如下：

    1.  
    2. bool MemTable::Add(SequenceNumber s, ValueType type,
    3.                    const Slice& key, /* user key */
    4.                    const Slice& value, bool allow_concurrent,
    5.                    MemTablePostProcessInfo* post_process_info) {
    6.   // 一条key-value Entry的数据格式
    7.   //  key_size     : varint32 of internal_key.size()
    8.   //  key bytes    : char[internal_key.size()]
    9.   //  value_size   : varint32 of value.size()
    10.   //  value bytes  : char[value.size()]
    11.   uint32_t key_size = static_cast<uint32_t>(key.size());
    12.   uint32_t val_size = static_cast<uint32_t>(value.size());
    13.   uint32_t internal_key_size = key_size + 8;
    14.   const uint32_t encoded_len = VarintLength(internal_key_size) +
    15.                                internal_key_size + VarintLength(val_size) +
    16.                                val_size;
    17.   char* buf = nullptr;
    18.   // 通过判断key-value的类型来选择memtable, 范围删除的kv插入range_del_table_
    19.   std::unique_ptr<MemTableRep>& table =
    20.       type == kTypeRangeDeletion ? range_del_table_ : table_;
    21.   KeyHandle handle = table->Allocate(encoded_len, &buf);
    22.   //...
    23.   // 是否允许并发插入
    24.   if (!allow_concurrent) {
    25.     // 是否制定了函数提取key的前缀
    26.     if (insert_with_hint_prefix_extractor_ != nullptr &&
    27.         insert_with_hint_prefix_extractor_->InDomain(key_slice)) {
    28.       // ...
    29.       bool res = table->InsertWithHint(handle, &insert_hints_[prefix]);
    30.     } else {
    31.       // 插入key-value pair
    32.       bool res = table->Insert(handle);
    33.       if (UNLIKELY(!res)) {
    34.         return res;
    35.       }
    36.     }
    37.   } else {
    38.     // 插入key-value pair
    39. bool res = table->InsertConcurrently(handle);
    40.     if (UNLIKELY(!res)) {
    41.       return res;
    42.     }
    43.   }
    44.   return true;
    45. }

    Add () 函数将用户的 key 和 value 封装成一个 buf，然后根据不同的条件调用 table->Insert () 插入至 Memtable。table 就是 Memtable 的工厂类实现，默认 SkiplistRep, 即通过调用 SkipList 的 Insert () 完成 key 的插入。

    Memtable 定义的操作接口  Get () 如下：

    1.  
    2. bool MemTable::Get(const LookupKey& key, std::string* value, Status* s,
    3.                    MergeContext* merge_context,
    4.                    RangeDelAggregator* range_del_agg, SequenceNumber* seq,
    5.                    const ReadOptions& read_opts, ReadCallback* callback,
    6.                    bool* is_blob_index) {
    7.   // 在range_del_table_上初始化一个迭代器
    8.   std::unique_ptr<InternalIterator> range_del_iter(
    9.       NewRangeTombstoneIterator(read_opts));
    10.   Status status = range_del_agg->AddTombstones(std::move(range_del_iter));
    11.   if (!status.ok()) {
    12.     *s = status;
    13.     return false;
    14.   }
    15.  
    16.   Slice user_key = key.user_key();
    17.   // 利用前缀提取过滤判断key是否存在
    18.   bool const may_contain =
    19.       nullptr == prefix_bloom_

    Memtable 的 Get () 调用了 SkipListRep 的 Get () 接口，最终是通过 SkipList 的 FindGreaterOrEqual () 来查找。查找出来的 key 会被传入的回调函数 SaveValu 并 e () 根据 type 处理，例如 ktypeDeletion 就返回 NotFound ()。

    三、什么是 SkipList？

    SkipList 即跳跃表，在普通单向链表的基础上增加了一些索引，而且这些索引是分层的，从而可以快速地查到数据。如下是一个典型的跳跃表构建过程：

    初始我们有个带头结点的有序链表 a，而后每相邻两个节点增加一个指针，让指针指向下下个节点，得到表 b。这样所有新增指针连成了一个新的链表，但它包含的节点个数只有原来的一半。其后我们对第二层链表再次进行此操作，得到表 c。重复这个过程，直到采样出的节点只剩一个，如图 d。这样便完成了跳跃表的构建。跳跃表查找过程如下：

    从 head 开始，head 的 level 为 4，判断 head 后继节点值小于 <12，此时当前节点变为 6，继续查找；节点 6 的 level 为 3，判断后继节点值为 NIL，因此 level 降低到 2；判断 x -> forward [2] -> key (25) > 17，继续降低 level 到 1；判断 x -> forward [1] -> key (9) < 17，此时 x 变为 x ->forward [1]，x 成为节点 9；节点 9 的判断 x -> forward [1] -> key 为 17，因此找到节点，直接返回。

    跳跃表插入过程如下：

    我们以上图为例，list -> leve=4，如果要插入节点 17，首先确定搜索路径，与之前步骤类似。

    创建新节点 Node (17)，并为其生成 level (随机)，该 level 可能值为 [1, MaxLevel]，此时需要对比，如果 level < list -> level，需要先将突出部分从 header 指向它，这里新生成的节点 Node (17) 的 level  为 5，超过了 list 当前的最大 level，于是将 update [4] 设置为 header，后续直接将 Node (17) 作为 header 的后继。

    最后是设置搜索路径上每个节点的后继关系，这样我们便完成了节点的插入。我们来看一下 SkipList 的具体代码实现：

    InlineSkipList 数据结构 >>

    1.  
    2. class InlineSkipList {
    3.  private:
    4.   struct Node;
    5.   struct Splice;
    6.  
    7.  public:
    8.   using DecodedKey = \
    9.     typename std::remove_reference<Comparator>::type::DecodedType;
    10. …
    11.   Allocator* const allocator_; 
    12.   Comparator const compare_;
    13.   Node* const head_;
    15.   std::atomic<int> max_height_;  
    16.  
    17.   Splice* seq_splice_;
    18. };

    Node 的数据结构 >>

    1.  
    2. template <class Comparator>
    3. struct InlineSkipList<Comparator>::Node {
    4.  
    5.  private:
    6.   // 存放该节点的next_节点的数组
    7.   // 数组大小为该节点的height，当调用NewNode()分配内存初始化整个数组
    8.   std::atomic<Node*> next_[1];
    9. };

    Node 的数据结构如图，它将 key 和链表每层的指针连续存储，通过 next_[-n] 这种方式来访问每层的 next 指针，此外在 new 新节点时会把该节点高度写在 next_[0] 的前 4 个字节处，当完成插入后，next_[0] 会恢复成指向同层的下一个节点的指针。

    四、InlineSkipList 插入

    Memtable 的 Add () 通过 SkipList 的 Insert () 来查找，下面是 Insert () 的具体实现：

    1.  
    2. bool InlineSkipList<Comparator>::Insert(const char* key, Splice* splice,
    3.                                         bool allow_partial_splice_fix) {
    4.   Node* x = reinterpret_cast(const_cast(key)) - 1; // x即为next_[0]
    5.   const DecodedKey key_decoded = compare_.decode_key(key);
    6.   int height = x->UnstashHeight();
    7.   assert(height >= 1 && height <= kMaxHeight_);
    8.   int max_height = max_height_.load(std::memory_order_relaxed);
    9.   // 更新max_height
    10.   while (height > max_height) {
    11.     if (max_height_.compare_exchange_weak(max_height, height)) {
    12.       // successfully updated it
    13.       max_height = height;
    14.       break;
    15.     }
    16.     // 否则重试，可能因为其他人增加了它而退出循环
    17.   }
    18.   assert(max_height <= kMaxPossibleHeight);
    19.  
    20.   // 插入节点的时候，需要借助一个Splice对象，该对象主要保存着最近一次插入的节点快照
    21.   // 它保存着一个prev和next的节点指针数组，由Level可以索引到对应Level的节点
    22.   int recompute_height = 0;
    23.   if (splice->height_ < max_height) {
    24.     // 当重置splice
    25.     splice->prev_[max_height] = head_;
    26.     splice->next_[max_height] = nullptr;
    27.     splice->height_ = max_height;
    28.     recompute_height = max_height;
    29.   } else {
    30.     while (recompute_height < max_height) {
    31.       if (splice->prev_[recompute_height]->Next(recompute_height) !=
    32.           splice->next_[recompute_height]) { //判断该层的splice是否紧密，即prev_->Next是否等于next_
    33.         ++recompute_height;
    34.       } else if (splice->prev_[recompute_height] != head_ &&
    35.                  !KeyIsAfterNode(key_decoded,
    36.                                  splice->prev_[recompute_height])) { //小于splice当前层的prev_
    37.         // ...
    38.       } else if (KeyIsAfterNode(key_decoded,
    39.                                 splice->next_[recompute_height])) { //大于splice当前层的prev_
    40.         // ...
    41.       } else {
    42.         // 找到了合适的level
    43.         break;
    44.       }
    45.     }
    46.   }
    47.   assert(recompute_height <= max_height);
    48.   if (recompute_height > 0) {//计算splice
    49.     RecomputeSpliceLevels(key_decoded, splice, recompute_height); // 找到要插入的key合适的splice
    50.   }
    51.  
    52.   bool splice_is_valid = true;
    53.   if (UseCAS) {//CAS无锁机制
    54.     //...
    55.   } else {
    56.     for (int i = 0; i < height; ++i) {
    57.       if (i >= recompute_height &&
    58.           splice->prev_[i]->Next(i) != splice->next_[i]) { // 确保splice此Level有效，如果无效的话再查找一次
    59.         FindSpliceForLevel<false>(key_decoded, splice->prev_[i], nullptr, i,
    60.                                   &splice->prev_[i], &splice->next_[i]);
    61.       }
    62.       // Checking for duplicate keys on the level 0 is sufficient
    63.       if (UNLIKELY(i == 0 && splice->next_[i] != nullptr &&
    64.                    compare_(x->Key(), splice->next_[i]->Key()) >= 0)) {
    65.         // duplicate key
    66.         return false;
    67.       }
    68.       if (UNLIKELY(i == 0 && splice->prev_[i] != head_ &&
    69.                    compare_(splice->prev_[i]->Key(), x->Key()) >= 0)) {
    70.         // duplicate key
    71.         return false;
    72.       }
    73.       //…
    74.       x->NoBarrier_SetNext(i, splice->next_[i]); //将新节点next指向对应的next节点
    75.       splice->prev_[i]->SetNext(i, x); //将splice的prev节点的next指向新节点
    76.     }
    77.   }
    78.   if (splice_is_valid) {//将新节点Height下的prev节点都设置为该节点，因为原先的prev和next之间已经不连续了。
    79.     for (int i = 0; i < height; ++i) {
    80.       splice->prev_[i] = x;
    81.     }
    82.     //...
    83.   } else {
    84.     splice->height_ = 0;
    85.   }
    86.   return true;
    87. }

    五、InlineSkipList 查找

    Memtable 的 Get () 通过 SkipList 的 FindGreaterOrEqual () 来查找，下面是 FindGreaterOrEqual () 的具体实现：

    1.  
    2. InlineSkipList<Comparator>::FindGreaterOrEqual(const char* key) const {
    3.  
    4.   Node* x = head_;
    5.   int level = GetMaxHeight() - 1;//从最高层开始查找
    6.   Node* last_bigger = nullptr;
    7.   const DecodedKey key_decoded = compare_.decode_key(key);
    8.   while (true) {
    9.     Node* next = x->Next(level); 
    10.     if (next != nullptr) {
    11.       PREFETCH(next->Next(level), 0, 1);
    12.     }
    13.     // Make sure the lists are sorted
    14.     assert(x == head_ || next == nullptr || KeyIsAfterNode(next->Key(), x));
    15.     // Make sure we haven't overshot during our search
    16.     assert(x == head_ || KeyIsAfterNode(key_decoded, x));
    17.     int cmp = (next == nullptr || next == last_bigger)
    18.                   ? 1
    19.                   : compare_(next->Key(), key_decoded);
    20.     if (cmp == 0 || (cmp > 0 && level == 0)) { // 找到相等的key或者查找的key不在此范围内
    21.       return next;
    22.     } else if (cmp < 0) { //待查找 key 比 next 大，则在该层继续查找
    23.       x = next;
    24.     } else { // 待查找 key 不大于 next，且没到底，则继续往下层查找
    25.       // Switch to next list, reuse compare_() result
    26.       last_bigger = next;
    27.       level--;
    28.     }
    29.   }
    30. }
  • 相关阅读:
    MySQL5.7慢查询实践
    MySQL -- JDBC
    机器学习实战读书笔记——端到端的机器学习项目
    华为认证 | HCIA、HCIP、HCIE，难度区别在哪里？
    缓存-基础理论和Guava Cache介绍
    技术：如何设计zkVM电路
    一个网址包含有什么？ URL-简介
    229. 多数元素 II Python
    Spring之事务实现原理及其注解@Transactional底层和传播机制原理
    python安全脚本开发简单思路
  • 原文地址：https://blog.csdn.net/ZNBase/article/details/127886018