Google LevelDB 原始碼解說 (四) Memtable Add&Get

Memtable有屬於自己的add和get，Immutable Memtable也大同小異，這邊對Memtable add&get的實作做說明，實作內容在 Memtable.cc 中

Add

這裡的內部格式為
| internal_key size | key_data | squence+type | value_size | value_data |
                                       (Internal key)         

void MemTable::Add(SequenceNumber s, ValueType type,
                   const Slice& key,
                   const Slice& value) {
  // Format of an entry is concatenation of:
  //  key_size     : varint32 of internal_key.size()
  //  key bytes    : char[internal_key.size()]
  //  value_size   : varint32 of value.size()
  //  value bytes  : char[value.size()]
  size_t key_size = key.size();
  size_t val_size = value.size();
//前篇文章有說 internal key = key+seq+type 
  size_t internal_key_size = key_size + 8;
  //skiplist內部格式
  const size_t encoded_len =
      VarintLength(internal_key_size) + internal_key_size +
      VarintLength(val_size) + val_size;
//分配記憶體到buf中
  char* buf = arena_.Allocate(encoded_len);
//儲存internal_key_size到buf中，之後會再詳細解釋Encode相關文章
  char* p = EncodeVarint32(buf, internal_key_size);
//將key.data存入
  memcpy(p, key.data(), key_size);
//將指標移到sequence num前
  p += key_size;
//將sequence+num 打包存入p(buf)中
  EncodeFixed64(p, (s << 8) | type);
//將指標移到儲存value地址之前
  p += 8;
//儲存value size
  p = EncodeVarint32(p, val_size);
//儲存value的值
  memcpy(p, value.data(), val_size);
  assert(p + val_size == buf + encoded_len);
//將整條結構存入skiplist中
  table_.Insert(buf);
}

Get

bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {
  Slice memkey = key.memtable_key();

  //從skiplist中尋找key值
  Table::Iterator iter(&table_);
  iter.Seek(memkey.data());
  if (iter.Valid()) {
    // entry format is:
    //    klength  varint32
    //    userkey  char[klength]
    //    tag      uint64
    //    vlength  varint32
    //    value    char[vlength]
    // Check that it belongs to same user key.  We do not check the
    // sequence number since the Seek() call above should have skipped
    // all entries with overly large sequence numbers.
    const char* entry = iter.key();
    uint32_t key_length;
    //通過以下語句獲取internal key size，因為varint最大為5，所以長度為entry到entry+5
    const char* key_ptr = GetVarint32Ptr(entry, entry+5, &key_length);
    //比較返回的key與user_key是否相同，key_length-8是因為internal key包含了sequence numbers和type
    if (comparator_.comparator.user_comparator()->Compare(Slice(key_ptr, key_length - 8),key.user_key()) == 0) {
      // Correct user key
      //對key解碼
      const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
      //比較類型，如果不是kTypeDeletion，則獲取value的size跟value的值
      switch (static_cast<ValueType>(tag & 0xff)) {
        case kTypeValue: {
          Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
          value->assign(v.data(), v.size());
          return true;
        }
        case kTypeDeletion:
          *s = Status::NotFound(Slice());
          return true;
      }
    }
  }
  return false;
}