文章目錄

• Open
• Get
• Put、Delete、Write

Open

數(shù)據(jù)庫 Open 操作主要用于創(chuàng)建新的 LevelDB 數(shù)據(jù)庫或打開一個(gè)已存在的數(shù)據(jù)庫。Open 操作的主要函數(shù)共需傳遞 3 個(gè)參數(shù)：兩個(gè)輸入?yún)?shù) options 與 dbname，一個(gè)輸出參數(shù) dbptr。

首先我們來看看它的代碼：

// https://github.com/google/leveldb/blob/master/db/db_impl.ccStatus DB::Open(const Options& options, const std::string& dbname, DB** dbptr) {*dbptr = nullptr;//初始化dbimplDBImpl* impl = new DBImpl(options, dbname);impl->mutex_.Lock();VersionEdit edit;//嘗試恢復(fù)之前已經(jīng)存在的數(shù)據(jù)庫文件中的數(shù)據(jù)bool save_manifest = false;Status s = impl->Recover(&edit, &save_manifest);//判斷Memtable是否為空if (s.ok() && impl->mem_ == nullptr) {//創(chuàng)建新的Log和MemTableuint64_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文件if (s.ok() && save_manifest) {edit.SetPrevLogNumber(0); edit.SetLogNumber(impl->logfile_number_);//生成新的版本s = impl->versions_->LogAndApply(&edit, &impl->mutex_);}if (s.ok()) {//請(qǐng)理無用的文件impl->RemoveObsoleteFiles();//嘗試進(jìn)行Compactionimpl->MaybeScheduleCompaction();}impl->mutex_.Unlock();if (s.ok()) {assert(impl->mem_ != nullptr);*dbptr = impl;} else {delete impl;}return s; }

具體的實(shí)現(xiàn)流程如下圖所示：

Open執(zhí)行流程

初始化一個(gè) DBImpl 的對(duì)象 impl，將相關(guān)的參數(shù)選項(xiàng) options 與數(shù)據(jù)庫名稱 dbname 作為構(gòu)造函數(shù)的參數(shù)。

調(diào)用 DBImpl 對(duì)象的 Recover 函數(shù)，嘗試恢復(fù)之前存在的數(shù)據(jù)庫文件數(shù)據(jù)。

進(jìn)行 Recover 操作后，判斷 impl 對(duì)象中的 MemTable 對(duì)象指針 mem_ 是否為空，如果為空，則進(jìn)入第 4 步，不為空則進(jìn)入第 5 步。

創(chuàng)建新的 Log 文件以及對(duì)應(yīng)的 MemTable 對(duì)象。這一步主要分別實(shí)例化 log::Writer 和 MemTable 兩個(gè)對(duì)象，并賦值給 impl 中對(duì)應(yīng)的成員變量，后續(xù)通過 impl 中的成員變量操作 Log 文件和 MemTable。

判斷是否需要保存 Manifest 相關(guān)信息，如果需要，則保存相關(guān)信息。

判斷前面步驟是否都成功了，如果成功，則調(diào)用 DeleteObsoleteFiles 函數(shù)對(duì)一些過時(shí)文件進(jìn)行刪除，且調(diào)用 MaybeScheduleCompaction 函數(shù)嘗試進(jìn)行數(shù)據(jù)文件的 Compaction 操作。

Get

Get 主要用于從 LevelDB 中獲取對(duì)應(yīng)的鍵-值對(duì)數(shù)據(jù)，它是單個(gè)數(shù)據(jù)讀取的主要接口。Get 的主要參數(shù)為數(shù)據(jù)讀參數(shù)選項(xiàng) options、鍵 key，以及一個(gè)用于返回?cái)?shù)據(jù)值的 string 類型指針 value。其代碼如下：

// https://github.com/google/leveldb/blob/master/db/db_impl.ccStatus DBImpl::Get(const ReadOptions& options, const Slice& key,std::string* value) {Status s;MutexLock l(&mutex_);SequenceNumber snapshot;//獲取序列號(hào)并賦值給snapshotif (options.snapshot != nullptr) {snapshot =static_cast<const SnapshotImpl*>(options.snapshot)->sequence_number();} else {snapshot = versions_->LastSequence();}MemTable* mem = mem_;MemTable* imm = imm_;Version* current = versions_->current();mem->Ref();if (imm != nullptr) imm->Ref();current->Ref();bool have_stat_update = false;Version::GetStats stats;{mutex_.Unlock();//首先查找memtableLookupKey lkey(key, snapshot);if (mem->Get(lkey, value, &s)) {//如果查找不到，接著查找immutable} else if (imm != nullptr && imm->Get(lkey, value, &s)) {//如果還是沒找到，則繼續(xù)查找SSTable} else {s = current->Get(options, lkey, value, &stats);have_stat_update = true;}mutex_.Lock();}if (have_stat_update && current->UpdateStats(stats)) {MaybeScheduleCompaction();}mem->Unref();if (imm != nullptr) imm->Unref();current->Unref();return s; }

具體的實(shí)現(xiàn)流程如下圖所示：

Get執(zhí)行流程

Get 在查詢讀取數(shù)據(jù)時(shí)，依次從 MemTable、Immutable MemTable 以及當(dāng)前保存的 SSTable 文件中進(jìn)行查找。如果在 MemTabel 中找到，立即返回對(duì)應(yīng)的數(shù)值，如果沒有找到，再從 Immutable MemTable 中查找。而如果Immutable MemTable 中還是沒有找到，則會(huì)從持久化的文件 SSTable 中查找，直到找出該鍵對(duì)應(yīng)的數(shù)值為止。

SequenceNumber 有什么用呢？

其主要作用是對(duì) DB 的整個(gè)存儲(chǔ)空間進(jìn)行時(shí)間刻度上的序列備份，即要從 DB 中獲取某一個(gè)數(shù)據(jù)，不僅需要其對(duì)應(yīng)的鍵 key，而且需要其對(duì)應(yīng)的時(shí)間序列號(hào)。對(duì)數(shù)據(jù)庫進(jìn)行寫操作會(huì)改變序列號(hào)，每進(jìn)行一次寫操作，則序列號(hào)加 1。

Put、Delete、Write

Put 主要有3個(gè)參數(shù)：寫操作參數(shù) opt、操作數(shù)據(jù)的 key 與操作數(shù)據(jù)新值 value。其代碼如下：

// https://github.com/google/leveldb/blob/master/db/db_impl.ccStatus DBImpl::Put(const WriteOptions& o, const Slice& key, const Slice& val) {return DB::Put(o, key, val); }Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {WriteBatch batch;batch.Put(key, value);return Write(opt, &batch); }

從上面的代碼可以看出， Put 其實(shí)也是將單條數(shù)據(jù)的操作變更為一個(gè)批量操作，然后調(diào)用 Write 進(jìn)行實(shí)現(xiàn)。

Delete 不會(huì)直接刪除數(shù)據(jù)，而是在對(duì)應(yīng)位置插入一個(gè) key 的刪除標(biāo)志，然后在后續(xù)的 Compaction 過程中才最終去除這條 key-value 記錄。其代碼如下:

// https://github.com/google/leveldb/blob/master/db/db_impl.ccStatus DBImpl::Delete(const WriteOptions& options, const Slice& key) {return DB::Delete(options, key); }Status DB::Delete(const WriteOptions& opt, const Slice& key) {WriteBatch batch;batch.Delete(key);return Write(opt, &batch); }

從上面的代碼可以看出 Delete 的本質(zhì)其實(shí)也是一個(gè) Write 操作。

在介紹 Write 之前，首先介紹其封裝的消息結(jié)構(gòu) Writer 與任務(wù)隊(duì)列 writes_。

Writer 用于保存基本信息，如批量操作 batch、狀態(tài)信息 status、是否同步 sync、是否完成 done 以及用于多線程操作的條件變量cv 。

// https://github.com/google/leveldb/blob/master/db/db_impl.ccstruct DBImpl::Writer {explicit Writer(port::Mutex* mu): batch(nullptr), sync(false), done(false), cv(mu) {}Status status; //狀態(tài)WriteBatch* batch; //批量寫入對(duì)象bool sync;//表示是否已經(jīng)同步了bool done;//表示是否已經(jīng)處理完成port::CondVar cv;//這個(gè)是條件變量 };

接著看看任務(wù)隊(duì)列 writers_，該隊(duì)列對(duì)象中的元素節(jié)點(diǎn)為 Writer 對(duì)象指針?？梢?writes_ 與寫操作的緩存空間有關(guān)，批量操作請(qǐng)求均存儲(chǔ)在這個(gè)隊(duì)列中，按順序執(zhí)行，已完成的出隊(duì)，而未執(zhí)行的則在這個(gè)隊(duì)列中處于等待狀態(tài)。

std::deque<Writer*> writers_ GUARDED_BY(mutex_);

writers_隊(duì)列示意圖

Write 主要有兩個(gè)參數(shù)：WriteOptions 對(duì)象與 WriteBatch 對(duì)象。WriteOptions 主要包含一些關(guān)于寫操作的參數(shù)選項(xiàng)，而WriteBatch對(duì)象，相當(dāng)于一個(gè)緩沖區(qū)，用于定義、保存一系列的批量操作。其代碼實(shí)現(xiàn)如下：

// https://github.com/google/leveldb/blob/master/db/db_impl.ccStatus DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {//實(shí)例化一個(gè)Writer對(duì)象b并插入writers_隊(duì)列中等待執(zhí)行Writer w(&mutex_);w.batch = updates;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;}Status status = MakeRoomForWrite(updates == nullptr);uint64_t last_sequence = versions_->LastSequence();Writer* last_writer = &w;if (status.ok() && updates != nullptr) { //合并寫入操作WriteBatch* write_batch = BuildBatchGroup(&last_writer);WriteBatchInternal::SetSequence(write_batch, last_sequence + 1);last_sequence += WriteBatchInternal::Count(write_batch);{mutex_.Unlock();//將更新寫入日志文件中，并且將日志文件寫入磁盤中status = log_->AddRecord(WriteBatchInternal::Contents(write_batch));bool sync_error = false;if (status.ok() && options.sync) {status = logfile_->Sync();if (!status.ok()) {sync_error = true;}}//將更新寫入Memtable中if (status.ok()) {status = WriteBatchInternal::InsertInto(write_batch, mem_);}mutex_.Lock();if (sync_error) {RecordBackgroundError(status);}}if (write_batch == tmp_batch_) tmp_batch_->Clear();versions_->SetLastSequence(last_sequence);}//由于和并寫入操作一次可能會(huì)處理多個(gè)writer_隊(duì)列中的元素，因此將所有已經(jīng)處理的元素狀態(tài)進(jìn)行變更，并且發(fā)送signal信號(hào)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;}//通知writers_隊(duì)列中的第一個(gè)元素，發(fā)送signal信號(hào)if (!writers_.empty()) {writers_.front()->cv.Signal();}return status; }

具體的實(shí)現(xiàn)流程如下圖所示：

Write執(zhí)行流程

• 實(shí)例化一個(gè) Writer 對(duì)象，并將其插入所示的 writers_ 隊(duì)列中。

• 通過 Writer 中的條件變量 cv 調(diào)用 wait 方法將該線程掛起，等待其他線程發(fā)送 signal 信號(hào)，并且等待隊(duì)列前面的 Writer 操作全部執(zhí)行完畢：

  • 如果線程收到了 signal 信號(hào)：則解除阻塞。
  • 如果線程沒有收到了 signal 信號(hào)：說明隊(duì)列前面仍有其他的 Writer 操作，那么該線程會(huì)再次調(diào)用 wait 方法實(shí)現(xiàn)阻塞，從而保證了 Writer 操作按照隊(duì)列生成次序執(zhí)行。

• 當(dāng)輪到本線程操作時(shí)，首先通過 MakeRoomForWrite 函數(shù)進(jìn)行內(nèi)存空間分配。

• 當(dāng)獲取到需要的內(nèi)存后，根據(jù)一系列的批量操作，對(duì) Log 文件以及 MemTable 分別進(jìn)行更新。

• 依據(jù)批量操作的數(shù)目更新 SequenceNumber。

• 通過 Writer 中的條件變量 cv 發(fā)送 signal 信號(hào)，以通知處于等待狀態(tài)的其他線程開始執(zhí)行。

總結(jié)

以上是生活随笔為你收集整理的LevelDB 源码剖析（九）DBImpl模块：Open、Get、Put、Delete、Write的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

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