深圳企业网站制作报价,青岛公司网站,网站建设项目需求书,win安装wordpressSST Compaction 文章目录 SST Compaction什么是Compact SST什么时候触发Compact SST第一处 读取 Key 的时候第二处 使用迭代器遍历数据库时第三处 写入 Key 时第四处 刚打开数据库时 如何进行Compact SST计算Compaction范围进行Compaction创建一个迭代器用于读取Compaction范围…SST Compaction 文章目录 SST Compaction什么是Compact SST什么时候触发Compact SST第一处 读取 Key 的时候第二处 使用迭代器遍历数据库时第三处 写入 Key 时第四处 刚打开数据库时 如何进行Compact SST计算Compaction范围进行Compaction创建一个迭代器用于读取Compaction范围内的Key-Value通过迭代器遍历每个Key-Value丢弃或者写入新的SST LevelDB中有两种Compaction一种是Compact MemTable另一种是Compact SST。Compact MemTable是将MemTable落盘为SST文件Compact SST是将多个SST文件合并为一个SST文件。
本章讲述的是Compact SST的过程。
什么是Compact SST
相比于Compact MemTableCompact SST复杂的多。首先看一下一次SST Compaction的示意图。 Level-0 中浅蓝色的三个 SST 文件加上 Level-1 中的绿色的 SST 文件这 4 个文件进行了合并输出两个新的 SST 文件替换原有的 SST 文件。
什么时候触发Compact SST
Compaction 的入口为MaybeScheduleCompaction()MaybeScheduleCompaction()里面会判断是需要Compact SST还是Compact MemTable。
我们来看下有哪些地方调用了MaybeScheduleCompaction()。
第一处 读取 Key 的时候
我们调用DBImpl::Get()读取某个Key的时候LevelDB 会按照MemTable Immutable MemTable SST的顺序查找如果在MemTable或者Immutable MemTable中找到了那么就不会触发 Compaction。但如果Key是在 SST 中找到的这个 SST 的allowed_seeks就会减 1。当allowed_seeks为 0 时就表示这个 SST 需要Compact了。
所以在DBImpl::Get()中如果是从 SST 中查找的 Key就需要调用一下MaybeScheduleCompaction()尝试触发 Compaction。
Status DBImpl::Get(const ReadOptions options, const Slice key, std::string* value) {// ...bool have_stat_update false;Version::GetStats stats;{mutex_.Unlock();LookupKey lkey(key, snapshot);if (mem-Get(lkey, value, s)) {// 从 MemTable 中查找成功// 不会触发 Compaction。} else if (imm ! nullptr imm-Get(lkey, value, s)) {// 从 Immutable MemTable 中查找成功// 也不会触发 Compaction。} else {// 如果查找 SST 了有可能会触发 Compactions current-Get(options, lkey, value, stats);have_stat_update true;}mutex_.Lock();}// 如果是从 SST 中查找的 Key并且该 SST 的 Seek 次数// 已经超过了阈值那么就会触发 Compaction。if (have_stat_update current-UpdateStats(stats)) {MaybeScheduleCompaction();}// ...return s;
}第二处 使用迭代器遍历数据库时
每当调用一次it-Next()或者it-Prev()移动迭代器时迭代器内部都会调用一次DBIter::ParseKey()将当前Key解析出来。
而在DBIter::ParseKey()中会定期采样当前 Key看看这个Key是否存在于多个SST中。如果是的话就会将这个Key所在的SST的allowed_seeks减 1然后调用MaybeScheduleCompaction()尝试触发 Compaction。
这样做的目的是定期检查SST中的Key是否存在于多个SST中如果是的话就通过Compaction将这个Key所在的SST合并到更高 Level 的SST中这样就可以减少SST的数量提高读取效率。
inline bool DBIter::ParseKey(ParsedInternalKey* ikey) {Slice k iter_-key();// 当一个 iterator 已读取的数据大小超过 bytes_until_read_sampling_ 后// 就会用当前 key 采一次样查看这个 key.user_key 是否存在于多个(两个及以上) SST // 中。如果是的话就把 key 所在的 SST.allowed_seeks 减 1然后调用// MaybeScheduleCompaction() 尝试触发 Compaction。size_t bytes_read k.size() iter_-value().size();while (bytes_until_read_sampling_ bytes_read) {bytes_until_read_sampling_ RandomCompactionPeriod();db_-RecordReadSample(k);}assert(bytes_until_read_sampling_ bytes_read);bytes_until_read_sampling_ - bytes_read;if (!ParseInternalKey(k, ikey)) {status_ Status::Corruption(corrupted internal key in DBIter);return false;} else {return true;}
}void DBImpl::RecordReadSample(Slice key) {MutexLock l(mutex_);if (versions_-current()-RecordReadSample(key)) {MaybeScheduleCompaction();}
}第三处 写入 Key 时
写入Key(进MemTable)之前会在DBImpl::MakeRoomForWrite()里检查MemTable是否已满。如果已满就会调用MaybeScheduleCompaction()尝试触发 Compaction。此处的Compaction指的是Compact MemTable在此就不详细赘述了忘记的同学可以回头参考大白话解析LevelDB 2: MemTable 落盘为 SST 文件。
第四处 刚打开数据库时
在DBImpl::Open()中会调用MaybeScheduleCompaction()尝试触发Compaction。
刚打开数据的时候为什么需要尝试触发Compaction呢
因为当数据库上次关闭时可能还有些没完成的Compaction比如Compaction进行中途机器断电了。
所以当数据库打开时需要尝试触发一次Compaction检查下有没有未完成的Compaction。
Status DB::Open(const Options options, const std::string dbname, DB** dbptr) {// ...// 读取数据库文件恢复数据库状态。Status s impl-Recover(edit, save_manifest);// ...if (s.ok()) {impl-RemoveObsoleteFiles();// 当数据库关闭时可能有些还没完成的 Compaction。// 所以打开数据库时尝试触发一次 Compaction检查// 下有没有未完成的 Compaction。impl-MaybeScheduleCompaction();}// ...return s;
}如何进行Compact SST
在MaybeScheduleCompaction()中会通过versions_-NeedsCompaction()判断是否满足Compaction条件。
若条件满足会将background_compaction_scheduled_标志位设置为true然后将DBImpl::BGWork()加入线程池中在后台线程中进行Compaction。
void DBImpl::MaybeScheduleCompaction() {mutex_.AssertHeld();if (background_compaction_scheduled_) {// 已经有 Compaction 在后台线程中执行了。} else if (shutting_down_.load(std::memory_order_acquire)) {// 数据库正在被关闭不再进行 Compaction。} else if (!bg_error_.ok()) {// 存在错误不再进行 Compaction。} else if (imm_ nullptr manual_compaction_ nullptr !versions_-NeedsCompaction()) {// 不满足 Compaction 条件} else {// 满足 Compaction 条件把 Compaction Job 加入到后台线程池中。background_compaction_scheduled_ true;env_-Schedule(DBImpl::BGWork, this);}
}DBImpl::BGWork()只是层包装最终会调用到DBImpl::BackgroundCompaction()也就是Compaction的实现函数。
DBImpl::BackgroundCompaction()里先计算出本次Compaction的范围然后调用DoCompactionWork()进行Compaction。
void DBImpl::BackgroundCompaction() {mutex_.AssertHeld();// 先判断是否存在 Immutable MemTable如果存在// 就将本次 Compaction 判定为 MemTable Compaction。if (imm_ ! nullptr) {CompactMemTable();return;}// 否则为判定为 SST Compaction进入 SST Compaction 的流程。Compaction* c;bool is_manual (manual_compaction_ ! nullptr);InternalKey manual_end;// 如果是用户主动发起的手动 Compaction本次 Compaction 的范围// 是由用户指定的需要从 manual_compaction_ 中读取而不是由 LevelDB 计算得出。// 否则本次 Compaction 的范围由PickCompaction()计算得出。// 无论是手动 Compaction 还是自动 Compaction最终都会把 Compaction 所// 涉及的 SST 文件编号记录到Compaction* c对象中。// c-inputs_[0] 中存放的是 Compaction Level 所涉及的 SST 文件编号。// c-inputs_[1] 中存放的是 Compaction Level1 所涉及的 SST 文件编号。if (is_manual) {ManualCompaction* m manual_compaction_;c versions_-CompactRange(m-level, m-begin, m-end);m-done (c nullptr);if (c ! nullptr) {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()));} else {// 由 leveldb 计算 Compaction 范围c versions_-PickCompaction();}Status status;if (c nullptr) {// 经过上面的计算发现本次 Compaction 不需要进行直接返回。} else if (!is_manual c-IsTrivialMove()) {// IsTrivialMove() 表示本次 Compaction 只需要简单地将 SST 文件从// level 层移动到 level1 层即可不需要进行 SST 文件合并。// 在 TrivialMove 的情况下level 层需要 Compaction 的 SST 文件// 只能有一个。assert(c-num_input_files(0) 1);// 获取 level 层需要 Compaction 的第 0 个 SST 文件的元数据信息。 FileMetaData* f c-input(0, 0);// 编辑 VersionEdit将 level 层需要 Compact 的 SST 从 level 层移除// 并将其添加到 level1 层。c-edit()-RemoveFile(c-level(), f-number);c-edit()-AddFile(c-level() 1, f-number, f-file_size, f-smallest, f-largest);// Apply 该 VersionEdit将其应用到当前 VersionSet 中。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_castunsigned long long(f-number), c-level() 1,static_castunsigned long long(f-file_size), status.ToString().c_str(),versions_-LevelSummary(tmp));} else {// 需要进行 SST 文件合并的 Compaction。// 构造一个 CompactionState 对象用于记录本次 Compaction // 需要新生成的 SST 信息。CompactionState* compact new CompactionState(c);// 进行真正的 SST Compaction 操作将 Compaction SST// 文件合并生成新的 SST 文件。status DoCompactionWork(compact);if (!status.ok()) {RecordBackgroundError(status);}// Compaction 结束后的清理工作工作。CleanupCompaction(compact);// Compaction 完成后Input Version 就不再需要了将其释放。c-ReleaseInputs();// 移除数据库中不再需要的文件。RemoveObsoleteFiles();}delete c;if (status.ok()) {// 没有异常不需要进行任何异常处理。} else if (shutting_down_.load(std::memory_order_acquire)) {// 如果当前正在关闭数据库那错误就先不需要处理了留到下次打开// 数据时再处理先以最快的时间关闭数据库。} else {Log(options_.info_log, Compaction error: %s, status.ToString().c_str());}if (is_manual) {ManualCompaction* m manual_compaction_;if (!status.ok()) {// Compaction 失败了需要把 m-done 标记为 true// 防止重复 Compact 该范围。m-done true;}if (!m-done) {// Compaction 完成需要把 m-begin 更新为本次// Compaction 的结尾以便下次继续 Compact。m-tmp_storage manual_end;m-begin m-tmp_storage;}manual_compaction_ nullptr;}
}计算Compaction范围
如果是手动触发的Compaction那么初始范围由用户指定最终通过versions_-CompactRange()计算出Compaction的范围。如果是自动触发的Compaction那么最终通过versions_-PickCompaction()计算出Compaction的范围。
versions_-CompactRange()的实现可移步参考大白话解析LevelDB: VersionSet versions_-PickCompaction()的实现可移步参考大白话解析LevelDB: VersionSet
进行Compaction
如果计算出的Compaction范围是nullptr表示当前不需要进行Compaction直接返回。如果计算出的Compaction范围符合TrivialMove条件表示只需要将SST文件从level层移动到level1层即可不需要进行SST文件合并。否则的话就需要进行SST文件合并DoCompactionWork(c)将Compaction范围内的SST文件合并为一个新的SST文件。
DoCompactionWork(CompactionState* compact)的实现如下。
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 nullptr);assert(compact-outfile nullptr);// 将当前存活着的最小 snapshot 版本号记录到compact-smallest_snapshot中。// Compaction 的时候会将 Sequence Number 小于compact-smallest_snapshot的// Key 都扔掉因为已经没有 snapshot 需要这些 Key 了。if (snapshots_.empty()) {compact-smallest_snapshot versions_-LastSequence();} else {compact-smallest_snapshot snapshots_.oldest()-sequence_number();}// 创建一个 Iterator 来读取所有的 Compaction SST。Iterator* input versions_-MakeInputIterator(compact-compaction);// 开始 Compaction在这期间可以先释放 mutex_让其他线程可以继续读写数据库。mutex_.Unlock();input-SeekToFirst();Status status;ParsedInternalKey ikey;std::string current_user_key;bool has_current_user_key false;SequenceNumber last_sequence_for_key kMaxSequenceNumber;// Compaction 的过程会比较耗时每处理完一个 Key 就检查下是否正在关闭数据库。// 如果检查到正在关闭数据库就先终止 Compaction等下次打开数据库的时候再继续 Compaction。while (input-Valid() !shutting_down_.load(std::memory_order_acquire)) {// 如果检测到有 Immutable MemTable 存在优先处理 Immutable MemTable// 把 Immutable MemTable 转成 SST 文件。if (has_imm_.load(std::memory_order_relaxed)) {const uint64_t imm_start env_-NowMicros();mutex_.Lock();if (imm_ ! nullptr) {CompactMemTable();// Compact MemTable 完成后唤醒在MakeRoomForWrite()中等待的线程。background_work_finished_signal_.SignalAll();}mutex_.Unlock();imm_micros (env_-NowMicros() - imm_start);}Slice key input-key();// 检查本次 Compaction 的范围是否太大如果太大的话就缩小本次的 Compaction 范围// Compact 到 key 为止即可后面的 key 等下次 Compaction 的时候再处理。if (compact-compaction-ShouldStopBefore(key) compact-builder ! nullptr) {status FinishCompactionOutputFile(compact, input);if (!status.ok()) {break;}}bool drop false;if (!ParseInternalKey(key, ikey)) {// 如果解析 Key 失败那可能是数据损坏的情况。// 把坏掉的 Key 保留下来写到新 SST 里让用户// 来处理。current_user_key.clear();has_current_user_key false;last_sequence_for_key kMaxSequenceNumber;} else {if (!has_current_user_key ||user_comparator()-Compare(ikey.user_key, Slice(current_user_key)) ! 0) {current_user_key.assign(ikey.user_key.data(), ikey.user_key.size());has_current_user_key true;last_sequence_for_key kMaxSequenceNumber;}// 当 key 的类型为 Deletion 时// 或者 key 的 Sequence Number 小于 compact-smallest_snapshot// 这个 key 就会被丢弃。// 举个例子:// compact-smallest_snapshot 4// 读到的 key 为: key1-seq5, key1-seq4, key1-seq3, key1-seq2, key1-seq1。// key1-seq5 和 key1-seq4 会被保留写入到新 SST 中// 剩下的 key1-seq3, key1-seq2, key1-seq1 都会被丢弃。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)) {drop true;}last_sequence_for_key ikey.sequence;}
#if 0Log(options_.info_log, Compact: %s, seq %d, type: %d %d, drop: %d, is_base: %d, %d smallest_snapshot: %d,ikey.user_key.ToString().c_str(),(int)ikey.sequence, ikey.type, kTypeValue, drop,compact-compaction-IsBaseLevelForKey(ikey.user_key),(int)last_sequence_for_key, (int)compact-smallest_snapshot);
#endif// 如果当前 key 不需要被丢弃就把它写入到新的 SST 中。if (!drop) {// 如果当前没有打开的 Output SST 文件就打开一个新的 Output SST 文件。// 通过 compact-builder 来控制该新 SST 文件的写入。if (compact-builder nullptr) {status OpenCompactionOutputFile(compact);if (!status.ok()) {break;}}// 设置这个 Output SST 文件的最小 key 和最大 key。if (compact-builder-NumEntries() 0) {compact-current_output()-smallest.DecodeFrom(key);}compact-current_output()-largest.DecodeFrom(key);// 往新 SST 中写入当前的 Key-Value。compact-builder-Add(key, input-value());// Output SST 大小达到了阈值就完成该 Ouput SST 的构建重新开一个新的 SST。if (compact-builder-FileSize() compact-compaction-MaxOutputFileSize()) {status FinishCompactionOutputFile(compact, input);if (!status.ok()) {break;}}}input-Next();}// Compaction 和关闭数据库撞上了终止 Compaction。if (status.ok() shutting_down_.load(std::memory_order_acquire)) {status Status::IOError(Deleting DB during compaction);}// 如果还有没收尾的 SST做个收尾。if (status.ok() compact-builder ! nullptr) {status FinishCompactionOutputFile(compact, input);}// 检查 input 迭代器有木有错误。if (status.ok()) {status input-status();}delete input;input nullptr;// 记录下本次 Compaction 所花费的时间。CompactionStats stats;stats.micros env_-NowMicros() - start_micros - imm_micros;// 记录下本次 Compaction 一共读取了多少字节。for (int which 0; which 2; which) {for (int i 0; i compact-compaction-num_input_files(which); i) {stats.bytes_read compact-compaction-input(which, i)-file_size;}}// 记录本次 Compaction 一共写入了多少字节。for (size_t i 0; i compact-outputs.size(); i) {stats.bytes_written compact-outputs[i].file_size;}mutex_.Lock();// 记录本次 Compaction 被推到哪个 level.stats_[compact-compaction-level() 1].Add(stats);// 将本次 Compaction 产生的改动(删除和新增了哪些 SST)生成一个 VersionEdit// 并应用到最新的 Version 上产生一个新的 Version。if (status.ok()) {status InstallCompactionResults(compact);}if (!status.ok()) {RecordBackgroundError(status);}VersionSet::LevelSummaryStorage tmp;Log(options_.info_log, compacted to: %s, versions_-LevelSummary(tmp));return status;
}创建一个迭代器用于读取Compaction范围内的Key-Value
通过Iterator* input versions_-MakeInputIterator(compact-compaction)创建一个迭代器该迭代器可按顺序读取Compaction范围所有内的Key-Value。
通过迭代器遍历每个Key-Value丢弃或者写入新的SST
leveldb 会先读取当前最小的Snapshot版本号记录到compact-smallest_snapshot中。
如果当前Key的版本号小于compact-smallest_snapshot或者Key的类型为Deletion那么这个Key就会被丢弃。
否则的话将当前Key写入新的SST。
