compaction流程

1.首先找score最高的level，如果level的score>=1，則將這個level進行compaction

2.根據(jù)不同策略，從level中選擇一個sst文件進行compact，對于level0，由于sst文件之間(minkey,maxkey)有重疊，所以可能有多個。

3.從level中選出的文件，我們能計算出(minkey,maxkey)

4.從level+1中選出與(minkey,maxkey)有重疊的sst文件

5.多個sst文件進行歸并排序，合并寫出到sst文件

6.根據(jù)壓縮策略，對寫出的sst文件進行壓縮

7.合并結(jié)束后，利用VersionEdit更新VersionSet，更新統(tǒng)計信息

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;
}

//如果有未調(diào)度的compaction（unscheduled_compactions_），且設置的compaction任務數(shù)還未到達限制，則調(diào)度新的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_）
//調(diào)用：1、只要新的versionset被install （cfd）
//     2、resume db的時候    （所有cfd）
//     3、ReleaseSnapshot   （所有cfs）
//NeedsCompaction 根據(jù)不同的策略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的分數(shù)，如果有大于等于1的就說明需要compaction
    if (vstorage->CompactionScore(i) >= 1) {
      return true;
    }
  }
  return false;
}

Rocksdb如何計算compaction的score

//核心函數(shù)CompactionScore涉及兩個變量,這兩個變量分別保存了level以及每個level所對應的score(score越高compact優(yōu)先級越高)，而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)以及文件個數(shù)(num_sorted_runs).
      for (auto* f : files_[level]) {
        if (!f->being_compacted) {
          total_size += f->compensated_file_size;
          num_sorted_runs++;
        }
      }
      //如果是universal，那么把所有l(wèi)evel的文件加進來（當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 {
        //用文件個數(shù)除以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則是反過來算，根據(jù)最大的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;
        }
         //動態(tài)設置有點復雜，不看了
      }
      }
    }
    //這兩個數(shù)組會在NeedsCompaction函數(shù)中被查詢
    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何時調(diào)用？

當有文件變動時，會調(diào)用version::LogAndApply->ProcessManifestWrites->AppendVersion->ComputeCompactionScore