Write
LevelDB提供了write和put两个接口进行插入操作，但是put实际上是调用write实现的，所以我在这里只分析write函数：
Status DBImpl::Write(const WriteOptions &options, WriteBatch *my_batch)首先初始化一个Writer对象，Writer对象用于封装一个插入操作，LevelDB用一个deque来管理Writer对象，新建的Writer对象被插入到这个deque的尾部，如果Writer对象未被处理且不在deque头部，则会一直等待：
Writer w(&mutex_);
w.batch = my_batch;
w.sync = options.sync;
w.done = false;
MutexLock l(&mutex_);
writers_.push_back(&w);
while (!w.done && &w != writers_.front())
{
w.cv.Wait();
}
if (w.done)
{
return w.status;
}然后调用MakeRoomForWrite函数保证memtable中有插入的空间：
// May temporarily unlock and wait.
Status status = MakeRoomForWrite(my_batch == nullptr);
uint64_t last_sequence = versions_->LastSequence();
Writer *last_writer = &w;接下来调用BuildBatchGroup函数将此时writers_队列中的Writer对象全部封装为一个WriteBatch，也就是说LevelDB实际上一次会处理当前的所有插入任务：
if (status.ok() && my_batch != nullptr)
{ // nullptr batch is for compactions
WriteBatch *updates = BuildBatchGroup(&last_writer);
WriteBatchInternal::SetSequence(updates, last_sequence + 1);
last_sequence += WriteBatchInternal::Count(updates);再调用函数将KV值插入memtable中：
// 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();
status = log_->AddRecord(WriteBatchInternal::Contents(updates));
bool sync_error = false;
if (status.ok() && options.sync)
{
status = logfile_->Sync();
if (!status.ok())
{
sync_error = true;
}
}
if (status.ok())
{
status = WriteBatchInternal::InsertInto(updates, mem_);
}
mutex_.Lock();
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);
}将队列中此次已经处理的Writer对象都删除，并且给那些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对象，发送信号激活队首的Writer对象：
// Notify new head of write queue
if (!writers_.empty())
{
writers_.front()->cv.Signal();
}
return status;Write函数调用的MakeRoomForWrite函数为：
// REQUIRES: mutex_ is held
// REQUIRES: this thread is currently at the front of the writer queue
Status DBImpl::MakeRoomForWrite(bool force)函数将一直进行循环，判断各个条件并执行相应操作，直到memtable中有足够空间可以插入。
如果level0的文件数量超过阈值，且这是第一次检测到这种情况，那么sleep1ms：
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.
mutex_.Unlock();
env_->SleepForMicroseconds(1000);
allow_delay = false; // Do not delay a single write more than once
mutex_.Lock();
}如果当前memtable中有足够的空间，则跳出循环：
else if (!force &&
(mem_->ApproximateMemoryUsage() <= options_.write_buffer_size))
{
// There is room in current memtable
break;
}如果当前memtable中空间不足，immutable memtable也没有被写出，则等待compact的背景线程完成compact（immutable memtable需要compact）：
else if (imm_ != nullptr)
{
// 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");
background_work_finished_signal_.Wait();
}如果当前memtable空间不足，level0中的文件数量超过了阈值，且不是第一次检测到这种情况，则等待compact的背景线程完成compact（level0需要compact）：
else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger)
{
// There are too many level-0 files.
Log(options_.info_log, "Too many L0 files; waiting...\n");
background_work_finished_signal_.Wait();
}如果以上情况都不存在，则说明可以将当前memtable写入immutable memtable，然后创建一个新的memtable，当然需要调用MaybeScheduleCompaction函数，因为产生了immutable memtable需要compact：
else
{
// 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 = nullptr;
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_);
mem_->Ref();
force = false; // Do not force another compaction if have room
MaybeScheduleCompaction();
}231 Love u