Status DBImpl::Write(const WriteOptions& options,WriteBatch* updates) 
{
  Status status;
  MutexLock l(&mutex_);  // 锁定互斥体，同一时间只能有一个线程更新数据
  LoggerId self;   
  // 获取Logger的使用权,如果有其他线程拥有所有权，则等待至其释放所有权。
  AcquireLoggingResponsibility(&self);
  status = MakeRoomForWrite(false);  // May temporarily release lock and wait
  uint64_t last_sequence = versions_->LastSequence();  // 获取当前的版本号
  if (status.ok()) {
    // 将当前版本号加1后作为本次更新的日志的版本，
    // 一次批量更新可能包含多个操作，这些操作都用一个版本有一个好处：
    // 本次更新的所有操作，要么都可见，要么都不可见，不存在一部分可见，另一部分不可见的情况。
    WriteBatchInternal::SetSequence(updates,last_sequence + 1);
    // 但是本次更新可能有多个操作，跳过与操作数相等的版本号，保证不被使用
    last_sequence += WriteBatchInternal::Count(updates);

    // 将batch写入log，然后应用到memtable中
    {
      assert(logger_ == &self);
      mutex_.Unlock();
      // 这里，可以解锁，因为在AcquireLoggingResponsibility()方法中已经获取了Logger的拥有权，
      // 其他线程即使获得了锁，但是由于&self != logger，其会阻塞在AcquireLoggingResponsibility()方法中。
      // 将更新写入log文件，如果设置了每次写入进行sync，则将其同步到磁盘，这个操作可能比较长，
      // 防止了mutex_对象长期被占用，因为其还负责其他一些资源的同步
      status = log_->AddRecord(WriteBatchInternal::Contents(updates));
      if (status.ok() && options.sync) {
        status = logfile_->Sync();
      }
      if (status.ok()) {
        // 成功写入了log后，才写入memtable
        status = WriteBatchInternal::InsertInto(updates,mem_);
      }
      // 重新锁定mutex_
      mutex_.Lock();
      assert(logger_ == &self);
    }
    // 更新版本号
    versions_->SetLastSequence(last_sequence);
  }
  // 释放对logger的所有权，并通知等待的线程，然后解锁
  ReleaseLoggingResponsibility(&self);
 return status;
}

// force参数表示强制新起一个memtable
Status DBImpl::MakeRoomForWrite(bool force) {
 mutex_.AssertHeld();
 assert(logger_ != NULL);
 bool allow_delay = !force;
 Status s;
 while (true) {
  if (!bg_error_.ok()) {
   // 后台线程存在问题，则返回错误，不接受更新
   s = bg_error_;
   break;
  } else if (
    allow_delay &&
    versions_->NumLevelFiles(0) >= config::kL0_SlowdownWritesTrigger) {
      // 如果不是强制写入，而且level 0的sstable超过8个，则本次更新阻塞1毫秒，
      // leveldb将sstable分为多个等级，其中level 0中的不同表的key是可能重叠的，
      // 如果l0的sstable过多，会导致查询性能下降，这时需要适当降低更新速度，让
      // 后台线程进行compaction操作，但是设计者不希望让某次写操作等待数秒，
      // 而是让每次更新操作分担延迟，即每次写操作阻塞1毫秒，平衡读写速率；
      // 另外，理论上这也能让compaction线程获得更多的cpu时间（当然，
      // 这是假定compaction与更新操作共享一个CPU时才有意义）
   mutex_.Unlock();
   env_->SleepForMicroseconds(1000);
   allow_delay = false; // 最多延迟一次，下次不延迟
   mutex_.Lock();
  } else if (!force &&
       (mem_->ApproximateMemoryUsage() <= options_.write_buffer_size)) {
   // 如果当前memtable已使用的空间小于write_buffer_size，则跳出，更新到当前memtable即可。
      // 当force为true时，第一次循环会走后面else逻辑，切换了memtable后force被置为false，
      // 第二次循环时就可以在此跳出了
   break;
  } else if (imm_ != NULL) {
      // 如果当前memtable已经超过write_buffer_size,且备用的memtable也在被使用，则阻塞更新并等待
   bg_cv_.Wait();
  } else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger) {
   // 如果当前memtable已使用的空间小于write_buffer_size，但是备用的memtable未被使用，
      // 则检查level 0的sstable个数，如超过12个，则阻塞更新并等待
   Log(options_.info_log,"waiting...n");
   bg_cv_.Wait();
  } else {
   // 否则，使用新的id新创建一个log文件，并将当前memtable切换为备用的memtable，新建一个
      // memtable，然后将数据写入当前的新memtable，即切换log文件与memtable，并告诉后台线程
      // 可以进行compaction操作了
   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()) {
    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_);
   mem_->Ref();
   force = false;  // 下次判断可以不新建memtable了
   MaybeScheduleCompaction();
  }
 }
 return s;
}
void DBImpl::AcquireLoggingResponsibility(LoggerId* self) {
 while (logger_ != NULL) {
  logger_cv_.Wait();
 }
 logger_ = self;
}

void DBImpl::ReleaseLoggingResponsibility(LoggerId* self) {
 assert(logger_ == self);
 logger_ = NULL;
 logger_cv_.SignalAll();
}

2.4 Skip List

template<typename Key,class Comparator>
int SkipList<Key,Comparator>::RandomHeight() {
  // Increase height with probability 1 in kBranching
  static const unsigned int kBranching = 4;
  int height = 1;
  while (height < kMaxHeight && ((rnd_.Next() % kBranching) == 0)) {
    height++;
  }
  return height;
}

int InternalKeyComparator::Compare(const Slice& akey,const Slice& bkey) const {
  int r = user_comparator_->Compare(ExtractUserKey(akey),ExtractUserKey(bkey));
  if (r == 0) {
    // 比较后面8个字节构造的整数，第一个字节的type为Least Significant Byte
    const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);
    const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);
    if (anum > bnum)  // 注意：整数大反而key比较小
    {
      r = -1;
    } else if (anum < bnum) {
      r = +1;
    }
  }
  return r;
}