文章目錄

• 日志結構
• 讀寫流程
• • 寫入
  • 讀取
• 崩潰恢復

當向 LevelDB 寫入數(shù)據(jù)時，只需要將數(shù)據(jù)寫入內(nèi)存中的 MemTable，而由于內(nèi)存是易失性存儲，因此 LevelDB 需要一個額外的持久化文件：預寫日志（Write-Ahead Log，WAL），又稱重做日志。這是一個追加修改、順序寫入磁盤的文件。當宕機或者程序崩潰時 WAL 能夠保證寫入成功的數(shù)據(jù)不會丟失。將 MemTable 成功寫入 SSTable 后，相應的預寫日志就可以刪除了。

日志結構

Log 文件以塊為基本單位，一條記錄可能全部寫到一個塊上，也可能跨幾個塊。記錄的格式如下圖所示：

Log記錄格式

首先我們來看看 Log 中的數(shù)據(jù)格式，代碼如下：

// https://github.com/google/leveldb/blob/master/db/log_format.hnamespace log {enum RecordType {// Zero is reserved for preallocated fileskZeroType = 0,kFullType = 1,// For fragmentskFirstType = 2,kMiddleType = 3,kLastType = 4};static const int kMaxRecordType = kLastType;static const int kBlockSize = 32768;// Header is checksum (4 bytes), length (2 bytes), type (1 byte).static const int kHeaderSize = 4 + 2 + 1;} // namespace log

結合上面的代碼和圖片，我們可以看到每一個塊大小為 32768 字節(jié)，并且每一個塊由頭部和正文組成。頭部由 4 字節(jié)校驗，2 字節(jié)的長度與 1 字節(jié)的類型構成，即每一個塊的開始 7 字節(jié)屬于頭部。頭部中的類型字段有如下 4 種：

• kZeroType：為預分配的文件保留。
• kFullType：表示一條記錄完整地寫到了一個塊上。
• kFirstType：表示該條記錄的第一部分。
• kMiddleType：表示該條記錄的中間部分。
• kLastType：表示該條記錄的最后一部分。

通過記錄結構可以推測出 Log 文件的讀取流程，即首先根據(jù)頭部的長度字段確定需要讀取多少字節(jié)，然后根據(jù)頭部類型字段確定該條記錄是否已經(jīng)完整讀取，如果沒有完整讀取，繼續(xù)按該流程進行，直到讀取到記錄的最后一部分，其頭部類型為 kLastType。

讀寫流程

寫入

Log 的讀取主要由 Writer 中的 AddRecord 實現(xiàn)，代碼如下：

// https://github.com/google/leveldb/blob/master/db/log_writer.hclass Writer {public:explicit Writer(WritableFile* dest);Writer(WritableFile* dest, uint64_t dest_length);Writer(const Writer&) = delete;Writer& operator=(const Writer&) = delete;~Writer();Status AddRecord(const Slice& slice);private:Status EmitPhysicalRecord(RecordType type, const char* ptr, size_t length);WritableFile* dest_;int block_offset_; // Current offset in blockuint32_t type_crc_[kMaxRecordType + 1]; };// https://github.com/google/leveldb/blob/master/db/log_writer.ccStatus Writer::AddRecord(const Slice& slice) {const char* ptr = slice.data();size_t left = slice.size();Status s;//begin表明本條記錄是第一次寫入，即當前塊中第一條記錄bool begin = true;do {//當前塊剩余空間，用于判斷頭部能否完整寫入const int leftover = kBlockSize - block_offset_;assert(leftover >= 0);if (leftover < kHeaderSize) {//如果塊剩余空間小于七個字節(jié)且不等于0，說明當前無法完整寫入數(shù)據(jù)，此時填充\x00，從下一個塊寫入if (leftover > 0) {static_assert(kHeaderSize == 7, "");dest_->Append(Slice("\x00\x00\x00\x00\x00\x00", leftover));}//此時塊正好寫滿，將block_offset_置為0，表明開始寫入新的塊block_offset_ = 0;}assert(kBlockSize - block_offset_ - kHeaderSize >= 0);//計算塊剩余空間const size_t avail = kBlockSize - block_offset_ - kHeaderSize;//計算當前塊能夠寫入的數(shù)據(jù)大小（塊剩余空間和記錄剩余內(nèi)容中最小的）const size_t fragment_length = (left < avail) ? left : avail;RecordType type;//end表明該記錄是否已經(jīng)完整寫入，即最后一條記錄const bool end = (left == fragment_length);//根據(jù)begin與end來確定記錄類型if (begin && end) {//記錄為第一條且同時又是最后一條，說明當前是完整的記錄，狀態(tài)為kFullTypetype = kFullType;} else if (begin) {//記錄為第一條，狀態(tài)為kFirstTypetype = kFirstType;} else if (end) {//記錄為最后一條，標記狀態(tài)為kLastTypetype = kLastType;} else {//記錄不為第一條，也并非最后一條，則說明是中間狀態(tài)，標記為kMiddleTypetype = kMiddleType;}//將數(shù)據(jù)按照格式寫入，并刷新到磁盤文件中s = EmitPhysicalRecord(type, ptr, fragment_length);ptr += fragment_length;left -= fragment_length;begin = false;} while (s.ok() && left > 0);//循環(huán)至數(shù)據(jù)完全寫入或者寫入失敗時才停止return s; }

寫入流程如下：

判斷頭部能否完整寫入，如果不能則將剩余空間用 \x00 填充，接著從新的塊開始寫入。

根據(jù) begin 和 end 判斷記錄類型。

將數(shù)據(jù)按照格式寫入，并刷新到磁盤文件中。

循環(huán)至數(shù)據(jù)完全寫入或者寫入失敗后停止，將結果返回。

讀取

Log 的讀取主要由 Reader 中的 ReadRecord 實現(xiàn)，代碼如下：

// https://github.com/google/leveldb/blob/master/db/log_reader.hclass Reader {public:// Interface for reporting errors.class Reporter {public:virtual ~Reporter();virtual void Corruption(size_t bytes, const Status& status) = 0;};Reader(SequentialFile* file, Reporter* reporter, bool checksum,uint64_t initial_offset);Reader(const Reader&) = delete;Reader& operator=(const Reader&) = delete;~Reader();bool ReadRecord(Slice* record, std::string* scratch);uint64_t LastRecordOffset();private:enum {kEof = kMaxRecordType + 1,kBadRecord = kMaxRecordType + 2}; };// https://github.com/google/leveldb/blob/master/db/log_reader.ccbool Reader::ReadRecord(Slice* record, std::string* scratch) {if (last_record_offset_ < initial_offset_) {if (!SkipToInitialBlock()) {return false;}}scratch->clear();record->clear();bool in_fragmented_record = false;uint64_t prospective_record_offset = 0;Slice fragment;while (true) {//ReadPhysicalRecord讀取log文件并將記錄保存到fragment，同時返回記錄的類型const unsigned int record_type = ReadPhysicalRecord(&fragment);uint64_t physical_record_offset =end_of_buffer_offset_ - buffer_.size() - kHeaderSize - fragment.size();if (resyncing_) {if (record_type == kMiddleType) {continue;} else if (record_type == kLastType) {resyncing_ = false;continue;} else {resyncing_ = false;}}//根據(jù)記錄的類型來判斷是否需要將當前記錄附加到scratch后并繼續(xù)讀取switch (record_type) {//類型為kFullType則說明當前是完整的記錄，直接賦值給record后返回case kFullType:if (in_fragmented_record) {if (!scratch->empty()) {ReportCorruption(scratch->size(), "partial record without end(1)");}}prospective_record_offset = physical_record_offset;scratch->clear();*record = fragment;last_record_offset_ = prospective_record_offset;return true;//類型為kFirstType則說明當前是第一部分，先將記錄復制到scratch后繼續(xù)讀取case kFirstType:if (in_fragmented_record) {if (!scratch->empty()) {ReportCorruption(scratch->size(), "partial record without end(2)");}}prospective_record_offset = physical_record_offset;scratch->assign(fragment.data(), fragment.size());in_fragmented_record = true;break;//類型為kMiddleType則說明當前是中間部分，先將記錄追加到scratch后繼續(xù)讀取case kMiddleType://初始讀取到的類型為kMiddleType或者kLastType，則需要忽略并且繼續(xù)偏移if (!in_fragmented_record) {ReportCorruption(fragment.size(),"missing start of fragmented record(1)");} else {scratch->append(fragment.data(), fragment.size());}break;//類型為kLastType則說明當前為最后，繼續(xù)追加到scratch，并將scratch賦值給record并返回case kLastType:if (!in_fragmented_record) {ReportCorruption(fragment.size(),"missing start of fragmented record(2)");} else {scratch->append(fragment.data(), fragment.size());*record = Slice(*scratch);last_record_offset_ = prospective_record_offset;return true;}break;//如果狀態(tài)為kEof、kBadRecord時說明日志損壞，此時清空scratch并返回falsecase kEof:if (in_fragmented_record) {scratch->clear();}return false;case kBadRecord:if (in_fragmented_record) {ReportCorruption(scratch->size(), "error in middle of record");in_fragmented_record = false;scratch->clear();}break;//未定義的類型，輸出日志，剩余同上處理default: {char buf[40];std::snprintf(buf, sizeof(buf), "unknown record type %u", record_type);ReportCorruption((fragment.size() + (in_fragmented_record ? scratch->size() : 0)),buf);in_fragmented_record = false;scratch->clear();break;}}}return false; }

執(zhí)行流程如下：

ReadRecord 讀取一條記錄到 fragment 變量中，并且返回該條記錄的類型。

根據(jù)記錄的類型來判斷是否需要將當前記錄附加到 scratch 后并繼續(xù)讀取：

• kFullType：當前是完整的記錄，直接賦值給 record 后返回。
• kFirstType：當前是第一部分，先將記錄覆蓋到 scratch 后繼續(xù)讀取。
• kMiddleType：當前是中間部分，先將記錄追加到 scratch 后繼續(xù)讀取。
• kLastType：當前為最后部分，繼續(xù)追加到 scratch，并將完整的 scratch 賦值給 record 后返回。
• 其它/異常：清空 scratch 并返回 false，如果是未定義類型需要輸出日志。

這里還有一個需要注意的細節(jié)，由于讀取 Log 文件時可以從指定偏移量開始，所以如果初始讀取到的類型為 kMiddleType 或者 kLastType，則需要忽略并且繼續(xù)偏移，直到碰見第一個 kFirstType。

崩潰恢復

當打開一個 LevelDB 的數(shù)據(jù)文件時，需先檢驗是否進行崩潰恢復，如果需要，則會從 Log 文件生成一個MemTable，其實現(xiàn)代碼如下：

// https://github.com/google/leveldb/blob/master/db/db_impl.ccStatus DBImpl::RecoverLogFile(uint64_t log_number, bool last_log,bool* save_manifest, VersionEdit* edit,SequenceNumber* max_sequence) {struct LogReporter : public log::Reader::Reporter {Env* env;Logger* info_log;const char* fname;Status* status; // null if options_.paranoid_checks==falsevoid Corruption(size_t bytes, const Status& s) override {Log(info_log, "%s%s: dropping %d bytes; %s",(this->status == nullptr ? "(ignoring error) " : ""), fname,static_cast<int>(bytes), s.ToString().c_str());if (this->status != nullptr && this->status->ok()) *this->status = s;}};mutex_.AssertHeld();//打開log文件std::string fname = LogFileName(dbname_, log_number);SequentialFile* file;Status status = env_->NewSequentialFile(fname, &file);if (!status.ok()) {MaybeIgnoreError(&status);return status;}//創(chuàng)建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);log::Reader reader(file, &reporter, true /*checksum*/, 0 /*initial_offset*/);Log(options_.info_log, "Recovering log #%llu",(unsigned long long)log_number);//讀取所有的records并寫入一個memtablestd::string scratch;Slice record;WriteBatch batch;int compactions = 0;MemTable* mem = nullptr;//循環(huán)讀取日志文件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();}//將records寫入memtablestatus = 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;}//如果memtable大于閾值，則將其轉換成sstable(默認4MB) if (mem->ApproximateMemoryUsage() > options_.write_buffer_size) {compactions++;*save_manifest = true;status = WriteLevel0Table(mem, edit, nullptr);mem->Unref();mem = nullptr;if (!status.ok()) {break;}}}delete file;//判斷是否應該繼續(xù)重復使用最后一個日志文件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_ = new MemTable(internal_comparator_);mem_->Ref();}}}if (mem != nullptr) {if (status.ok()) {*save_manifest = true;status = WriteLevel0Table(mem, edit, nullptr);}mem->Unref();}return status; }

具體的邏輯如下：

打開 log，創(chuàng)建 log reader 開始讀取數(shù)據(jù)。

循環(huán)讀取日志文件，并將其寫入 MemTable 中。

如果 MemTable 過大，則將其轉換為 SSTable。

總結

以上是生活随笔為你收集整理的LevelDB 源码剖析（六）WAL模块：LOG 结构、读写流程、崩溃恢复的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。