综述

在RocksDB中，将MemTable刷新到磁盘之后,将会有很多sstable,而这些sstable则是可能包含了相同的key的不同时间的值，这样子就会导致两个问题:

  1. 浪费磁盘空间
  2. 读取内容将会非常慢.

而compact就是用来解决上面两个问题的，简单来说compact就是读取几个sstable然后合并为一个(或者多个)sstable. 而什么时候合并，合并的时候如何来挑选sstable,这个就是compcation strategy.一般来说compact strategy的目的都是为了更低的amplification:

  • 避免一次读请求读取太多的sstables.
    • 读放大
  • 避免一些临时数据(deleted/overwritten/expired)在磁盘上停留时间过长
  • 避免磁盘上临时空间过大
    • 空间放大
  • 避免compact相同的数据太多次
    • 写放大

而在RockDB中实现了多种compact　strategy,不同的strategy有不同的侧重，这里我们只分析默认的strategy, 那就是leveled-N compaction.

在Leveled compaction中，所有的SSTables被分为很多levels(level0/1/2/3…).

  • 最新的SSTable(从memtable中刷新下来的)是属于Level0
    • 每一个SSTable都是有序的
    • 只有Level0的SSTable允许overlap
  • 除了level0之外其他的level的总的SSTable大小有一个最大的限制
    • 通过level_compaction_dynamic_level_bytes来计算
  • 在Level0，如果积攒够了足够的(level0_file_num_compaction_trigger)SSTable,则就会进行compact.
    • 一般来说会把全部的SSTables compact到下一个level(Level1).
    • 不会写一个很大的SSTable,
  • 一般来说百分之90的空间都是给最后一级level的.

源码

Compact运行的条件

先来看在RocksDB中是什么时候会引起compact.在RocksDB中所有的compact都是在后台线程中进行的，这个线程就是BGWorkCompaction.这个线程只有在两种情况下被调用，一个是 手动compact(RunManualCompaction),一个就是自动(MaybeScheduleFlushOrCompaction),我们主要来看自动的compact,而MaybeScheduleFlushOrCompaction这个函数我们在之前介绍flush的时候已经介绍过了，简单来说就是会在切换WAL(SwitchWAL)或者writebuffer满的时候(HandleWriteBufferFull)被调用.

我们来看在MaybeScheduleFlushOrCompaction中compact的调用.这里可以看到RocksDB中后台运行的compact会有一个限制(max_compactions).而我们可以看到这里还有一个变量 unscheduled_compactions_，这个变量表示需要被compact的columnfamily的队列长度.

  1.   while (bg_compaction_scheduled_ < bg_job_limits.max_compactions &&
  2.          unscheduled_compactions_ > 0) {
  3.     CompactionArg* ca = new CompactionArg;
  4.     ca->db = this;
  5.     ca->prepicked_compaction = nullptr;
  6.     bg_compaction_scheduled_++;
  7.     unscheduled_compactions_--;
  8.     env_->Schedule(&DBImpl::BGWorkCompaction, ca, Env::Priority::LOW, this,
  9.                    &DBImpl::UnscheduleCallback);
  10.   }

类似flush的逻辑，compact的时候RocksDB也有一个队列叫做DBImpl::compaction_queue_.

  1.   std::deque<ColumnFamilyData*> compaction_queue_;

然后我们来看这个队列何时被更新,其中unscheduled_compactions_和队列的更新是同步的，因此只有compaction_queue_更新之后，调用compact后台线程才会进入compact处理.

  1. void DBImpl::SchedulePendingCompaction(ColumnFamilyData* cfd) {
  2.   if (!cfd->queued_for_compaction() && cfd->NeedsCompaction()) {
  3.     AddToCompactionQueue(cfd);
  4.     ++unscheduled_compactions_;
  5.   }
  6. }

上面的核心函数是NeedsCompaction,通过这个函数来判断是否有sst需要被compact，因此接下来我们就来详细分析这个函数.当满足下列几个条件之一就将会更新compact队列

  • 有超时的sst(ExpiredTtlFiles)
  • files_marked_for_compaction_或者bottommost_files_marked_for_compaction_都不为空
    • 后面会介绍这两个队列
  • 遍历所有的level的sst,然后判断是否需要compact
    • 最核心的条件(上面两个队列都是在这里更新的).
  1. bool LevelCompactionPicker::NeedsCompaction(
  2.     const VersionStorageInfo* vstorage) const {
  3.   if (!vstorage->ExpiredTtlFiles().empty()) {
  4.     return true;
  5.   }
  6.   if (!vstorage->BottommostFilesMarkedForCompaction().empty()) {
  7.     return true;
  8.   }
  9.   if (!vstorage->FilesMarkedForCompaction().empty()) {
  10.     return true;
  11.   }
  12.   for (int i = 0; i <= vstorage->MaxInputLevel(); i++) {
  13.     if (vstorage->CompactionScore(i) >= 1) {
  14.       return true;
  15.     }
  16.   }
  17.   return false;
  18. }

因此接下来我们来分析最核心的CompactionScore,这里将会涉及到两个变量,这两个变量分别保存了level以及每个level所对应的score(这里score越高表示compact优先级越高)，而score小于１则表示不需要compact.

  1.   std::vector<double> compaction_score_;
  2.   std::vector<int> compaction_level_;

这两个vector是在VersionStorageInfo::ComputeCompactionScore中被更新,因此我们来看这个函数，这个函数中会对level-0和其他的level区别处理。 首先来看level-0的处理:

  1. 首先会计算level-0下所有文件的大小(total_size)以及文件个数(num_sorted_runs).
  2. 用文件个数除以level0_file_num_compaction_trigger来得到对应的score
  3. 如果当前不止一层level,那么将会从上面的score和(total_size/max_bytes_for_level_base)取最大值.

之所以要做第三步，主要还是为了防止level-0的文件size过大，那么当它需要compact的时候有可能会需要和level-1 compact,那么此时就有可能会有一个很大的compact.

  1. if (level == 0) {
  2.       int num_sorted_runs = 0;
  3.       uint64_t total_size = 0;
  4.       for (auto* f : files_[level]) {
  5.         if (!f->being_compacted) {
  6.           total_size += f->compensated_file_size;
  7.           num_sorted_runs++;
  8.         }
  9.       }
  10. .........................
  11.       score = static_cast<double>(num_sorted_runs) /
  12.                 mutable_cf_options.level0_file_num_compaction_trigger;
  13.         if (compaction_style_ == kCompactionStyleLevel && num_levels() > 1) {
  14.           score = std::max(
  15.               score, static_cast<double>(total_size) /
  16.                      mutable_cf_options.max_bytes_for_level_base);
  17.         }
  18.       }

然后是非level-0的处理,这里也是计算level的文件大小然后再除以MaxBytesForLevel，然后得到当前level的score.

  1.       uint64_t level_bytes_no_compacting = 0;
  2.       for (auto f : files_[level]) {
  3.         if (!f->being_compacted) {
  4.           level_bytes_no_compacting += f->compensated_file_size;
  5.         }
  6.       }
  7.       score = static_cast<double>(level_bytes_no_compacting) /
  8.               MaxBytesForLevel(level);

上面我们看到有一个MaxBytesForLevel,这个函数的作用就是得到当前level的最大的文件大小.而这个函数实现也很简单.

  1. uint64_t VersionStorageInfo::MaxBytesForLevel(int level) const {
  2.   // Note: the result for level zero is not really used since we set
  3.   // the level-0 compaction threshold based on number of files.
  4.   assert(level >= 0);
  5.   assert(level < static_cast<int>(level_max_bytes_.size()));
  6.   return level_max_bytes_[level];
  7. }

可以看到核心就是level_max_bytes_这个数组，接下来我们就来看这个数组是在哪里被初始化的。level_max_bytes这个数组是在VersionStorageInfo::CalculateBaseBytes 这个函数中被初始化，这里RocksDB有一个option叫做level_compaction_dynamic_level_bytes,这个配置如果被设置，那么level_max_bytes将会这样 设置(这里我们只关注level):

  • 如果是level-1那么level-1的的文件大小限制为options.max_bytes_for_level_base.

  • 如果level大于1那么当前level-i的大小限制为(其中max_bytes这两个变量都是options中设置的)

    1. Target_Size(Ln+1) = Target_Size(Ln) * max_bytes_for_level_multiplier * max_bytes_for_level_multiplier_additional[n].

    举个例子,如果max_bytes_for_level_base=1024,max_bytes_for_level_multiplier=10,然后max_bytes_for_level_multiplier_additional未设置，那么L1, L2,L3的大小限制分别为1024,10240,102400.

下面是对应代码.

  1.   if (!ioptions.level_compaction_dynamic_level_bytes) {
  2.     base_level_ = (ioptions.compaction_style == kCompactionStyleLevel) ? 1 : -1;
  4.     // Calculate for static bytes base case
  5.     for (int i = 0; i < ioptions.num_levels; ++i) {
  6.       if (i == 0 && ioptions.compaction_style == kCompactionStyleUniversal) {
  7.         level_max_bytes_[i] = options.max_bytes_for_level_base;
  8.       } else if (i > 1) {
  9.         level_max_bytes_[i] = MultiplyCheckOverflow(
  10.             MultiplyCheckOverflow(level_max_bytes_[i - 1],
  11.                                   options.max_bytes_for_level_multiplier),
  12.             options.MaxBytesMultiplerAdditional(i - 1));
  13.       } else {
  14.         level_max_bytes_[i] = options.max_bytes_for_level_base;
  15.       }
  16.     }
  17.   }

然后我们来看如果设置了level_compaction_dynamic_level_bytes会如何来计算.如果设置了dynamic,那么就说明每次计算出来的每个level的最大值都是不一样的, 首先我们要知道调用CalculateBaseBytes是在每次创建version的时候。因此他是这样计算的.最大的level(num_levels -1 )的大小限制是不计入计算的，然后就是这样计算.

  1. Target_Size(Ln-1) = Target_Size(Ln) / max_bytes_for_level_multiplier

举个例子，假设调用CalculateBaseBytes的时候，max_bytes_for_level_base是1G,然后num_levels = 6，然后当前最大的level的大小为256G,那么从L1-L6的大小是 0, 0, 0.276GB, 2.76GB, 27.6GB 和 276GB.

首先计算第一个非空的level.

  1.     for (int i = 1; i < num_levels_; i++) {
  2.       uint64_t total_size = 0;
  3.       for (const auto& f : files_[i]) {
  4.         total_size += f->fd.GetFileSize();
  5.       }
  6.       if (total_size > 0 && first_non_empty_level == -1) {
  7.         first_non_empty_level = i;
  8.       }
  9.       if (total_size > max_level_size) {
  10.         max_level_size = total_size;
  11.       }
  12.     }

得到最小的那个非0的level的size.

  1.       uint64_t base_bytes_max = options.max_bytes_for_level_base;
  2.       uint64_t base_bytes_min = static_cast<uint64_t>(
  3.           base_bytes_max / options.max_bytes_for_level_multiplier);
  5.       // Try whether we can make last level's target size to be max_level_size
  6.       uint64_t cur_level_size = max_level_size;
  7.       for (int i = num_levels_ - 2; i >= first_non_empty_level; i--) {
  8.         // Round up after dividing
  9.         cur_level_size = static_cast<uint64_t>(
  10.             cur_level_size / options.max_bytes_for_level_multiplier);
  11.       }

找到base_level_size，一般来说也就是cur_level_size.

  1.         // Find base level (where L0 data is compacted to).
  2.         base_level_ = first_non_empty_level;
  3.         while (base_level_ > 1 && cur_level_size > base_bytes_max) {
  4.           --base_level_;
  5.           cur_level_size = static_cast<uint64_t>(
  6.               cur_level_size / options.max_bytes_for_level_multiplier);
  7.         }
  8.         if (cur_level_size > base_bytes_max) {
  9.           // Even L1 will be too large
  10.           assert(base_level_ == 1);
  11.           base_level_size = base_bytes_max;
  12.         } else {
  13.           base_level_size = cur_level_size;
  14.         }

然后给level_max_bytes_ 赋值

  1.       uint64_t level_size = base_level_size;
  2.       for (int i = base_level_; i < num_levels_; i++) {
  3.         if (i > base_level_) {
  4.           level_size = MultiplyCheckOverflow(
  5.               level_size, options.max_bytes_for_level_multiplier);
  6.         }
  7.         // Don't set any level below base_bytes_max. Otherwise, the LSM can
  8.         // assume an hourglass shape where L1+ sizes are smaller than L0. This
  9.         // causes compaction scoring, which depends on level sizes, to favor L1+
  10.         // at the expense of L0, which may fill up and stall.
  11.         level_max_bytes_[i] = std::max(level_size, base_bytes_max);
  12.       }
  13.     }

Compact实现细节

分析完毕何时会触发Compact,那么我们接下来来分析如何Compact.其中Compact的所有操作都在DBImpl::BackgroundCompaction中进行，因此接下来我们来分析 这个函数. 首先是从compaction_queue_队列中读取第一个需要compact的column family.

  1.     // cfd is referenced here
  2.     auto cfd = PopFirstFromCompactionQueue();
  3.     // We unreference here because the following code will take a Ref() on
  4.     // this cfd if it is going to use it (Compaction class holds a
  5.     // reference).
  6.     // This will all happen under a mutex so we don't have to be afraid of
  7.     // somebody else deleting it.
  8.     if (cfd->Unref()) {
  9.       delete cfd;
  10.       // This was the last reference of the column family, so no need to
  11.       // compact.
  12.       return Status::OK();
  13.     }

然后就是选取当前CF中所需要compact的内容.

  1.       c.reset(cfd->PickCompaction(*mutable_cf_options, log_buffer));

从上面可以看到PickCompaction这个函数，而这个函数会根据设置的不同的Compact策略调用不同的方法，这里我们只看默认的LevelCompact的对应函数.

  1. Compaction* LevelCompactionBuilder::PickCompaction() {
  2.   // Pick up the first file to start compaction. It may have been extended
  3.   // to a clean cut.
  4.   SetupInitialFiles();
  5.   if (start_level_inputs_.empty()) {
  6.     return nullptr;
  7.   }
  8.   assert(start_level_ >= 0 && output_level_ >= 0);
  10.   // If it is a L0 -> base level compaction, we need to set up other L0
  11.   // files if needed.
  12.   if (!SetupOtherL0FilesIfNeeded()) {
  13.     return nullptr;
  14.   }
  16.   // Pick files in the output level and expand more files in the start level
  17.   // if needed.
  18.   if (!SetupOtherInputsIfNeeded()) {
  19.     return nullptr;
  20.   }
  22.   // Form a compaction object containing the files we picked.
  23.   Compaction* c = GetCompaction();
  25.   TEST_SYNC_POINT_CALLBACK("LevelCompactionPicker::PickCompaction:Return", c);
  27.   return c;
  28. }

这里PickCompaction分别调用了三个主要的函数.

  • SetupInitialFiles　这个函数主要用来初始化需要Compact的文件.
  • SetupOtherL0FilesIfNeeded 如果需要compact的话，那么还需要再设置对应的L0文件
  • SetupOtherInputsIfNeeded 选择对应的输出文件

先来看SetupInitialFiles,这个函数他会遍历所有的level,然后来选择对应需要compact的input和output.

这里可看到，他会从之前计算好的的compact信息中得到对应的score.

  1. void LevelCompactionBuilder::SetupInitialFiles() {
  2.   // Find the compactions by size on all levels.
  3.   bool skipped_l0_to_base = false;
  4.   for (int i = 0; i < compaction_picker_->NumberLevels() - 1; i++) {
  5.     start_level_score_ = vstorage_->CompactionScore(i);
  6.     start_level_ = vstorage_->CompactionScoreLevel(i);
  7.     assert(i == 0 || start_level_score_ <= vstorage_->CompactionScore(i - 1));
  8. ................................................................
  9.   }

只有当score大于一才有必要进行compact的处理(所有操作都在上面的循环中).这里可以看到如果是level0的话，那么output_level 则是vstorage_->base_level(),否则就是level+1. 这里base_level()可以认为就是level1或者是最小的非空的level(之前CalculateBaseBytes中计算).

  1. if (start_level_score_ >= 1) {
  2.       if (skipped_l0_to_base && start_level_ == vstorage_->base_level()) {
  3.         // If L0->base_level compaction is pending, don't schedule further
  4.         // compaction from base level. Otherwise L0->base_level compaction
  5.         // may starve.
  6.         continue;
  7.       }
  8.       output_level_ =
  9.           (start_level_ == 0) ? vstorage_->base_level() : start_level_ + 1;
  10.       if (PickFileToCompact()) {
  11.         // found the compaction!
  12.         if (start_level_ == 0) {
  13.           // L0 score = `num L0 files` / `level0_file_num_compaction_trigger`
  14.           compaction_reason_ = CompactionReason::kLevelL0FilesNum;
  15.         } else {
  16.           // L1+ score = `Level files size` / `MaxBytesForLevel`
  17.           compaction_reason_ = CompactionReason::kLevelMaxLevelSize;
  18.         }
  19.         break;
  20.       } else {
  21.         // didn't find the compaction, clear the inputs
  22.   ......................................................
  23.         }
  24.       }
  25.     }

上面的代码中我们可以看到最终是通过PickFileToCompact来选择input以及output文件.因此我们接下来就来分这个函数.

首先是得到当前level(start_level_)的未compacted的最大大小的文件

  1.   // Pick the largest file in this level that is not already
  2.   // being compacted
  3.   const std::vector<int>& file_size =
  4.       vstorage_->FilesByCompactionPri(start_level_);
  5.   const std::vector<FileMetaData*>& level_files =
  6.       vstorage_->LevelFiles(start_level_);

紧接着就是这个函数最核心的功能了，它会开始遍历当前的输入level的所有待compact的文件，然后选择一些合适的文件然后compact到下一个level.

  1. unsigned int cmp_idx;
  2.   for (cmp_idx = vstorage_->NextCompactionIndex(start_level_);
  3.        cmp_idx < file_size.size(); cmp_idx++) {
  4. ..........................................    
  5.   }

然后我们来详细分析上面循环中所做的事情 首先选择好文件之后，将会扩展当前文件的key的范围，得到一个”clean cut”的范围， 这里”clean cut”是这个意思，假设我们有五个文件他们的key range分别为:

  1.     f1[a1 a2] f2[a3 a4] f3[a4 a6] f4[a6 a7] f5[a8 a9]

如果我们第一次选择了f3,那么我们通过clean cut，则将还会选择f2,f4，因为他们都是连续的. 选择好之后，会再做一次判断，这次是判断是否正在compact的out_level的文件范围是否和我们选择好的文件的key有重合，如果有，则跳过这个文件. 这里之所以会有这个判断，主要原因还是因为compact是会并行的执行的.

  1. int index = file_size[cmp_idx];
  2.     auto* f = level_files[index];
  4.     // do not pick a file to compact if it is being compacted
  5.     // from n-1 level.
  6.     if (f->being_compacted) {
  7.       continue;
  8.     }
  10.     start_level_inputs_.files.push_back(f);
  11.     start_level_inputs_.level = start_level_;
  12.     if (!compaction_picker_->ExpandInputsToCleanCut(cf_name_, vstorage_,
  13.                                                     &start_level_inputs_) ||
  14.         compaction_picker_->FilesRangeOverlapWithCompaction(
  15.             {start_level_inputs_}, output_level_)) {
  16.       // A locked (pending compaction) input-level file was pulled in due to
  17.       // user-key overlap.
  18.       start_level_inputs_.clear();
  19.       continue;
  20.     }

选择好输入文件之后，接下来就是选择输出level中需要一起被compact的文件(output_level_inputs). 实现也是比较简单，就是从输出level的所有文件中找到是否有和上面选择好的input中有重合的文件，如果有，那么则需要一起进行compact.

  1.     InternalKey smallest, largest;
  2.     compaction_picker_->GetRange(start_level_inputs_, &smallest, &largest);
  3.     CompactionInputFiles output_level_inputs;
  4.     output_level_inputs.level = output_level_;
  5.     vstorage_->GetOverlappingInputs(output_level_, &smallest, &largest,
  6.                                     &output_level_inputs.files);
  7.     if (!output_level_inputs.empty() &&
  8.         !compaction_picker_->ExpandInputsToCleanCut(cf_name_, vstorage_,
  9.                                                     &output_level_inputs)) {
  10.       start_level_inputs_.clear();
  11.       continue;
  12.     }
  13.     base_index_ = index;
  14.     break;

继续分析PickCompaction，我们知道在RocksDB中level-0会比较特殊，那是因为只有level-0中的文件是无序的，而在上面的操作中， 我们是假设在非level-0,因此接下来我们需要处理level-0的情况,这个函数就是SetupOtherL0FilesIfNeeded.

这里如果start_level_为０，也就是level-0的话，才会进行下面的处理，就是从level-0中得到所有的重合key的文件，然后加入到start_level_inputs中.

  1.   if (start_level_ == 0 && output_level_ != 0) {
  2.     // Two level 0 compaction won't run at the same time, so don't need to worry
  3.     // about files on level 0 being compacted.
  4.     assert(compaction_picker_->level0_compactions_in_progress()->empty());
  5.     InternalKey smallest, largest;
  6.     compaction_picker_->GetRange(start_level_inputs_, &smallest, &largest);
  7.     // Note that the next call will discard the file we placed in
  8.     // c->inputs_[0] earlier and replace it with an overlapping set
  9.     // which will include the picked file.
  10.     start_level_inputs_.files.clear();
  11.     vstorage_->GetOverlappingInputs(0, &smallest, &largest,
  12.                                     &start_level_inputs_.files);
  14.     // If we include more L0 files in the same compaction run it can
  15.     // cause the 'smallest' and 'largest' key to get extended to a
  16.     // larger range. So, re-invoke GetRange to get the new key range
  17.     compaction_picker_->GetRange(start_level_inputs_, &smallest, &largest);
  18.     if (compaction_picker_->IsRangeInCompaction(
  19.             vstorage_, &smallest, &largest, output_level_, &parent_index_)) {
  20.       return false;
  21.     }
  22.   }

假设start_level_inputs被扩展了，那么对应的output也需要被扩展，因为非level0的其他的level的文件key都是不会overlap的. 那么此时就是会调用SetupOtherInputsIfNeeded.

  1.   if (output_level_ != 0) {
  2.     output_level_inputs_.level = output_level_;
  3.     if (!compaction_picker_->SetupOtherInputs(
  4.             cf_name_, mutable_cf_options_, vstorage_, &start_level_inputs_,
  5.             &output_level_inputs_, &parent_index_, base_index_)) {
  6.       return false;
  7.     }
  9.     compaction_inputs_.push_back(start_level_inputs_);
  10.     if (!output_level_inputs_.empty()) {
  11.       compaction_inputs_.push_back(output_level_inputs_);
  12.     }
  14.     // In some edge cases we could pick a compaction that will be compacting
  15.     // a key range that overlap with another running compaction, and both
  16.     // of them have the same output level. This could happen if
  17.     // (1) we are running a non-exclusive manual compaction
  18.     // (2) AddFile ingest a new file into the LSM tree
  19.     // We need to disallow this from happening.
  20.     if (compaction_picker_->FilesRangeOverlapWithCompaction(compaction_inputs_,
  21.                                                             output_level_)) {
  22.       // This compaction output could potentially conflict with the output
  23.       // of a currently running compaction, we cannot run it.
  24.       return false;
  25.     }
  26.     compaction_picker_->GetGrandparents(vstorage_, start_level_inputs_,
  27.                                         output_level_inputs_, &grandparents_);
  28.   }

最后就是构造一个compact然后返回.

  1.   // Form a compaction object containing the files we picked.
  2.   Compaction* c = GetCompaction();
  4.   TEST_SYNC_POINT_CALLBACK("LevelCompactionPicker::PickCompaction:Return", c);
  6.   return c;

最后再回到BackgroundCompaction中，这里就是在得到需要compact的文件之后，进行具体的compact. 这里我们可以看到核心的数据结构就是CompactionJob,每一次的compact都是一个job,最终对于文件的compact都是在 CompactionJob::run中实现.

  1. CompactionJob compaction_job(
  2.         job_context->job_id, c.get(), immutable_db_options_,
  3.         env_options_for_compaction_, versions_.get(), &shutting_down_,
  4.         preserve_deletes_seqnum_.load(), log_buffer, directories_.GetDbDir(),
  5.         GetDataDir(c->column_family_data(), c->output_path_id()), stats_,
  6.         &mutex_, &bg_error_, snapshot_seqs, earliest_write_conflict_snapshot,
  7.         snapshot_checker, table_cache_, &event_logger_,
  8.         c->mutable_cf_options()->paranoid_file_checks,
  9.         c->mutable_cf_options()->report_bg_io_stats, dbname_,
  10.         &compaction_job_stats);
  11.     compaction_job.Prepare();
  13.     mutex_.Unlock();
  14.     compaction_job.Run();
  15.     TEST_SYNC_POINT("DBImpl::BackgroundCompaction:NonTrivial:AfterRun");
  16.     mutex_.Lock();
  18.     status = compaction_job.Install(*c->mutable_cf_options());
  19.     if (status.ok()) {
  20.       InstallSuperVersionAndScheduleWork(
  21.           c->column_family_data(), &job_context->superversion_context,
  22.           *c->mutable_cf_options(), FlushReason::kAutoCompaction);
  23.     }
  24.     *made_progress = true;

在RocksDB中，Compact是会多线程并发的执行，而这里怎样并发，并发多少线程都是在CompactionJob中实现的，简单来说，当你的compact的文件range不重合的话，那么都是可以并发执行的。

我们先来看CompactionJob::Prepare函数，在这个函数中主要是做一些执行前的准备工作，首先是取得对应的compact的边界，这里每一个需要并发的compact都被抽象为一个sub compaction.因此在GenSubcompactionBoundaries会解析到对应的sub compaction以及边界.解析完毕之后，则将会把对应的信息全部加入sub_compact_states中。

  1. void CompactionJob::Prepare() {
  2.   ..........................
  3.   if (c->ShouldFormSubcompactions()) {
  4.     const uint64_t start_micros = env_->NowMicros();
  5.     GenSubcompactionBoundaries();
  6.     MeasureTime(stats_, SUBCOMPACTION_SETUP_TIME,
  7.                 env_->NowMicros() - start_micros);
  9.     assert(sizes_.size() == boundaries_.size() + 1);
  11.     for (size_t i = 0; i <= boundaries_.size(); i++) {
  12.       Slice* start = i == 0 ? nullptr : &boundaries_[i - 1];
  13.       Slice* end = i == boundaries_.size() ? nullptr : &boundaries_[i];
  14.       compact_->sub_compact_states.emplace_back(c, start, end, sizes_[i]);
  15.     }
  16.     MeasureTime(stats_, NUM_SUBCOMPACTIONS_SCHEDULED,
  17.                 compact_->sub_compact_states.size());
  18.   }
  19. ......................................
  20. }

因此我们来详细分析GenSubcompactionBoundaries，这个函数比较长，我们来分开分析，首先是遍历所有的需要compact的level,然后取得每一个level的边界(也就是最大最小key)。

  1. void CompactionJob::GenSubcompactionBoundaries() {
  2. ...........................
  3.   // Add the starting and/or ending key of certain input files as a potential
  4.   // boundary
  5.   for (size_t lvl_idx = 0; lvl_idx < c->num_input_levels(); lvl_idx++) {
  6.     int lvl = c->level(lvl_idx);
  7.     if (lvl >= start_lvl && lvl <= out_lvl) {
  8.       const LevelFilesBrief* flevel = c->input_levels(lvl_idx);
  9.       size_t num_files = flevel->num_files;
  10. .....................
  11.       if (lvl == 0) {
  12.         // For level 0 add the starting and ending key of each file since the
  13.         // files may have greatly differing key ranges (not range-partitioned)
  14.         for (size_t i = 0; i < num_files; i++) {
  15.           bounds.emplace_back(flevel->files[i].smallest_key);
  16.           bounds.emplace_back(flevel->files[i].largest_key);
  17.         }
  18.       } else {
  19.         // For all other levels add the smallest/largest key in the level to
  20.         // encompass the range covered by that level
  21.         bounds.emplace_back(flevel->files[0].smallest_key);
  22.         bounds.emplace_back(flevel->files[num_files - 1].largest_key);
  23.         if (lvl == out_lvl) {
  24.           // For the last level include the starting keys of all files since
  25.           // the last level is the largest and probably has the widest key
  26.           // range. Since it's range partitioned, the ending key of one file
  27.           // and the starting key of the next are very close (or identical).
  28.           for (size_t i = 1; i < num_files; i++) {
  29.             bounds.emplace_back(flevel->files[i].smallest_key);
  30.           }
  31.         }
  32.       }
  33.     }
  34.   }
  35. ......................

然后则是对取得的bounds进行排序以及去重.

  1.   std::sort(bounds.begin(), bounds.end(),
  2.             [cfd_comparator](const Slice& a, const Slice& b) -> bool {
  3.               return cfd_comparator->Compare(ExtractUserKey(a),
  4.                                              ExtractUserKey(b)) < 0;
  5.             });
  6.   // Remove duplicated entries from bounds
  7.   bounds.erase(
  8.       std::unique(bounds.begin(), bounds.end(),
  9.                   [cfd_comparator](const Slice& a, const Slice& b) -> bool {
  10.                     return cfd_comparator->Compare(ExtractUserKey(a),
  11.                                                    ExtractUserKey(b)) == 0;
  12.                   }),
  13.       bounds.end());

接近着就来计算理想情况下所需要的subcompactions的个数以及输出文件的个数.

  1.   // Group the ranges into subcompactions
  2.   const double min_file_fill_percent = 4.0 / 5;
  3.   int base_level = v->storage_info()->base_level();
  4.   uint64_t max_output_files = static_cast<uint64_t>(std::ceil(
  5.       sum / min_file_fill_percent /
  6.       MaxFileSizeForLevel(*(c->mutable_cf_options()), out_lvl,
  7.           c->immutable_cf_options()->compaction_style, base_level,
  8.           c->immutable_cf_options()->level_compaction_dynamic_level_bytes)));
  9.   uint64_t subcompactions =
  10.       std::min({static_cast<uint64_t>(ranges.size()),
  11.                 static_cast<uint64_t>(c->max_subcompactions()),
  12.                 max_output_files});

最后更新boundaries_，这里会根据根据文件的大小，通过平均的size,来吧所有的range分为几份，最终这些都会保存在boundaries_中.

  1.   if (subcompactions > 1) {
  2.     double mean = sum * 1.0 / subcompactions;
  3.     // Greedily add ranges to the subcompaction until the sum of the ranges'
  4.     // sizes becomes >= the expected mean size of a subcompaction
  5.     sum = 0;
  6.     for (size_t i = 0; i < ranges.size() - 1; i++) {
  7.       sum += ranges[i].size;
  8.       if (subcompactions == 1) {
  9.         // If there's only one left to schedule then it goes to the end so no
  10.         // need to put an end boundary
  11.         continue;
  12.       }
  13.       if (sum >= mean) {
  14.         boundaries_.emplace_back(ExtractUserKey(ranges[i].range.limit));
  15.         sizes_.emplace_back(sum);
  16.         subcompactions--;
  17.         sum = 0;
  18.       }
  19.     }
  20.     sizes_.emplace_back(sum + ranges.back().size);
  21.   }

然后我们来看CompactJob::Run的实现,在这个函数中，就是会遍历所有的sub_compact,然后启动线程来进行对应的compact工作，最后等到所有的线程完成，然后退出.

  1.  // Launch a thread for each of subcompactions 1...num_threads-1
  2.   std::vector<port::Thread> thread_pool;
  3.   thread_pool.reserve(num_threads - 1);
  4.   for (size_t i = 1; i < compact_->sub_compact_states.size(); i++) {
  5.     thread_pool.emplace_back(&CompactionJob::ProcessKeyValueCompaction, this,
  6.                              &compact_->sub_compact_states[i]);
  7.   }
  9.   // Always schedule the first subcompaction (whether or not there are also
  10.   // others) in the current thread to be efficient with resources
  11.   ProcessKeyValueCompaction(&compact_->sub_compact_states[0]);
  13.   // Wait for all other threads (if there are any) to finish execution
  14.   for (auto& thread : thread_pool) {
  15.     thread.join();
  16.   }
  18.   if (output_directory_) {
  19.     output_directory_->Fsync();
  20.   }

最后我们可以看到最终compact工作是在CompactionJob::ProcessKeyValueCompaction是实现的，这个函数我们暂时就不分析了，我们只需要知道所有的compact工作都是在这个函数中执行的.