[leveldb] 3.put/delete操做

0.導讀

LevelPut的流程：

 

Put操做首先將操做記錄寫入log文件，而後寫入memtable，返回寫成功。總體來看是這樣，可是會引起下面的問題：

1. 寫log的時候是實時刷到磁盤的嗎？ 2. 寫入的時候memtable過大了咋辦？ 3. 同時多個線程併發寫咋辦？ ......

 

 

下面的分析中就會面對這些問題，有些答案很清晰，有些涉及到超級多細節。

step by step

在開始以前，咱們知道Write操做是要記錄到log文件中的。那麼一個記錄它的格式是怎樣的呢？看圖：

 

這裏特別解釋一下，Delete操做也是經過Put實現的，只是圖中的類型字段是0，而正常Write的操做類型是1，由此區分寫操做和刪除操做！

熟悉了LevelDB整個脈絡以後, Put方法是至關簡單的, 一章就能夠解決. 數據刪除和寫入是一個概念, 刪除就是寫入特殊deletion marker; 批量(batch)和單條寫入也是一個概念, 單條寫入就是隻有一條記錄的batch. 整個流程很短, 基本上寫個log就行了, 所以速度很快.

step 1

 

Put interface

很簡單吧。這裏講解一下那三個參數：

1. WriteOptions：提供一些寫操做的配置項，例如要不要寫log的時候立刻flush磁盤
2. key和value就是對應的keyValue，slice只是做者本身封裝的char數組存儲數據而已。<b>大牛喜歡把全部東西都封裝一下，賦予數據結構意義！</b>這在大的工程裏面是頗有意義的，既方便操做，也方便思考（這樣就不用思考底層的真實的char數組仍是啥）。

Put對於多線程的處理很是精妙, 主體在DBImpl::Write函數中.

插入：

Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {
  WriteBatch batch;        //leveldb中無論單個插入仍是多個插入都是以WriteBatch的方式進行的
  batch.Put(key, value);
  return Write(opt, &batch);
}

一條記錄包含以下內容： 
Type、Key、Value 
當要插入記錄時，Type爲kTypeValue，當要刪除記錄時，Type爲kTypeDeletion，同時中每個batch都有一個對當前批處理記錄信息的統計（sequence（8字節）和count（4字節），共12字節） 
因而可知，當咱們要刪除一個數據時，並非直接從內存中刪除，而是插入一條帶有刪除標誌的記錄

在本例中要插入數據：key=」lili」; value=」hihi」; 
由以前對WriterBatch的分析可知，獲得的batch爲： 
01 00 00 00 00 00 00 00 01 00 00 00 （前8字節表示是第一個batch，後4字節表示此batch中只有一條記錄） 
01(kTypeValue) 04(Key.size) 6C 69 6C 69(lili) 04(value.size) 68 69 68 69(hihi) 
共12+1+1+4+1+4=23字節=0x17

Delete也相似，只是調用了WriteBatch 的 Delete(key)， 這樣再內部會以不一樣的形式編碼傳遞至下一步進行處理。具體的WriteBatch的實現和編碼方式在稍後的文章中進行介紹。Delete和Put都調用了Write，，這裏的Write是在DBImpl::Write中經過虛函數的形式實現對其調用的，咱們接着看Write的流程

 1 Status DBImpl::Write(const WriteOptions& options, WriteBatch* my_batch) {
 2   // A begin
 3   Writer w(&mutex_);
 4   w.batch = my_batch;
 5   w.sync = options.sync;
 6   w.done = false;
 7   // A end
 8  
 9   // B begin
/*
 mutex l上鎖以後, 到了"w.cv.Wait()"的時候, 會先釋放鎖等待, 而後收到signal時再次上鎖. 這段代碼的做用就是多線程在提交任務的時候,
 一個接一個push_back進隊列. 但只有位於隊首的線程有資格繼續運行下去. 目的是把多個寫請求合併成一個大batch提高效率.
*/

10   MutexLock l(&mutex_);
11   writers_.push_back(&w);
12   while (!w.done && &w != writers_.front()) {
13     w.cv.Wait();
14   }
15   if (w.done) {
16     return w.status;
17   }
18   // B end
19  
20   // May temporarily unlock and wait.
21   Status status = MakeRoomForWrite(my_batch == NULL);
22   uint64_t last_sequence = versions_->LastSequence();
23   Writer* last_writer = &w;
24   if (status.ok() && my_batch != NULL) {  // NULL batch is for compactions
25     WriteBatch* updates = BuildBatchGroup(&last_writer);
26     WriteBatchInternal::SetSequence(updates, last_sequence + 1);
27     last_sequence += WriteBatchInternal::Count(updates);
28  
29     // Add to log and apply to memtable.  We can release the lock
30     // during this phase since &w is currently responsible for logging
31     // and protects against concurrent loggers and concurrent writes
32     // into mem_.
33     {
34       mutex_.Unlock();
35       status = log_->AddRecord(WriteBatchInternal::Contents(updates));
36       bool sync_error = false;
37       if (status.ok() && options.sync) {
38         status = logfile_->Sync();
39         if (!status.ok()) {
40           sync_error = true;
41         }
42       }
43       if (status.ok()) {
44         status = WriteBatchInternal::InsertInto(updates, mem_);
45       }
46       mutex_.Lock();
47       if (sync_error) {
48         // The state of the log file is indeterminate: the log record we
49         // just added may or may not show up when the DB is re-opened.
50         // So we force the DB into a mode where all future writes fail.
51         RecordBackgroundError(status);
52       }
53     }
54     if (updates == tmp_batch_) tmp_batch_->Clear();
55  
56     versions_->SetLastSequence(last_sequence);
57   }
58  
59   while (true) {
60     Writer* ready = writers_.front();
61     writers_.pop_front();
62     if (ready != &w) {
63       ready->status = status;
64       ready->done = true;
65       ready->cv.Signal();
66     }
67     if (ready == last_writer) break;
68   }
69  
70   // Notify new head of write queue
71   if (!writers_.empty()) {
72     writers_.front()->cv.Signal();
73   }
74  
75   return status;
76 }

因此從流程能夠清晰的看到插入刪除的流程主要爲：

1. 將這條KV記錄以順序寫的方式追加到log文件末尾；

2. 將這條KV記錄插入內存中的Memtable中，在插入過程當中若是恰好後臺進程在compaction會短暫停頓覺得後臺進程compaction騰出時間及cpu

這裏涉及到一次磁盤讀寫操做和內存SkipList的插入操做，可是這裏的磁盤寫時文件的順序追加寫入效率是很高的，因此並不會致使寫入速度的下降；

並且從流程分析咱們知道，在插入（刪除）過程當中若是多線程同時進行，那麼這些操做將會將操做的同步設置相同的相鄰的操做合併爲一個批插入，這樣可使整個系統的總吞吐量更大。因此一次插入記錄操做只會等待一次磁盤文件追加寫和內存SkipList插入操做，這是爲什麼leveldb寫入速度如此高效的根本緣由。

　　假設同時有w1, w2, w3, w4, w5, w6 併發請求寫入。

　　B部分代碼讓競爭到mutex資源的w1獲取了鎖。w1將它要寫的數據添加到了writers_隊列裏去，此時隊列只有一個w1, 從而其順利的進行buildbatchgroup。當運行到34行時mutex_互斥鎖釋放，之因此這兒能夠釋放mutex_，是由於其它的寫操做都不知足隊首條件，進而不會進入log和memtable寫入階段。這時(w2, w3, w4, w5, w6)會競爭鎖，因爲B段代碼中不知足隊首條件，均等待並釋放鎖了。從而隊列可能會如(w3, w5, w2, w4).

　　繼而w1進行log寫入和memtable寫入。 當w1完成log和memtable寫入後，進入46行代碼，則mutex_又鎖住，這時B段代碼中隊列由於獲取不到鎖則隊列不會修改。

　　隨後59行開始，w1被pop出來，因爲ready==w, 而且ready==last_writer,因此直接到71行代碼，喚醒了此時處於隊首的w3.

      w3喚醒時，發現本身是隊首，能夠順利的進行進入buildbatchgroup，在該函數中，遍歷了目前全部的隊列元素，造成一個update的batch，即將w3, w5, w2, w4合併爲一個batch. 並將last_writer置爲此時處於隊尾的最後一個元素w4，34行代碼運行後，由於釋放了鎖資源，隊列可能隨着dbimpl::write的調用而更改，如隊列情況可能爲(w3, w5, w2, w4, w6, w9, w8).

 　　35-45行的代碼將w3, w5, w2, w4整個的batch寫入log和memtable. 到65行，分別對w5, w2, w4進行了一次cond signal.當判斷到完w4 == lastwriter時，則退出循環。72行則對隊首的w6喚醒，從而按上述步驟依次進行下去。

　　這樣就造成了多個併發write 合併爲一個batch寫入log和memtable的機制。