LevelDB源码分析-Write

Posted 2020-12-28 yunanlong

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...
");
            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...
");
            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