compaction流程

1.首先找score最高的level，如果level的score>=1，则将这个level进行compaction

2.根据不同策略，从level中选择一个sst文件进行compact，对于level0，由于sst文件之间(minkey,maxkey)有重叠，所以可能有多个。

3.从level中选出的文件，我们能计算出(minkey,maxkey)

4.从level+1中选出与(minkey,maxkey)有重叠的sst文件

5.多个sst文件进行归并排序，合并写出到sst文件

6.根据压缩策略，对写出的sst文件进行压缩

7.合并结束后，利用VersionEdit更新VersionSet，更新统计信息

compaction何时运行？

1、手动compact(RunManualCompaction),

2、自动MaybeScheduleFlushOrCompaction

其中自动compaction中(MaybeScheduleFlushOrCompaction)：

//如果有问题就先不做了
if (bg_compaction_paused_ > 0) {
  // we paused the background compaction
  return;
} else if (error_handler_.IsBGWorkStopped()) {
  // Compaction is not part of the recovery sequence from a hard error. We
  // might get here because recovery might do a flush and install a new
  // super version, which will try to schedule pending compactions. Bail
  // out here and let the higher level recovery handle compactions
  return;
}

//如果有手动的compaction正在进行，就不做后台的compaction了
if (HasExclusiveManualCompaction()) {
  // only manual compactions are allowed to run. don't schedule automatic
  // compactions
  TEST_SYNC_POINT("DBImpl::MaybeScheduleFlushOrCompaction:Conflict");
  return;
}

//如果有未调度的compaction（unscheduled_compactions_），且设置的compaction任务数还未到达限制，则调度新的compaction
while (bg_compaction_scheduled_ < bg_job_limits.max_compactions &&
       unscheduled_compactions_ > 0) {
  CompactionArg* ca = new CompactionArg;
  ca->db = this;
  ca->prepicked_compaction = nullptr;
  bg_compaction_scheduled_++;
  unscheduled_compactions_--;
  env_->Schedule(&DBImpl::BGWorkCompaction, ca, Env::Priority::LOW, this,
                 &DBImpl::UnscheduleCompactionCallback);
}

//SchedulePendingCompaction判断是否有compaction需要进行（更新compaction_queue_和unscheduled_compactions_）
//调用：1、只要新的versionset被install （cfd）
//     2、resume db的时候    （所有cfd）
//     3、ReleaseSnapshot   （所有cfs）
//NeedsCompaction 根据不同的策略pick需要compaction的文件
void DBImpl::SchedulePendingCompaction(ColumnFamilyData* cfd) {
  if (!cfd->queued_for_compaction() && cfd->NeedsCompaction()) {
    AddToCompactionQueue(cfd);
    ++unscheduled_compactions_;
  }
}
//以level compaction为例
//当存在有超期文件。
//files_marked_for_compaction_或者bottommost_files_marked_for_compaction_都不为空
//遍历所有的level的sst,然后判断是否需要compact（计算scope）跟新以上两个队列.
bool LevelCompactionPicker::NeedsCompaction(
    const VersionStorageInfo* vstorage) const {
  if (!vstorage->ExpiredTtlFiles().empty()) {
    return true;
  }
  if (!vstorage->FilesMarkedForPeriodicCompaction().empty()) {
    return true;
  }
  if (!vstorage->BottommostFilesMarkedForCompaction().empty()) {
    return true;
  }
  if (!vstorage->FilesMarkedForCompaction().empty()) {
    return true;
  }
  for (int i = 0; i <= vstorage->MaxInputLevel(); i++) {
    //直接返回每一level的分数，如果有大于等于1的就说明需要compaction
    if (vstorage->CompactionScore(i) >= 1) {
      return true;
    }
  }
  return false;
}

Rocksdb如何计算compaction的score

//核心函数CompactionScore涉及两个变量,这两个变量分别保存了level以及每个level所对应的score(score越高compact优先级越高)，而score小于１则表示不需要compaction.

  std::vector<double> compaction_score_;
  std::vector<int> compaction_level_;

void VersionStorageInfo::ComputeCompactionScore(
    const ImmutableCFOptions& immutable_cf_options,
    const MutableCFOptions& mutable_cf_options) {
  for (int level = 0; level <= MaxInputLevel(); level++) {
    double score;
    if (level == 0) {
      int num_sorted_runs = 0;
      uint64_t total_size = 0;
      //计算level-0下所有文件的大小(total_size)以及文件个数(num_sorted_runs).
      for (auto* f : files_[level]) {
        if (!f->being_compacted) {
          total_size += f->compensated_file_size;
          num_sorted_runs++;
        }
      }
      //如果是universal，那么把所有level的文件加进来（当l0对待）
      if (compaction_style_ == kCompactionStyleUniversal) {
        // For universal compaction, we use level0 score to indicate
        // compaction score for the whole DB. Adding other levels as if
        // they are L0 files.
        for (int i = 1; i < num_levels(); i++) {
          if (!files_[i].empty() && !files_[i][0]->being_compacted) {
            num_sorted_runs++;
          }
        }
      }

      if (compaction_style_ == kCompactionStyleFIFO) {
        //总大小/最大文件大小
        score = static_cast<double>(total_size) /
                mutable_cf_options.compaction_options_fifo.max_table_files_size;
        if (mutable_cf_options.compaction_options_fifo.allow_compaction) {
          score = std::max(
              static_cast<double>(num_sorted_runs) /
                  mutable_cf_options.level0_file_num_compaction_trigger,
              score);
        }
        //超期的文件和score比较
        if (mutable_cf_options.ttl > 0) {
          score = std::max(
              static_cast<double>(GetExpiredTtlFilesCount(
                  immutable_cf_options, mutable_cf_options, files_[level])),
              score);
        }

      } else {
        //用文件个数除以level0_file_num_compaction_trigger来得到对应的score
        score = static_cast<double>(num_sorted_runs) /
                mutable_cf_options.level0_file_num_compaction_trigger;
        if (compaction_style_ == kCompactionStyleLevel && num_levels() > 1) {
          // Level-based involves L0->L0 compactions that can lead to oversized
          // L0 files. Take into account size as well to avoid later giant
          // compactions to the base level.
          //当前不止一层level,则score和(total_size/max_bytes_for_level_base)取最大值.
          score = std::max(
              score, static_cast<double>(total_size) /
                     mutable_cf_options.max_bytes_for_level_base);
        }
      }
    } else {
      //非l0的得分为未compaction文件大小/这层允许的最大大小
      // Compute the ratio of current size to size limit.
      uint64_t level_bytes_no_compacting = 0;
      for (auto f : files_[level]) {
        if (!f->being_compacted) {
          level_bytes_no_compacting += f->compensated_file_size;
        }
      }
      //options.max_bytes_for_level_base
      //max_bytes_for_level_base=1,max_bytes_for_level_multiplier=10,那么L1, L2,L3的大小限制分别为1,10,100.
      //level_compaction_dynamic_level_bytes则是反过来算，根据最大的level的大小往前推
      //Target_Size(Ln-1) = Target_Size(Ln) / max_bytes_for_level_multiplier
      //VersionStorageInfo::CalculateBaseBytes初始化
      score = static_cast<double>(level_bytes_no_compacting) /
              MaxBytesForLevel(level);
      //计算过程
      {
        if (!ioptions.level_compaction_dynamic_level_bytes) {
             base_level_ = (ioptions.compaction_style == kCompactionStyleLevel) ? 1 : -1;

        // Calculate for static bytes base case
        for (int i = 0; i < ioptions.num_levels; ++i) {
        if (i == 0 && ioptions.compaction_style == kCompactionStyleUniversal) {
            level_max_bytes_[i] = options.max_bytes_for_level_base;
        } else if (i > 1) {
           //非level0，那么这一层就等于上一层*options.max_bytes_for_level_multiplier
           level_max_bytes_[i] = MultiplyCheckOverflow(
            MultiplyCheckOverflow(level_max_bytes_[i - 1],
                                  options.max_bytes_for_level_multiplier),
            options.MaxBytesMultiplerAdditional(i - 1));
        } else {
          //如果是CompactionStylelevel的且level=0 / level1，那么level_max_bytes=options.max_bytes_for_level_base
          level_max_bytes_[i] = options.max_bytes_for_level_base;
        }
         //动态设置有点复杂，不看了
      }
      }
    }
    //这两个数组会在NeedsCompaction函数中被查询
    compaction_level_[level] = level;
    compaction_score_[level] = score;
  }

  // sort all the levels based on their score. Higher scores get listed
  // first. Use bubble sort because the number of entries are small.
  //冒泡排序，把得分高的房前面
  for (int i = 0; i < num_levels() - 2; i++) {
    for (int j = i + 1; j < num_levels() - 1; j++) {
      if (compaction_score_[i] < compaction_score_[j]) {
        double score = compaction_score_[i];
        int level = compaction_level_[i];
        compaction_score_[i] = compaction_score_[j];
        compaction_level_[i] = compaction_level_[j];
        compaction_score_[j] = score;
        compaction_level_[j] = level;
      }
    }
  }
  ...
}

ComputeCompactionScore何时调用？

当有文件变动时，会调用version::LogAndApply->ProcessManifestWrites->AppendVersion->ComputeCompactionScore