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


相关文章