Open
LevelDB的初始化主要由Open函数实现：
Status DB::Open(const Options &options, const std::string &dbname,
DB **dbptr)首先，Open函数调用Recover函数将LevelDB的历史状态恢复：
*dbptr = nullptr;
DBImpl *impl = new DBImpl(options, dbname);
impl->mutex_.Lock();
VersionEdit edit;
// Recover handles create_if_missing, error_if_exists
bool save_manifest = false;
Status s = impl->Recover(&edit, &save_manifest);如果恢复后LevelDB当前的memtable为空则创建一个memtable和相应的log文件：
if (s.ok() && impl->mem_ == nullptr)
{
// Create new log and a corresponding memtable.
uint64_t new_log_number = impl->versions_->NewFileNumber();
WritableFile *lfile;
s = options.env->NewWritableFile(LogFileName(dbname, new_log_number),
&lfile);
if (s.ok())
{
edit.SetLogNumber(new_log_number);
impl->logfile_ = lfile;
impl->logfile_number_ = new_log_number;
impl->log_ = new log::Writer(lfile);
impl->mem_ = new MemTable(impl->internal_comparator_);
impl->mem_->Ref();
}
}如果需要将新建的manifest文件保存下来，就调用LogAndApply函数将当前的VersionEdit对象应用，并在
if (s.ok() && save_manifest)
{
edit.SetPrevLogNumber(0); // No older logs needed after recovery.
edit.SetLogNumber(impl->logfile_number_);
s = impl->versions_->LogAndApply(&edit, &impl->mutex_);
}最后删除旧文件并尝试进行compaction：
if (s.ok())
{
impl->DeleteObsoleteFiles();
impl->MaybeScheduleCompaction();
}
impl->mutex_.Unlock();
if (s.ok())
{
assert(impl->mem_ != nullptr);
*dbptr = impl;
}
else
{
delete impl;
}
return s;Recover函数用于恢复LevelDB，包括恢复VersionSet和memtable：
Status DBImpl::Recover(VersionEdit *edit, bool *save_manifest)创建存储LevelDB数据的目录（如果出现错误则会被忽略，可能已经存在）：
mutex_.AssertHeld();
// Ignore error from CreateDir since the creation of the DB is
// committed only when the descriptor is created, and this directory
// may already exist from a previous failed creation attempt.
env_->CreateDir(dbname_);
assert(db_lock_ == nullptr);
Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);
if (!s.ok())
{
return s;
}如果CURRENT文件不存在（当前LevelDB为新创建时），调用NewDB函数创建LevelDB所需的相关文件（manifest文件和CURRENT文件）：
if (!env_->FileExists(CurrentFileName(dbname_)))
{
if (options_.create_if_missing)
{
s = NewDB();
if (!s.ok())
{
return s;
}
}
else
{
return Status::InvalidArgument(
dbname_, "does not exist (create_if_missing is false)");
}
}
else
{
if (options_.error_if_exists)
{
return Status::InvalidArgument(
dbname_, "exists (error_if_exists is true)");
}
}调用versions_->Recover函数根据manifest文件恢复VersionSet：
s = versions_->Recover(save_manifest);
if (!s.ok())
{
return s;
}
SequenceNumber max_sequence(0);获取之前尚未处理的log文件：
// Recover from all newer log files than the ones named in the
// descriptor (new log files may have been added by the previous
// incarnation without registering them in the descriptor).
//
// Note that PrevLogNumber() is no longer used, but we pay
// attention to it in case we are recovering a database
// produced by an older version of leveldb.
const uint64_t min_log = versions_->LogNumber();
const uint64_t prev_log = versions_->PrevLogNumber();
std::vector<std::string> filenames;
s = env_->GetChildren(dbname_, &filenames);
if (!s.ok())
{
return s;
}
std::set<uint64_t> expected;
versions_->AddLiveFiles(&expected);
uint64_t number;
FileType type;
std::vector<uint64_t> logs;
for (size_t i = 0; i < filenames.size(); i++)
{
if (ParseFileName(filenames[i], &number, &type))
{
expected.erase(number);
if (type == kLogFile && ((number >= min_log) || (number == prev_log)))
logs.push_back(number);
}
}
if (!expected.empty())
{
char buf[50];
snprintf(buf, sizeof(buf), "%d missing files; e.g.",
static_cast<int>(expected.size()));
return Status::Corruption(buf, TableFileName(dbname_, *(expected.begin())));
}通过调用RecoverLogFile函数，按顺序根据这些log文件恢复LevelDB的memtable：
// Recover in the order in which the logs were generated
std::sort(logs.begin(), logs.end());
for (size_t i = 0; i < logs.size(); i++)
{
s = RecoverLogFile(logs[i], (i == logs.size() - 1), save_manifest, edit,
&max_sequence);
if (!s.ok())
{
return s;
}
// The previous incarnation may not have written any MANIFEST
// records after allocating this log number. So we manually
// update the file number allocation counter in VersionSet.
versions_->MarkFileNumberUsed(logs[i]);
}
if (versions_->LastSequence() < max_sequence)
{
versions_->SetLastSequence(max_sequence);
}
return Status::OK();RecoverLogFile函数用于根据log文件恢复memtable：
Status DBImpl::RecoverLogFile(uint64_t log_number, bool last_log,
bool *save_manifest, VersionEdit *edit,
SequenceNumber *max_sequence)首先打开log文件：
// Open the log file
std::string fname = LogFileName(dbname_, log_number);
SequentialFile *file;
Status status = env_->NewSequentialFile(fname, &file);
if (!status.ok())
{
MaybeIgnoreError(&status);
return status;
}接下来创建一个LogReporter对象用于读取log文件：
// Create the log reader.
LogReporter reporter;
reporter.env = env_;
reporter.info_log = options_.info_log;
reporter.fname = fname.c_str();
reporter.status = (options_.paranoid_checks ? &status : nullptr);
// We intentionally make log::Reader do checksumming even if
// paranoid_checks==false so that corruptions cause entire commits
// to be skipped instead of propagating bad information (like overly
// large sequence numbers).
log::Reader reader(file, &reporter, true /*checksum*/,
0 /*initial_offset*/);
Log(options_.info_log, "Recovering log #%llu",
(unsigned long long)log_number);然后读取log文件中的日志记录，将KV值插入memtable中，如果memtable的大小超过设定的阈值，就将其compact到level0：
// Read all the records and add to a memtable
std::string scratch;
Slice record;
WriteBatch batch;
int compactions = 0;
MemTable *mem = nullptr;
while (reader.ReadRecord(&record, &scratch) &&
status.ok())
{
if (record.size() < 12)
{
reporter.Corruption(
record.size(), Status::Corruption("log record too small"));
continue;
}
WriteBatchInternal::SetContents(&batch, record);
if (mem == nullptr)
{
mem = new MemTable(internal_comparator_);
mem->Ref();
}
status = WriteBatchInternal::InsertInto(&batch, mem);
MaybeIgnoreError(&status);
if (!status.ok())
{
break;
}
const SequenceNumber last_seq =
WriteBatchInternal::Sequence(&batch) +
WriteBatchInternal::Count(&batch) - 1;
if (last_seq > *max_sequence)
{
*max_sequence = last_seq;
}
if (mem->ApproximateMemoryUsage() > options_.write_buffer_size)
{
compactions++;
*save_manifest = true;
status = WriteLevel0Table(mem, edit, nullptr);
mem->Unref();
mem = nullptr;
if (!status.ok())
{
// Reflect errors immediately so that conditions like full
// file-systems cause the DB::Open() to fail.
break;
}
}
}
delete file;接着判断这个log文件是否可以继续使用，如果可以继续使用当前log文件，那么如果当前memtable不为空，则这个memtable也可以继续使用：
// See if we should keep reusing the last log file.
if (status.ok() && options_.reuse_logs && last_log && compactions == 0)
{
assert(logfile_ == nullptr);
assert(log_ == nullptr);
assert(mem_ == nullptr);
uint64_t lfile_size;
if (env_->GetFileSize(fname, &lfile_size).ok() &&
env_->NewAppendableFile(fname, &logfile_).ok())
{
Log(options_.info_log, "Reusing old log %s \n", fname.c_str());
log_ = new log::Writer(logfile_, lfile_size);
logfile_number_ = log_number;
if (mem != nullptr)
{
mem_ = mem;
mem = nullptr;
}
else
{
// mem can be nullptr if lognum exists but was empty.
mem_ = new MemTable(internal_comparator_);
mem_->Ref();
}
}
}最后如果这个log文件不能继续使用，那么将当前的memtable进行compact写入level0：
if (mem != nullptr)
{
// mem did not get reused; compact it.
if (status.ok())
{
*save_manifest = true;
status = WriteLevel0Table(mem, edit, nullptr);
}
mem->Unref();
}
return status;