leveldb源码剖析----compaction

根据前面的分析，背景线程的主体工作在BackgroundCompaction函数中完成。这个函数主要完成以下两个工作：

1. 如果imm_非空，则将imm_写入到磁盘中生成新的sstable文件
2. 对level中的文件进行合并。合并的目的主要是避免某个level中sstable文件过多，并且可以通过合并的过程删除掉过期的key-value和被用户删除的key-value。

这篇文章主要是从BackgroundCompaction函数开始，分析level中文件合并的过程

void DBImpl::BackgroundCompaction() { //在背景线程中执行mutex_.AssertHeld();if (imm_ != NULL) {CompactMemTable(); //将imm_写到level 0return;}

• 1
• 2
• 3
• 4
• 5
• 6
• 7

这部分主要是完成上面所说的第一个工作，将imm_写入level 0 中。因为这个函数需要处理很多整个数据库共享的数据结构，比如imm_，versions_等，因此必须保证在临界区中执行。

  Compaction* c;bool is_manual = (manual_compaction_ != NULL);InternalKey manual_end;if (is_manual) {ManualCompaction* m = manual_compaction_;c = versions_->CompactRange(m->level, m->begin, m->end); m->done = (c == NULL);if (c != NULL) {manual_end = c->input(0, c->num_input_files(0) - 1)->largest;}Log(options_.info_log,"Manual compaction at level-%d from %s .. %s; will stop at %s\n",m->level,(m->begin ? m->begin->DebugString().c_str() : "(begin)"),(m->end ? m->end->DebugString().c_str() : "(end)"),(m->done ? "(end)" : manual_end.DebugString().c_str()));}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17

这部分是在设置manual_compaction_ 的时候被执行，它主要是用于测试数据库的compaction功能，用于debug，数据库实际运行时一般不会运行这段代码，这里且略过，不过从功能实现上和后面的分析也是一样的。

else {c = versions_->PickCompaction(); //找出最适合compaction的level}

• 1
• 2
• 3

这个条件分支是一个重点，它负责从数据库中选出适合进行compaction的level。它将返回可以进行compaction的level中的文件元信息，这些元信息存储在Compaction类中

可以看一下compaction的数据成员：

这个compaction将会合并level和level+1中的部分文件。

如果我们深入到PickCompaction就会看到leveldb是怎么选择需要进行compaction的level的。

leveldb首先从当前的versions中选择一个最适合进行compaction的level，这里的最适合主要是通过版本控制里面的compaction_score_变量进行衡量，同时每个版本都会有一个变量跟踪当前版本中最适合进行compaction的level(current_->compaction_level_)。除此之外，leveldb为每个level中的文件维持一个 数组compact_pointer_，这个compact_pointer_[level]指向当前level中上次被compaction的最大key的值，因此下次对这个level进行compaction时，就要从key大于compact_pointer_[level]的文件开始，而且我们可以看到，通常是选择第一个largest key大于compact_pointer_[level]的文件作为当前level需要进行compaction的文件。

从当前的level选择出适合进行compaction的文件后，我们就可以通过版本控制中的元信息找到这个文件所包含的key的最大值和最小值，通过这个最大值和最小值，我们就可以找到level+1中有哪些文件和level中的这个文件的key可能有重合，然后将这些文件和level中的那个文件进行compaction，compaction后的文件放入到level+1中。需要注意的是，每次compaction可能生成多个文件，而不是一个compaction只生成一个文件。

回到上面的compaction类，通过上面步骤得到的level和level+1中需要进行compaction的文件的元信息放在compaction的数据成员inputs_数组中。其中inputs_[0]包含的是需要进行合并的level中的那个文件，inputs_[1]包含的是level+1中需要和level中的那个文件进行合并的所有文件元信息。

前面说过，一般都是从当前level中选择一个文件，然后从level+1中找到所有和这个文件的key范围有重合的文件进行合并，最后将合并的文件都放在level+1中，因此这保证了level+1中所有的文件的key不会有重合。而这里之所以只在level中选择一个文件，我们是基于level中的文件的key不会有重合的假设，这对于大于0的所有level都是成立的，但是对level 0 不成立，因为level 0 中可能会有重合的key。因此我们需要对level0 进行特殊处理，如果当前需要合并的level为0，则我们首先从level 0 中选择一个文件，然后找出level 0 中所有和这个文件的key重合的文件放入inputs_[0]中

这些可以从PickCompaction的实现中找到答案。

class compaction中的其他成员，主要是负责合并后文件的生成。比如max_output_file_size_控制合并生成的sstable文件的大小。grandparents_数组和grandparent_index_也是用于控制生成sstable文件的大小，grandparents_数组中维护的是level+2中和从level中选出的进行合并的文件的key范围重合的左右文件元信息，因此通过grandparents_数组，我们可以控制新合并生成放在level+1中的每个sstable文件不要和level+2中的过多文件有key范围重合，因为如果level+1中的某个文件和level+2中的很多文件都有key范围重合的话，那下次将level+1中的这个文件和level+2进行合并时，会比较耗时，因为level+2中和这个文件key重合的文件太多了。这里主要是平摊的思想，不要让某个文件承担太多的合并压力。

class compaction中的seen_key_主要是保证每个合并而成的sstable中都有key-value数据。想象一个场景，如果合并过程中第一个打算加入的key是一个比level+2中很多文件的最大key都大的数值，则我们可能会误以为当前合并而成的文件已经足够大了，准备把它写盘，但是实际却是当前文件中还没有key-value，于是就会出现问题。

class compaction中的其他成员变量主要是版本控制信息，后面再介绍。

level_ptrs_也是一个数组，它里面存储的是整型变量，记录每次遍历各层文件时的下标信息，主要用于判断一个key是否在level+2以及更高层的的文件中。可以看IsBaseLevelForKey函数的实现。

好了分析完compaction的结构信息，我们继续看BackgroundCompaction中的代码：

现在已经拿到了需要compaction的文件信息，这些信息就存储在c中

  Status status;if (c == NULL) {// Nothing to do} 

• 1
• 2
• 3
• 4

如果c为空，说明没有文件需要进行compaction，那就无事可做了

else if (!is_manual && c->IsTrivialMove()) { // Move file to next levelassert(c->num_input_files(0) == 1);FileMetaData* f = c->input(0, 0);c->edit()->DeleteFile(c->level(), f->number);c->edit()->AddFile(c->level() + 1, f->number, f->file_size,f->smallest, f->largest);status = versions_->LogAndApply(c->edit(), &mutex_); if (!status.ok()) {RecordBackgroundError(status);}VersionSet::LevelSummaryStorage tmp;Log(options_.info_log, "Moved #%lld to level-%d %lld bytes %s: %s\n",static_cast<unsigned long long>(f->number),c->level() + 1,static_cast<unsigned long long>(f->file_size),status.ToString().c_str(),versions_->LevelSummary(&tmp));}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21

这个条件分支主要是处理level+1中没有文件需要和level中的那个文件进行合并的情况。这种情况，很简单，直接把level中的那个需要合并的文件移动到level+1中即可。

{ // 否则进行compactionCompactionState* compact = new CompactionState(c); // c中包含需要compaction的文件的元信息status = DoCompactionWork(compact); if (!status.ok()) {RecordBackgroundError(status);}CleanupCompaction(compact);c->ReleaseInputs();DeleteObsoleteFiles();}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10

这个分支就是主要的工作了。此时说明level+1中有文件和level中的那个需要合并的文件key范围重合。因此需要将这个文件和level+1中的那些文件进行合并操作。主体工作是在DoCompactionWork函数中完成。

下面我们分析DoCompactionWork的源码，至于BackgroundCompaction函数，到这里已经基本把我们关心的工作完成了。后面我们就不会到BackgroundCompaction函数了。

DoCompactionWork函数的实现

leveldb的compaction操作主要是由DoCompactionWork函数完成：

Status DBImpl::DoCompactionWork(CompactionState* compact) { const uint64_t start_micros = env_->NowMicros();int64_t imm_micros = 0;  // Micros spent doing imm_ compactionsLog(options_.info_log,  "Compacting %d@%d + %d@%d files",compact->compaction->num_input_files(0),compact->compaction->level(),  compact->compaction->num_input_files(1),compact->compaction->level() + 1);assert(versions_->NumLevelFiles(compact->compaction->level()) > 0);assert(compact->builder == NULL);assert(compact->outfile == NULL);if (snapshots_.empty()) {compact->smallest_snapshot = versions_->LastSequence();} else {compact->smallest_snapshot = snapshots_.oldest()->number_;}// Release mutex while we're actually doing the compaction workmutex_.Unlock();

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21

最开始这部分不是我们关心的内容，先放着，后面再介绍

  Iterator* input = versions_->MakeInputIterator(compact->compaction);input->SeekToFirst();

• 1
• 2

这是获得一个可以遍历需要compaction的所有文件(level和level+1中所有需要进行compaction操作的文件)的迭代器，每个迭代器对应一个key-value，这样，我们通过这个迭代器就可以找到compaction结构中的inputs_数组里面的所有文件的key-value。leveldb里面提供了很多迭代器，它通过迭代器封装了文件内部访问的复杂细节。

  Status status;ParsedInternalKey ikey;std::string current_user_key;bool has_current_user_key = false;SequenceNumber last_sequence_for_key = kMaxSequenceNumber;

• 1
• 2
• 3
• 4
• 5

这些是对后面需要用到的一些变量的初始化。

后面就是一个大循环，这个循环依次通过上面的迭代器遍历所有参与compaction的文件的所有key，循环的主体工作是判断当前迭代器对应的key是否应该加入到新合并生成的文件中

  for (; input->Valid() && !shutting_down_.Acquire_Load(); ) { //每个input对应的是一个 K/V// Prioritize immutable compaction workif (has_imm_.NoBarrier_Load() != NULL) {const uint64_t imm_start = env_->NowMicros();mutex_.Lock();if (imm_ != NULL) {CompactMemTable(); //这里就是将imm_写入磁盘中bg_cv_.SignalAll();  // Wakeup MakeRoomForWrite() if necessary}mutex_.Unlock();imm_micros += (env_->NowMicros() - imm_start);}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12

每次循环开始先判断当前的imm_是否为空，如果为空的话，先将它写入磁盘，这主要是为了防止imm_没有及时写盘造成用户线程不能写mem。可以参见前面的分析。

    Slice key = input->key(); if (compact->compaction->ShouldStopBefore(key) &&compact->builder != NULL) {status = FinishCompactionOutputFile(compact, input); //生成一个sstableif (!status.ok()) {break;}}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8

这里先把迭代器对应的key提取出来，因为在此之前我们可能以及遍历过多个key-value了，也就是可能已经将多个key-value写入到新的sstable中了。这里通过ShouldStopBefore函数判断是否符合生成一个新的sstable的条件，如果符合的话就将这个sstable写盘，如果不符合的话，就继续往里面加key-value。

前面我们分析过，影响是否将当前的sstable写盘的主要有两个原因：

1. 当前的sstable是否已经足够大了
2. 当前的sstable是否和过多的level+2中的文件重合

这里处理的是第二个条件。这里需要注意的是，到目前位置这个迭代器对应的key-value都没有写入到当前的sstable中。

bool drop = false;

• 1

这是一个标记位，主要是用于标记一个key是否应该加入到当前的sstable中。如果drop=true则说明这个当前key应该被丢弃，反则反之。

   if (!ParseInternalKey(key, &ikey)) { //把sequence，type，key都解析出来，放在ikey中// Do not hide error keyscurrent_user_key.clear();has_current_user_key = false;last_sequence_for_key = kMaxSequenceNumber;}

• 1
• 2
• 3
• 4
• 5
• 6

首先对key-value进行解析，前面我们分析过，sstable中存储的key-value是以如下的形式：

internal_key_size|key|sequence|type|value_size|value

• 1

这里主要是把key，sequence，type解析出来。

如果发现这个key不合法，则设置一些标记变量。后面我们再讲解这些变量的作用。这些标记位的作用就是保证这个key一定会被写入到新合并生成的sstable文件中。并且它不对后面的key-value是否被丢弃产生影响。leveldb之所以选择不丢弃不合法的key-value，我想主要是为了不想隐藏系统可能的一些错误？

如果key-value形式合法，则走到下面的一个大else

else {if (!has_current_user_key ||user_comparator()->Compare(ikey.user_key,Slice(current_user_key)) != 0) { //如果等于0，表示这个key和之前度过的key相同，不必再添加了// First occurrence of this user keycurrent_user_key.assign(ikey.user_key.data(), ikey.user_key.size());has_current_user_key = true;last_sequence_for_key = kMaxSequenceNumber; //}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9

这个是判断当前迭代器的key和前面加入的一个key是否相等，如果相等的话，那说明这个key是一个过期的key，应该被丢弃，如果这个key是第一次出现，则将其设置为current_user_key，(1). 但是还不能确定把他加入到新的sstable中去，因为可能这个key的type是kTypeDeletion，表示这个key已经被用户删除了，因此自然应该把它丢弃掉；(2). 除此之外，对于过期的key，我们也不是一定会像前面说的那样把它丢弃掉，因为可能系统开启了快照，这样老旧的key也得保存下来。

所以后面有两个条件分支，分别处理这两种情况：

1. 对于过期的key，通过检查这个key的序列号，判断它是否在系统快照中，如果在的话，即使它是过期key也不能丢弃。
2. 对于非过期的key，检查这个key的type，看它是否是kTypeDeletion，即是否已经被用户删除了。

      if (last_sequence_for_key <= compact->smallest_snapshot) {// Hidden by an newer entry for same user keydrop = true;    // (A)} else if (ikey.type == kTypeDeletion &&ikey.sequence <= compact->smallest_snapshot &&compact->compaction->IsBaseLevelForKey(ikey.user_key)) {// For this user key:// (1) there is no data in higher levels// (2) data in lower levels will have larger sequence numbers// (3) data in layers that are being compacted here and have//     smaller sequence numbers will be dropped in the next//     few iterations of this loop (by rule (A) above).// Therefore this deletion marker is obsolete and can be dropped.drop = true;}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15

对于第一个if，从最前面的if我们看到，当我们在发现当前key是第一次出现时会设置last_sequence_for_key = kMaxSequenceNumber;因此走这个if说明该key肯定是一个过期key了。所以判断它的序列号是否在快照序列号中，不是的话就标记drop = true，将其丢弃。

至此为止，当前key要么是第一次出现，要么是有快照保护，好像不能丢弃。但是事情还没完，我们还不能据此判断该key是否应该被保存下来，我们还要判断它的type，但是事情远没有我们想得那么简单，除了判断key的type之外，我们还要做其他判断.也就是说，当一个key为kTypeDeletion时它还不一定是要被删除的。为什么呢。

想象一个场景：用户在调用delete删除一个key时，这个key在数据库中有一个过期的key存在，而这个过期key还来不及和这个被删除的key合并，如果在这种情况下，我们直接将这个被标记为删除的key丢弃，那数据库中还会存在一个过期的key，而这个过期的key在丢弃那个被删除的key时瞬间就变成正常不过期的key了，于是下次读key时，会读到这个本应该过期的key，按道理应该是找不到key才对。所以为了系统正常运行，我们每次丢弃一个标记为kTypeDeletion的key时，必须保证数据库中不存在它的过期key，否则就得将它保留，直到后面它和这个过期的key合并为止，合并之后再丢弃。

所以这里会调用IsBaseLevelForKey函数判断level+2及level+2之后的所有level中没有和这个标记为删除的key相同的key，只要有，就肯定是过期key了。

last_sequence_for_key = ikey.sequence;}

• 1
• 2

这里就是标记last_sequence_for_key 变量，用它来标记当前key的序列号。

    if (!drop) {// Open output file if necessaryif (compact->builder == NULL) {status = OpenCompactionOutputFile(compact);if (!status.ok()) {break;}}if (compact->builder->NumEntries() == 0) {compact->current_output()->smallest.DecodeFrom(key);}compact->current_output()->largest.DecodeFrom(key);compact->builder->Add(key, input->value()); //把这个键值加入新建的sstable文件中// Close output file if it is big enough  // 如果当前结果文件已经足够大，则关闭文件，以后的compaction结果再放到新的文件中if (compact->builder->FileSize() >=compact->compaction->MaxOutputFileSize()) {status = FinishCompactionOutputFile(compact, input);if (!status.ok()) {break;}}}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23

OK，如果drop为false，说明当前key应该被保留下来。下面就将当前迭代器对应的key-value加入到sstable中，就是通过TableBuilder完成这些工作。前面我们讲过了。

加入这个key-value到sstable后，还要判断当前的sstable是不是满足写盘条件，即满足生成一个完整sstable的文件。

input->Next();

• 1

继续下一个key-value。迭代器封装了所有细节。后面我们会专门介绍leveldb中的各种迭代器。

大循环之后就完成了文件合并的核心工作，循环之后是一些设置版本信息和写日志的工作，这个我们后面再介绍了。

总结

compaction是leveldb中最核心的东西了。(1). 前面我们说过，当用户调用delete删除一个键值时，leveldb并没有真正把它删除掉，而只是简单将这个键值对的type标记为kTypeDeletion，然后和正常的键值对一样写盘。这个特点使得leveldb的写操作很快，但是问题也是很显然的，就是会造成大量无效数据，占用磁盘空间。(2). 除此之外，leveldb添加数据时并不会将过期的key-value覆盖，而是通过序列号将其当成完全不同的key写入进去，因此会使得系统中存在很多过期数据。，毫无疑问这也是很占用空间的。

而compaction操作可以解决这些问题。通过compaction，leveldb可以将过期的key丢弃，而且在一定条件下丢弃标记为kTypeDeletion的数据。同时通过compaction控制了每层的文件数目。可以说compaction是leveldb的写操作得以高效的主要原因。

每次compaction操作是根据版本统计信息找到最适合进行compaction的level，然后从这个level中选择一个最合适的compaction文件，再去level+1中找出所有和这个文件的key重合的文件，最后将level中的这个文件和level+1中的那些文件进行合并，并将合并生成的所有sstable文件都放入到level+1层中。当然当level=0时，因为level0中的文件可能存在key重合，因此需要一些特殊处理。