leveldb 学习记录(三) MemTable 与 Immutable Memtable

前文: 

leveldb 学习记录(一) skiplist 

leveldb 学习记录(二) Slice

存储格式:

leveldb数据在内存中以 Memtable存储(核心结构是skiplist 已介绍),当达到一定容量则转换为Immutable Memtable,由后台线程存储进磁盘中.同时另开一个新 Memtable,记录数据.

Memtable记录修改新kv对,可读可写.Immutable Memtable不可更改.

Memtable使用的就是skiplist记录key value

class MemTable {
 public:
  // MemTables are reference counted.  The initial reference count
  // is zero and the caller must call Ref() at least once.
  explicit MemTable(const InternalKeyComparator& comparator);
//简配版应用计数 初始化时候需要引用ref将计数+1
  // Increase reference count.
  void Ref() { ++refs_; }

  // Drop reference count.  Delete if no more references exist.
//unref调用减少应用计数.计数为0 则删除自己  
void Unref() {
    --refs_;
    assert(refs_ >= 0);
    if (refs_ <= 0) {
      delete this;
    }
  }

//内存使用相关,暂时不关注
size_t ApproximateMemoryUsage();

//迭代器 类似MEMTABLE 中元素的指针
Iterator* NewIterator();

 

//KEY是按次序排序,所以结构体内有比较key的定义
struct KeyComparator {
const InternalKeyComparator comparator;
explicit KeyComparator(const InternalKeyComparator& c) : comparator(c) { }
int operator()(const char* a, const char* b) const;
};

 

//私有类中包含构造复制函数,达到禁止复制的目的
private:
// No copying allowed
MemTable(const MemTable&);
void operator=(const MemTable&);
}

Add Get 添加与读取函数, 删除和修改也是添加完成.

修改删除的优化:

实际上的kv删除或者修改,均未删除之前相同的Key记录,只是新增一个修改后的kv对或者带有删除标记的kv对.

因为系统在查找kv对是以由新至旧次序查找,所以肯定是查找到最新的删除或者修改值.

真正的冗余的老KV对在后面compac操作中才是真正的删除(后继介绍)

  // Add an entry into memtable that maps key to value at the
  // specified sequence number and with the specified type.
  // Typically value will be empty if type==kTypeDeletion.
  void Add(SequenceNumber seq, ValueType type,
           const Slice& key,
           const Slice& value);

  // If memtable contains a value for key, store it in *value and return true.
  // If memtable contains a deletion for key, store a NotFound() error
  // in *status and return true.
  // Else, return false.
  bool Get(const LookupKey& key, std::string* value, Status* s);

Add 函数添加 kTypeDeletion类的kv对,表示删除, value内容为空

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()]
  //  插入格式为
  //|--------|-------------------------|---------------------|
  //|key_size|char[internal_key.size()]|value_size|value_size|
  //|--------|-------------------------|---------------------|
  size_t key_size = key.size();
  size_t val_size = value.size();
  size_t internal_key_size = key_size + 8;
  const size_t encoded_len =
      VarintLength(internal_key_size) + internal_key_size +
      VarintLength(val_size) + val_size;　　//最后要插入skiplist的buf的长度 
  char* buf = arena_.Allocate(encoded_len);
  char* p = EncodeVarint32(buf, internal_key_size);　　//buf放入internal——key_size 32位
  memcpy(p, key.data(), key_size);　　　　　　　　　　　　//存放指针拷贝实际的key值
  p += key_size;　　　　　　　　　　　　　　　　　　　　　　　//指针偏移KEYSIZE字节
  EncodeFixed64(p, (s << 8) | type);　　　　　　　　　　　//存放64位的sequenceNumber 末尾8位空出 最后一位留给数据type
  p += 8;
  p = EncodeVarint32(p, val_size);　　　　　　　　　　　　　//存放实际val内容
  memcpy(p, value.data(), val_size);
  assert((p + val_size) - buf == encoded_len);
  table_.Insert(buf);  //skiplist insert
}

Get函数在MemTable中查找key ,查找成功返回TRUE,查找成功但是type为deletion,返回true并且status为NotFound()错误

其他情况返回false

查找有个细节  skiplist返回的是最近的大于或者等于GreaterOrEqual 所以只要关键字相同 不要求序列号sequence 

完全一样(序列号肯定是最新的最大的序列号)

然后代码里再次判断

comparator_.comparator.user_comparator()->Compare(
            Slice(key_ptr, key_length - 8),
            key.user_key()) == 0)

抛开sequence  仅仅比较key是否相等

bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {
  Slice memkey = key.memtable_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;
    const char* key_ptr = GetVarint32Ptr(entry, entry+5, &key_length);
    if (comparator_.comparator.user_comparator()->Compare(
            Slice(key_ptr, key_length - 8),
            key.user_key()) == 0) {
      // Correct user key
      const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
      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;
}

memtable 使用的InternalKey 代码如下

一个字符串的封装和 比较器InternalKeyComparator代码

// Modules in this directory should keep internal keys wrapped inside
// the following class instead of plain strings so that we do not
// incorrectly use string comparisons instead of an InternalKeyComparator.
class InternalKey {
 private:
  std::string rep_;
 public:
  InternalKey() { }   // Leave rep_ as empty to indicate it is invalid
  InternalKey(const Slice& user_key, SequenceNumber s, ValueType t) {
    AppendInternalKey(&rep_, ParsedInternalKey(user_key, s, t));
  }

  void DecodeFrom(const Slice& s) { rep_.assign(s.data(), s.size()); }
  Slice Encode() const {
    assert(!rep_.empty());
    return rep_;
  }

  Slice user_key() const { return ExtractUserKey(rep_); }

  void SetFrom(const ParsedInternalKey& p) {
    rep_.clear();
    AppendInternalKey(&rep_, p);
  }

  void Clear() { rep_.clear(); }

  std::string DebugString() const;
};

inline int InternalKeyComparator::Compare(
    const InternalKey& a, const InternalKey& b) const {
  return Compare(a.Encode(), b.Encode());
}

inline bool ParseInternalKey(const Slice& internal_key,ParsedInternalKey* result) {
　　const size_t n = internal_key.size();
　　if (n < 8) return false;
　　uint64_t num = DecodeFixed64(internal_key.data() + n - 8);
　　unsigned char c = num & 0xff; //最后一个字节 代表 类型type
　　result->sequence = num >> 8; //左移8位 获取序列号
　　result->type = static_cast<ValueType>(c);
　　result->user_key = Slice(internal_key.data(), n - 8); //除开信息位的8字节 其余便是数据 转化成 Slice
　　return (c <= static_cast<unsigned char>(kTypeValue));
}

class LookupKey //DBImpl::Get()查询使用的辅助类
使用两个指针 根据不同需求 提供不同的数据结构

可提供下列三种 Slice

Slice memtable_key()
Slice internal_key()
Slice user_key()

数据都存储在 char space_[200]; // Avoid allocation for short keys

但是如果存储数据过长 则需要重新分配内存

LookupKey::LookupKey(const Slice& user_key, SequenceNumber s) {
  size_t usize = user_key.size();
  size_t needed = usize + 13;  // A conservative estimate
  char* dst;
  if (needed <= sizeof(space_)) {　　需要更多的空间  则自行分配和删除
    dst = space_;
  } else {
    dst = new char[needed];
  }
  start_ = dst;
  dst = EncodeVarint32(dst, usize + 8);
  kstart_ = dst;
  memcpy(dst, user_key.data(), usize);
  dst += usize;
  EncodeFixed64(dst, PackSequenceAndType(s, kValueTypeForSeek));
  dst += 8;
  end_ = dst;
}
inline LookupKey::~LookupKey() {
  if (start_ != space_) delete[] start_;　　//自行删除
}

整个类代码如下

// A helper class useful for DBImpl::Get()
class LookupKey {            //DBImpl::Get()查询使用的辅助类
 public:
  // Initialize *this for looking up user_key at a snapshot with
  // the specified sequence number.
  LookupKey(const Slice& user_key, SequenceNumber sequence);

  ~LookupKey();

  // Return a key suitable for lookup in a MemTable.
  Slice memtable_key() const { return Slice(start_, end_ - start_); }

  // Return an internal key (suitable for passing to an internal iterator)
  Slice internal_key() const { return Slice(kstart_, end_ - kstart_); }

  // Return the user key
  Slice user_key() const { return Slice(kstart_, end_ - kstart_ - 8); }

 private:
  // We construct a char array of the form:
  //    klength  varint32               <-- start_
  //    userkey  char[klength]          <-- kstart_
  //    tag      uint64
  //                                    <-- end_
  // The array is a suitable MemTable key.
  // The suffix starting with "userkey" can be used as an InternalKey.
  const char* start_;
  const char* kstart_;
  const char* end_;
  char space_[200];      // Avoid allocation for short keys

  // No copying allowed
  LookupKey(const LookupKey&);
  void operator=(const LookupKey&);
};

inline LookupKey::~LookupKey() {
  if (start_ != space_) delete[] start_;
}

}

参考

https://blog.csdn.net/tankles/article/details/7663635

https://blog.csdn.net/sparkliang/article/details/8604424

http://www.cnblogs.com/haippy/archive/2011/12/04/2276064.html