LevelDB原理解析：数据的读写与合并是怎样发生的？

2022-04-18 00:00:00 数据 文件 线程 队列 写入


导语 | LevelDB是一款十分的存储引擎，具有极高的数据读写性能，尤其是写入性能，在笔者经历的多个项目中都有用到，因此本文打算结合LevelDB的部分源码对 LevelDB进行介绍，首先会介绍LevelDB的整体架构，然后围绕数据读写流程和合并流程展开介绍，希望与大家一同交流。文章作者：唐文博，腾讯优图实验室研究员。


一、LevelDB总体架构



LevelDB是一款写性能极高、可靠的单机存储引擎，是LSM-Tree的典型实现，LSM-Tree主要的思想是牺牲部分读性能，大化提升数据写入性能，因此LevelDB很适合被应用在写多读少的场景。

同时LevelDB还有数据在磁盘上按key顺序存储，支持按snapshot快照查询等特性。如下图所示，LevelDB主要由驻于内存的缓存结构和存在于磁盘的物理文件组成。


1. 内存缓存结构

  • Memtable：Memtable可读可写，内部由SkipList实现，用于在内存中缓存写操作。

  • Immutable Memtable：内部同样由SkipList实现，但是只读，当Memtable的大小达到设定的阈值时，会变成 Immutable Memtable，后续由后台线程通过compaction操作将数据顺序落盘，变成sstable文件。


2. 磁盘文件

  • sstable：sstable是磁盘上的存储文件，它将key有序存放，level0层的sstable由内存中的Immutable Memtable直接持久化生成，因为没有和当前层的其他文件合并过，因此level0层的sstable里的key会发生重叠，其余层的sstable文件均由当前层和前一层的sstable文件归并而来。

  • Manifest：Manifest文件是sstable的索引信息，用来记录每个sstable对应的key range、文件size等信息。

  • Log：Log文件主要是用于机器重启而不丢失数据，当向LevelDB写入一条数据时，它首先会向Log文件顺序写入一条操作日志，然后再向内存Memtable写入数据，这样即便机器掉电，也不会出现数据丢失的情况。

  • Current文件：当机器重启时，LevelDB会重新生成新的Manifest文件，所以Manifest文件可能存在多个，这里会使用Current文件记录当前使用的Manifest文件。


二、写入流程



LevelDB对外提供的写入接口有Put和Delete两种，两者本质上对应同一种操作，实际上都会向Memtable及Log文件中追加一条新纪录。

同时LevelDB支持调用端使用多线程并发写入数据，并且会使用写队列+合并写&WAL机制，将批量随机写转化成一次顺序写，从而提升写入性能。下边将结合部分源码来看看LevelDB具体是怎么实现的。

1. 具体写入流程

（1）封装WriteBatch和Writer对象

DB::Put会把key、value方法封装到WriteBatch中，DBImpl::Write方法会把WriteBatch对象封装到Writer对象中，此外Writer对象还封装了mutex_，条件变量等用来实现等待通知。 


Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {  WriteBatch batch;  batch.Put(key, value);  //调用DBImpl::Write方法    return Write(opt, &batch);}struct DBImpl::Writer {  Status status;  WriteBatch* batch;  bool sync;  bool done;  port::CondVar cv;  explicit Writer(port::Mutex* mu) : cv(mu) { }};

（2）Writer串行化入队

多个线程并行的写入操作，会通过抢锁串行化，线程将Writer放到写队列之后，会进入等待状态，直到满足如下两个条件：

  • 其他线程已经帮忙把Writer写入；

  • 抢到锁并且是写队列的首节点。




Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {  Writer w(&mutex_);  w.batch = updates;  w.sync = options.sync;  w.done = false;  MutexLock l(&mutex_);  writers_.push_back(&w);  //任务放到queue中，如果当前不是queue的头部则等待  //当某个线程将queue中自己对应的Writer写入磁盘时，可能也会将其他线程对应的Writer写入磁盘  while (!w.done && &w != writers_.front()) {    w.cv.Wait();  }  if (w.done) {    return w.status;  }

（3）确认写入空间足够

处于写队列头部的线程会调用MakeRoomForWrite方法，MakeRoomForWrite方法会检查Memtable是否有足够的空间写入，它会将内存占用过高的Memtable转换成Immutable，并构造一个新的Memtable进行写入，刚刚形成的Immutable则交由后台线程dump到level0层。
Status DBImpl::MakeRoomForWrite(bool force) {      //通过改变指针指向，将Memtable转换成Immutable      imm_ = mem_;      has_imm_.store(true, std::memory_order_release);      //生成新的Memtable      mem_ = new MemTable(internal_comparator_);      mem_->Ref();      //触发compaction      MaybeScheduleCompaction();}

（4）批量取任务，进行合并写

处于写队列头部的线程进行完MakeRoomForWrite检查之后，便会从writers_写队列里取出头部任务，同时会遍历队列中后面的Writer合并到自身，进行批量写，从而提高写入效率，终多个Writer任务会先被写入Log文件，然后被写入内存的MemTable。


//从队列中批量取任务 WriteBatch* write_batch = BuildBatchGroup(&last_writer);//将任务写入Log文件status = log_->AddRecord(WriteBatchInternal::Contents(write_batch));//将任务写入Memtablestatus = WriteBatchInternal::InsertInto(write_batch, mem_);

（5）唤醒正在等待的线程

线程写入完成后，会对写完的Writer出队，并唤醒正在等在的线程，同时也会唤醒写队列中新的头部Writer对应的线程。
  //last_writer指向writers_里合并的后一个Writer  //逐个遍历弹出writers_里的元素，并唤醒对应的正在等待的写线程，直到遇到last_writer  while (true) {    Writer* ready = writers_.front();    writers_.pop_front();    if (ready != &w) {      ready->status = status;      ready->done = true;      ready->cv.Signal();    }    if (ready == last_writer) break;  }  // 唤醒队列未写入的个Writer  if (!writers_.empty()) {    writers_.front()->cv.Signal();  }

后对写入步骤进行简单总结，如下图所示，三个写线程同时调用LevelDB的Put接口并发写入，三个线程首先会通过抢锁将构造的Writer对象串行的放入writers_写队列，这时Writer1处于写队列头部，thread1会执行批量写操作，不仅会把自己构造的Writer写入，还会从队列中取出thread2、thread3对应的Writer，后将三者一起写入Log文件及内存Memtable，thread2、thread3在push完之后则会进入等待状态。thread1写入完成之后，会唤醒处于等待状态的thread2和thread3。


三、读取流程



LevelDB的读取流程相对简单，从其中读取一个数据，会按照从上而下memtable -> immutable -> sstable的顺序读取，读不到则从下一个层级读取，因此LevelDB更适合读取新写入的数据。流程如下图：


Level0中的文件直接由Immutable Memtable通过dump产生，文件之间key可能相互重叠，所以需要对level0的每个文件依次查找。

对于其他层次，LevelDB的归并过程保证了其中的key互相不重叠并且有序，因此可以直接使用二分方式进行数据查找。部分代码如下：


{    mutex_.Unlock();    // First look in the memtable, then in the immutable memtable (if any).    LookupKey lkey(key, snapshot);    //先查找memtable    if (mem->Get(lkey, value, &s)) {    //再查找immutable memtable    } else if (imm != nullptr && imm->Get(lkey, value, &s)) {    } else {      //查找sstable      s = current->Get(options, lkey, value, &stats);      have_stat_update = true;    }    mutex_.Lock();  }

四、Compaction流程



Compaction是LevelDB中相对比较复杂的操作，这里仅对其中比较主要的点进行介绍。compactcion分为2种，一是minor compaction，另一种是major compaction。通过compaction操作可以达到以下几个效果：

  • 将内存中的数据持久化到磁盘；

  • 清理冗余数据，因为LevelDB的更新和删除操作具有延后性，两种操作实际上都会向LevelDB写入一条新记录，所以通过重新compaction整理数据，可以清理冗余数据，节省磁盘空间；

  • 通过compaction使level 0以下的文件层中的数据保持有序，这样便可以通过二分进行数据查找，同时也可以减少待查找的文件数量，提升读效率。


1. minor compaction


minor compaction相对简单，对应Immutable持久化到level0层的过程。但是如果这一步骤的处理耗时过长，那么就会导致内存中的Memtable无法写入但又没有办法及时转化成Immutable，所以高性能持久化是对minor compaction主要的要求。

为了提升数据持久化的速度，在对Immutable进行持久化时不会考虑不同文件间的重复和顺序问题，这样带来的问题是对读不够友好，读取数据时需要读取level0层的所有文件。

（1）触发minor compaction的时机

当内存中的memtable size 小于配置的阈值时，数据都会直接更新到memtable。超过大小后，memtable转化为Immutable，这时会由一个后台线程负责将Immutable持久化到磁盘成为level0的sstable文件。

（2）compaction具体流程

  • 将Immutable memtable落盘成SSTable文件


DBImpl::WriteLevel0Table会将Immutable memtable落盘成SSTable文件，同时会将文件信息记录到edit(用于存储文件的摘要信息，如key range, file_size等)中。值得注意的是，新生成的SSTable文件实际上并不总是被放到Level0层，如果新生成的sstable的key与当前Level1层所有文件都没有重叠，则会直接将文件放到Level1层。


Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit,                                Version* base) {  //生成sstable编号，用于构建文件名  FileMetaData meta;  meta.number = versions_->NewFileNumber();  Status s;  {    mutex_.Unlock();    //更新memtable中的全部数据到xxx.ldb文件    //meta记录key range, file_size等sst信息    s = BuildTable(dbname_, env_, options_, table_cache_, iter, &meta);    mutex_.Lock();  }  int level = ;  if (s.ok() && meta.file_size > ) {    const Slice min_user_key = meta.smallest.user_key();    const Slice max_user_key = meta.largest.user_key();    if (base != nullptr) {      //为新生成的sstable选择合适的level(不一定总是0）      level = base->PickLevelForMemTableOutput(min_user_key, max_user_key);    }    //level及file meta记录到edit    edit->AddFile(level, meta.number, meta.file_size, meta.smallest,                  meta.largest);  }}

（2）将edit信息记录到version_

WriteLevel0Table执行完成之后，会将新生成的edit信息记录到version_（version_是整个LevelDB的元信息。每当因为 compaction生成新的sstable时，version_就会随之改动）中，当前的version_作为数据库的一个新状态，后续的读写操作都会基于该状态。
//记录edit信息versions_->LogAndApply(&edit, &mutex_);//释放imm_空间imm_->Unref();imm_ = nullptr;has_imm_.Release_Store(nullptr);//清理无用文件DeleteObsoleteFiles();

2. major compation


major compaction负责将磁盘上的sstable进行合并，每合并一次，sstable中的数据就落到更底一层，数据慢慢被合并到底层的level。

这种设计带来的一个明显的好处是可以清理冗余数据，节省磁盘空间，因为之前被标记删的数据可以在major compaction的过程中被清理。

level0中数据文件之间是无序的，但被归并到level1之后，数据变得有序，这使读操作需要查询的文件数就会变少，因此，major compaction带来的另一个好处是可以提升读效率。

（1）触发major compaction的时机

  • level 0层：sstable文件个数超过指定个数。因为level0是从Immutable直接转储而来，所以用个数限制而不是文件大小。

  • level i层：第i层的sstable size总大小超过(10^i) MB。level越大，说明数据越冷，读取的几率越小，因此对于level更大的层，给定的size阈值更大，从而减少comaction次数。

  • 对于sstable文件还有seek次数限制，如果文件多次seek但是一直没有查找到数据，那么就应该被合并了，否则会浪费更多的seek。


（2）compaction流程

  • 选择合适的level及sstable文件用于合并


筛选文件会根据size_compaction规则（level0层的sstable文件个数或当前level的sstable size总大小）或者seek_compaction规则（文件空seek的次数）计算应当合并的文件。

对于size_compaction，leveldb首先为每一层计算一个score，后会选择score大的level层的文件进行合并：

  1. level 0层的score计算规则为：文件数 / 4；

  2. level i层的计算规则为：整个level所有的file size总和/(10^i)。



void VersionSet::Finalize(Version* v) {  // Precomputed best level for next compaction  int best_level = -1;  double best_score = -1  for (int level = ; level < config::kNumLevels - 1; level++) {    double score;    //对于level 0使用文件数/4计算score    if (level == ) {      score = v->files_[level].size() /              static_cast<double>(config::kL0_CompactionTrigger);    } else {      //对于非0层，使用该层文件的总大小      // Compute the ratio of current size to size limit.      const uint64_t level_bytes = TotalFileSize(v->files_[level]);      score =          static_cast<double>(level_bytes) / MaxBytesForLevel(options_, level);    }    if (score > best_score) {      best_level = level;      best_score = score;    }  } //使用compaction_level_记录需要合并的层，使用compaction_score_记录合并分数  v->compaction_level_ = best_level;  v->compaction_score_ = best_score;}

对于seek_compaction，会为每一个新的sstable文件维护一个allowed_seek的初始阈值，表示多容忍多少次seek miss，当allowed_seeks递减到小于0了，那么会将对应的文件标记为需要compact。


bool Version::UpdateStats(const GetStats& stats) {  FileMetaData* f = stats.seek_file;  if (f != nullptr) {    f->allowed_seeks--;    if (f->allowed_seeks <=  && file_to_compact_ == nullptr) {      file_to_compact_ = f;      file_to_compact_level_ = stats.seek_file_level;      return true;    }  }  return false;}

  • 根据key重叠情况扩大输入文件集合


根据key重叠情况扩大输入文件集合的基本思想是：所有有重叠的level+1层文件都要参与compact，得到这些文件后，反过来看下，在不增加level+1层文件的前提下，能否继续增加level层的文件。具体步骤如下：


  • 多路合并


多路合并会将上一步骤选出来的待合并sstable中的数据按序整理。如下，代码中VersionSet::MakeInputIterator函数返回了一个迭代器对象，通过遍历该迭代器对象，则可以得到全部有序的key集合。
Iterator* VersionSet::MakeInputIterator(Compaction* c) {  const int space = (c->level() ==  ? c->inputs_[].size() + 1 : 2);  // list存储所有Iterator  Iterator** list = new Iterator*[space];  int num = ;  for (int which = ; which < 2; which++) {    if (!c->inputs_[which].empty()) {      //第0层      if (c->level() + which == ) {        const std::vector<FileMetaData*>& files = c->inputs_[which];        // Iterator* Table::NewIterator        for (size_t i = ; i < files.size(); i++) {          list[num++] = table_cache_->NewIterator(              options, files[i]->number, files[i]->file_size);        }      } else {        // Create concatenating iterator for the files from this level        list[num++] = NewTwoLevelIterator(            // 遍历文件列表的iterator            new Version::LevelFileNumIterator(icmp_, &c->inputs_[which]),            &GetFileIterator, table_cache_, options);      }    }  }

参考链接：

[1] https://leveldb-handbook.readthedocs.io/zh/latest/basic.html
[2] https://draveness.me/bigtable-leveldb/
[3] https://izualzhy.cn/start-leveldb
来自：https://mp.weixin.qq.com/s/LIENlLDeF8shOnvhksR9RQ


相关文章