Status DBImpl::Put(const WriteOptions& o, const Slice& key, const Slice& val) {
  return DB::Put(o, key, val);
}

Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {
  WriteBatch batch;
  batch.Put(key, value);
  return Write(opt, &batch);
}

Writer w(&mutex_);
w.batch = my_batch;
w.sync = options.sync;
w.done = false;

struct DBImpl::Writer {
  Status status; // 本次写入结果
  WriteBatch* batch; //本次写入的WriteBatch
  bool sync; //本次写入对应的log是否立刻刷盘
  bool done; //本次写入是否完成
  port::CondVar cv; //条件锁，如果前面有别的写入正在执行，本次写入
  //会Wait，等待正在执行的写入执行完后唤醒它

  explicit Writer(port::Mutex* mu) : cv(mu) { }
};

MutexLock l(&mutex_);
writers_.push_back(&w);
while (!w.done && &w != writers_.front()) {
  w.cv.Wait();
}
if (w.done) {
  return w.status;
}

// May temporarily unlock and wait.
  Status status = MakeRoomForWrite(my_batch == NULL); // 做写入前的各种
  //检查，如是不是该停写，是不是该切memtable，是不是该compact
  uint64_t last_sequence = versions_->LastSequence(); // 取当前最大序列号
  Writer* last_writer = &w; // laster_writer的意思是打包多个write中的最
  //后一个writer
  if (status.ok() && my_batch != NULL) {  // NULL batch is for compactions
    WriteBatch* updates = BuildBatchGroup(&last_writer); // 合并writers里
    //多个writer的WriteBatch，就是上面说的deque里此时阻塞住的writer完成他
    //们的任务，打包后last_writer为该打包中的最后一个writer
    WriteBatchInternal::SetSequence(updates, last_sequence + 1); //本次写
    //入对应的序列号
    last_sequence += WriteBatchInternal::Count(updates); //更新last_sequence，
    //加的个数等于此次是WriteBatch中操作的个数

    // Add to log and apply to memtable.  We can release the lock
    // during this phase since &w is currently responsible for logging
    // and protects against concurrent loggers and concurrent writes
    // into mem_.
    {
      mutex_.Unlock(); //写日志和写memtable时可以放锁，
      //让别的write进入deque，尽可能减少互斥时间
      status = log_->AddRecord(WriteBatchInternal::Contents(updates)); //写日志
      bool sync_error = false;
      if (status.ok() && options.sync) {
        status = logfile_->Sync(); // 这里就是上面说的sync选项，
        //如果为true的话，对日志立刻刷盘
        if (!status.ok()) {
          sync_error = true;
        }
      }
      if (status.ok()) {
        status = WriteBatchInternal::InsertInto(updates, mem_); //将WriteBatch
        //中的操作依次写入memtable中
      }
      mutex_.Lock(); //再次加锁，需要互斥的更新versions的last_sequence
      if (sync_error) {
        // The state of the log file is indeterminate: the log record we
        // just added may or may not show up when the DB is re-opened.
        // So we force the DB into a mode where all future writes fail.
        RecordBackgroundError(status);
      }
    }
    if (updates == tmp_batch_) tmp_batch_->Clear();

    versions_->SetLastSequence(last_sequence);
  }
  //从deque的第一个writer开始pop，第一个writer一定是当前操作的writer，
  //直到pop到本次打包的最后一个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;
  }

  // Notify new head of write queue
  //唤醒此时的deque头writer，从它开始进行接下来的写入
  if (!writers_.empty()) {
    writers_.front()->cv.Signal();
  }

  return status;

Status DBImpl::MakeRoomForWrite(bool force) {
  mutex_.AssertHeld(); //此次是占有锁的
  assert(!writers_.empty());
  bool allow_delay = !force;
  Status s;
  while (true) {
    if (!bg_error_.ok()) { //如果此时已经发生了错误，直接返回错误
      // Yield previous error
      s = bg_error_;
      break;
    } else if (
        allow_delay &&
        versions_->NumLevelFiles(0) >= config::kL0_SlowdownWritesTrigger) {
      // We are getting close to hitting a hard limit on the number of
      // L0 files.  Rather than delaying a single write by several
      // seconds when we hit the hard limit, start delaying each
      // individual write by 1ms to reduce latency variance.  Also,
      // this delay hands over some CPU to the compaction thread in
      // case it is sharing the same core as the writer.
      //如果此时level 0中的文件个数已经大于等于kL0_SlowdownWritesTrigger，
      //放锁，sleep一秒来降低写入速度，给level 0的compact留出时间
      mutex_.Unlock();
      env_->SleepForMicroseconds(1000);
      allow_delay = false;  // Do not delay a single write more than once
      mutex_.Lock();
    } else if (!force &&
               (mem_->ApproximateMemoryUsage() <= options_.write_buffer_size)) {
      //如果此时memtable还有剩余空间，则不需要申请新的memtable，直接返回就好了
      // There is room in current memtable
      break;
    } else if (imm_ != NULL) {
      // 如果此时immutable memtable不为空，代表正在flush，此时memtable也满了，
      // 只能停写等待flush结束，结束后会被唤醒，注意这里不放锁
      // We have filled up the current memtable, but the previous
      // one is still being compacted, so we wait.
      Log(options_.info_log, "Current memtable full; waiting...\n");
      bg_cv_.Wait();
    } else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger) {
      //如果此时level 0的文件个数大于等于config::kL0_StopWritesTrigger，停写，
      //注意，这里不放锁，等compact结束后被唤醒
      // There are too many level-0 files.
      Log(options_.info_log, "Too many L0 files; waiting...\n");
      bg_cv_.Wait();
    } else {
      // memtable已经满了并且此时没有immutable memtable，切memtable成imm并生成
      //新的memtable，切日志
      // Attempt to switch to a new memtable and trigger compaction of old
      assert(versions_->PrevLogNumber() == 0);
      uint64_t new_log_number = versions_->NewFileNumber();
      WritableFile* lfile = NULL;
      s = env_->NewWritableFile(LogFileName(dbname_, new_log_number), &lfile);
      if (!s.ok()) {
        // Avoid chewing through file number space in a tight loop.
        versions_->ReuseFileNumber(new_log_number);
        break;
      }
      delete log_;
      delete logfile_;
      logfile_ = lfile;
      logfile_number_ = new_log_number;
      log_ = new log::Writer(lfile);
      imm_ = mem_;
      has_imm_.Release_Store(imm_);
      mem_ = new MemTable(internal_comparator_); // 生成新的memtable
      mem_->Ref();
      force = false;   // Do not force another compaction if have room
      MaybeScheduleCompaction(); //检查是否需要做compact，需要的话给后台
      //任务线程发送compact任务
    }
  }
  return s;