LevelDB原理解析：数据的读写与合并是怎样发生的？

Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {
  WriteBatch batch;
  batch.Put(key, value);
  //调用DBImpl::Write方法  
  return Write(opt, &batch);
}
struct DBImpl::Writer {
  Status status;
  WriteBatch* batch;
  bool sync;
  bool done;
  port::CondVar cv;
  explicit Writer(port::Mutex* mu) : cv(mu) { }
};

Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
  Writer w(&mutex_);
  w.batch = updates;
  w.sync = options.sync;
  w.done = false;
  MutexLock l(&mutex_);
  writers_.push_back(&w);
  //任务放到queue中，如果当前不是queue的头部则等待
  //当某个线程将queue中自己对应的Writer写入磁盘时，可能也会将其他线程对应的Writer写入磁盘
  while (!w.done && &w != writers_.front()) {
    w.cv.Wait();
  }
  if (w.done) {
    return w.status;
  }

Status DBImpl::MakeRoomForWrite(bool force) {
      //通过改变指针指向，将Memtable转换成Immutable
      imm_ = mem_;
      has_imm_.store(true, std::memory_order_release);
      //生成新的Memtable
      mem_ = new MemTable(internal_comparator_);
      mem_->Ref();
      //触发compaction
      MaybeScheduleCompaction();
}

//从队列中批量取任务 
WriteBatch* write_batch = BuildBatchGroup(&last_writer);
//将任务写入Log文件
status = log_->AddRecord(WriteBatchInternal::Contents(write_batch));
//将任务写入Memtable
status = WriteBatchInternal::InsertInto(write_batch, mem_);

  //last_writer指向writers_里合并的最后一个Writer
  //逐个遍历弹出writers_里的元素，并唤醒对应的正在等待的写线程，直到遇到last_writer
  while (true) {
    Writer* ready = writers_.front();
    writers_.pop_front();
    if (ready != &w) {
      ready->status = status;
      ready->done = true;
      ready->cv.Signal();
    }
    if (ready == last_writer) break;
  }
  // 唤醒队列未写入的第一个Writer
  if (!writers_.empty()) {
    writers_.front()->cv.Signal();
  }

{
    mutex_.Unlock();
    // First look in the memtable, then in the immutable memtable (if any).
    LookupKey lkey(key, snapshot);
    //先查找memtable
    if (mem->Get(lkey, value, &s)) {
    //再查找immutable memtable
    } else if (imm != nullptr && imm->Get(lkey, value, &s)) {
    } else {
      //查找sstable
      s = current->Get(options, lkey, value, &stats);
      have_stat_update = true;
    }
    mutex_.Lock();
  }

Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit,
                                Version* base) {
  //生成sstable编号，用于构建文件名
  FileMetaData meta;
  meta.number = versions_->NewFileNumber();
  Status s;
  {
    mutex_.Unlock();
    //更新memtable中的全部数据到xxx.ldb文件
    //meta记录key range, file_size等sst信息
    s = BuildTable(dbname_, env_, options_, table_cache_, iter, &meta);
    mutex_.Lock();
  }
  int level = 0;
  if (s.ok() && meta.file_size > 0) {
    const Slice min_user_key = meta.smallest.user_key();
    const Slice max_user_key = meta.largest.user_key();
    if (base != nullptr) {
      //为新生成的sstable选择合适的level(不一定总是0）
      level = base->PickLevelForMemTableOutput(min_user_key, max_user_key);
    }
    //level及file meta记录到edit
    edit->AddFile(level, meta.number, meta.file_size, meta.smallest,
                  meta.largest);
  }
}

//记录edit信息
versions_->LogAndApply(&edit, &mutex_);
//释放imm_空间
imm_->Unref();
imm_ = nullptr;
has_imm_.Release_Store(nullptr);
//清理无用文件
DeleteObsoleteFiles();

void VersionSet::Finalize(Version* v) {
  // Precomputed best level for next compaction
  int best_level = -1;
  double best_score = -1
  for (int level = 0; level < config::kNumLevels - 1; level++) {
    double score;
    //对于level 0使用文件数/4计算score
    if (level == 0) {
      score = v->files_[level].size() /
              static_cast<double>(config::kL0_CompactionTrigger);
    } else {
      //对于非0层，使用该层文件的总大小
      // Compute the ratio of current size to size limit.
      const uint64_t level_bytes = TotalFileSize(v->files_[level]);
      score =
          static_cast<double>(level_bytes) / MaxBytesForLevel(options_, level);
    }
    if (score > best_score) {
      best_level = level;
      best_score = score;
    }
  }
 //使用compaction_level_记录需要合并的层，使用compaction_score_记录合并分数
  v->compaction_level_ = best_level;
  v->compaction_score_ = best_score;
}

bool Version::UpdateStats(const GetStats& stats) {
  FileMetaData* f = stats.seek_file;
  if (f != nullptr) {
    f->allowed_seeks--;
    if (f->allowed_seeks <= 0 && file_to_compact_ == nullptr) {
      file_to_compact_ = f;
      file_to_compact_level_ = stats.seek_file_level;
      return true;
    }
  }
  return false;
}

Iterator* VersionSet::MakeInputIterator(Compaction* c) {
  const int space = (c->level() == 0 ? c->inputs_[0].size() + 1 : 2);
  // list存储所有Iterator
  Iterator** list = new Iterator*[space];
  int num = 0;
  for (int which = 0; which < 2; which++) {
    if (!c->inputs_[which].empty()) {
      //第0层
      if (c->level() + which == 0) {
        const std::vector<FileMetaData*>& files = c->inputs_[which];
        // Iterator* Table::NewIterator
        for (size_t i = 0; i < files.size(); i++) {
          list[num++] = table_cache_->NewIterator(
              options, files[i]->number, files[i]->file_size);
        }
      } else {
        // Create concatenating iterator for the files from this level
        list[num++] = NewTwoLevelIterator(
            // 遍历文件列表的iterator
            new Version::LevelFileNumIterator(icmp_, &c->inputs_[which]),
            &GetFileIterator, table_cache_, options);
      }
    }
  }