ETCD探索-MVCC

ETCD探索-MVCC

MVCC

MVCC模塊ETCD的存儲模塊，是ETCD核心模塊。

做爲一個開源項目，其代碼的封裝是值得咱們學習的。MVCC做爲底層模塊，對上層提供統一的方法，而這些方法都定義在kv.go這個文件中，很像一個頭文件(.h)。咱們能夠只看kv.go以及配合kv_test.go就能夠知道mvcc包是怎麼用的。

kv.go

type KV interface {  
    ReadView  
    WriteView  
    // Read creates a read transaction.  
    Read(trace *traceutil.Trace) TxnRead  

    // Write creates a write transaction.  
    Write(trace *traceutil.Trace) TxnWrite  

    // Hash computes the hash of the KV's backend.  
    Hash() (hash uint32, revision int64, err error)  

    // HashByRev computes the hash of all MVCC revisions up to a given revision.  
    HashByRev(rev int64) (hash uint32, revision int64, compactRev int64, err error)  

    // Compact frees all superseded keys with revisions less than rev.  
    Compact(trace *traceutil.Trace, rev int64) (<-chan struct{}, error)  

    // Commit commits outstanding txns into the underlying backend.  
    Commit()  

    // Restore restores the KV store from a backend.  
    Restore(b backend.Backend) error  
    Close() error  
}

(我只複製了最重要的Interface，請結合kv.go文件來看)
咱們是這樣來使用KV的

func test() {
    kv := mvcc.New(...)
    kv.Put(key, value, ...
    kv.Range(key, end, ...)  // Range就是Get方法
    ...
}

mvcc有一個New方法，返回ConsistentWatchableKV，它繼承自KV，用來實現ETCD的Watch機制。咱們如今討論KV。

---

做爲一個Key-Value存儲，存儲模塊至少要支持增刪改查。
KV首先定義了

Put(key, value []byte, lease lease.LeaseID) (rev int64)    // 「增」、「改」，WriteView中定義，KV繼承ReadView

Range(key, end []byte, ro RangeOptions) (r *RangeResult, err error)  // 「查」，ReadView中定義

DeleteRange(key, end []byte) (n, rev int64) // 「刪」，WriteView中定義，n是刪除的個數

這三個方法實現了數據的基本操做。
參數很好理解

• key、value
• leaseID 租約，熟悉ETCD都能理解，租約是附着(attach)在KeyValue上的
• rev 版本號，mvcc的體現。（這個後面說）
• end 不一樣於Get()方法，查詢都是範圍查找Range()，查找區間[key, end)。
• RangeOptions查詢參數，RangeResult查詢結果。（這兩個結構體參數很好理解，很少贅述）

同時KV提供了事務操做

Read(trace *traceutil.Trace) TxnRead

Write(trace *traceutil.Trace) TxnWrite

Commit() // 提交未提交的事務

顧名思義，一個讀事務、一個寫事務
參數

• trace 鏈路追蹤用，若是你不瞭解鏈路追蹤，徹底能夠忽略這個參數。這個參數不影響邏輯。
• TxnRead、TxnWrite分別繼承ReadView、WriteView，也就實現了Put、Range等

除了基本數據操做，KV提供了一些不容易理解的方法

Hash() (hash uint32, revision int64, err error)
HashByRev(rev int64) (hash uint32, revision int64, compactRev int64, err error)

Compact(trace *traceutil.Trace, rev int64) (<-chan struct{}, error)

Restore(b backend.Backend) error

這些方法從名字上咱們或許可以知道是在幹什麼，但殊不知道爲何要這麼幹。

• Hash()、HashByRev() 是將當前KV存儲的數據作CRC冗餘校驗，返回uint32 hash值。ETCD很重要的一個特性是可靠，即便出現各類網絡分區現象，也要保證數據一致性，因此就必需要有數據校驗的功能。Hash函數就是爲了知足這個功能。
• Compact() 將給定Rev以前的數據‘壓實’，即以前的數據再沒法訪問。ETCD不會留存全部的數據，長時間運行存儲空間不容許。因此會將舊版本數據Compact。(熟悉Raft的同窗知道Raft的日誌壓縮策略，Raft模塊日誌壓縮時會調用Compact，這塊在Raft模塊會說到)
• Restore() KV能夠從給定的Backend中恢復數據。

(建議讀者閱讀kv_test.go中全部的測試用例，即可以更深入瞭解 KV)
如此，咱們便有了這樣一個輪廓:

MVCC爲其餘模塊提供了這些功能。接下來，咱們就詳細看下這些功能是怎麼實現的。

---

在討論實現過程的時候，我不會仔細介紹每一行代碼，這樣難以理解而且也沒有必要。但我但願讀者可以本身閱讀並跑一遍測試用例。

在看代碼以前，請先試想，若是讓你來作底層的KV實現，你會怎麼作?
你可能會這麼作:

既然是KV存儲，用一個Map來存儲。若是須要持久化存儲，將Map中的數據拷貝一份到DB，Map中保留最新數據。

若是你是這麼想的，那你已經實現了底層功能。你作到了：

• 持久化存儲
• 緩存最近數據(map)
• 快速索引(map自身的hash函數)

或許經過封裝很容易作到：

• Watchable 監聽機制
• Lease 租約機制

但你很難作到：

• 多版本控制MVCC
• 前綴查詢(ETCD的重要能力 --prefix)

而且你這樣作有些不足：

• 索引依賴map的hash函數，難以免大量數據的hash碰撞，下降索引效率

看完這些你有其餘的想法實現上述難以作到的功能嗎？

你可能會放棄使用Map作內存存儲，而使用一種數據結構(B樹或B+樹)來代替。給每一kvpair附帶一個版本號。

咱們試想了這麼多，其實已經猜出了KV底層實現的80%

具體看下

---

mvcc包中，store是KV的具體實現，store支持Put、Range等操做。首先須要關注store中的一個變量:currentRev。它是一個「嚴格」遞增的版本號。「嚴格」是指currentRev的每次遞增都會有鎖。

store的每次寫操做(Put、Delete等)都會使currentRev遞增。currentRev就是Revision

理解這句話很重要。以下圖

緊接着，咱們須要知道value存在哪？value是以鍵值對的形式保存，鍵值對是以KeyValue結構體的方式存儲在store中，咱們須要知道KeyValue是怎麼定義的。

// 直觀的想，KeyValue多是這樣
type KeyValue struct{
    key   string
    value string
}

// 但它其實包含更多的東西
type KeyValue struct {  
    Key []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`  

    // 建立時的store版本號
    CreateRevision int64 `protobuf:"varint,2,opt,name=create_revision,json=createRevision,proto3" json:"create_revision,omitempty"` 
    
    // 最後一次修改的store版本號
    ModRevision int64 `protobuf:"varint,3,opt,name=mod_revision,json=modRevision,proto3" json:"mod_revision,omitempty"`  
    
    // KeyValue自身維護的版本號
    Version int64 `protobuf:"varint,4,opt,name=version,proto3" json:"version,omitempty"`  

    Value []byte `protobuf:"bytes,5,opt,name=value,proto3" json:"value,omitempty"`  

    // 租約ID
    Lease int64 `protobuf:"varint,6,opt,name=lease,proto3" json:"lease,omitempty"`  
}

這個時候，咱們能夠理解，每個KeyValue都對應一個Revision。

若是咱們想查詢一個KeyValue，能夠用Revision作索引

如上圖：
foo=bar建立時Revision=1。那我就能夠用1來找到foo=bar
同理：

• 3能夠找到foo1=bar1
• 4能夠找到r=b
• 6能夠找到foo=bar3

這麼作的意義在哪？在於它保存了KeyValue的全部變化，即便最後foo=bar5。我依然能夠找到foo以前的值。這就是mvcc多版本控制的根本所在。

這個時候咱們得出了用Revision找到KeyValue。但實際狀況咱們是用key去找到KeyValue。

那麼接下來的問題是怎麼用key找到Revision。

store中使用BTree實現key快速找到Revision。

(BTree具體實現能夠閱讀github.com/google/btree)

接下來咱們就能大概繪出這樣一幅輪廓圖：

其中Backend是KeyValue的存儲，也是咱們以後要討論的。
對應關係以下

當咱們想要查key爲foo對應的值時，咱們能夠指定Revision開始查詢(若是不指定，默認從最新開始查詢)

---

咱們找幾處代碼來論證以上所述。
首先找到store的Put具體實現。

// store -> storeTxnWrite -> Put -> put
func put(...) {
    ...
    tw.tx.UnsafeSeqPut(keyBucketName, ibytes, d)  // Backend 添加 Revision->KeyValue
    tw.s.kvindex.Put(key, idxRev)  // BTree Index 添加key->Revision索引
    ...
}

再看store的Range的具體實現

// store -> storeTxnRead -> Range -> rangeKeys
func rangeKeys(...) {
    ...
    revpairs := tr.s.kvindex.Revisions(key, end, rev) // 在Index中，用key查找Revision
    ...
    _, vs := tr.tx.UnsafeRange(keyBucketName, revBytes, nil, 0) // 用Revision在Backend中查找KeyValue
    ...
}

從這兩處就能簡單證實上面所述。

接下來討論Backend是怎麼實現的。
咱們都瞭解，ETCD的持久化存儲是基於BlotDB的，Backend就是對BlotDB封裝了一層。Backend在BlotDB上作了一層緩存，緩存最近的數據。

backend結構體是Backend接口的具體實現，咱們首先關注Range、Put是怎麼實現的

// backend -> ReadTx -> UnsafeRange
func UnsafeRange(...){
    ...
    keys, vals := rt.buf.Range(bucketName, key, endKey, limit) // 從緩存中查找
    if int64(len(keys)) == limit {     
       return keys, vals                    // 若是緩存中有所有數據，直接返回，不走DB
    }
    ...
    k2, v2 := unsafeRange(c, key, endKey, limit-int64(len(keys))) // 從DB中查找
}

// backend -> batchTxBuffered -> UnsafePut
func UnsafePut(...) {
    t.batchTx.UnsafePut(bucketName, key, value)  // 數據保存到DB  
    t.buf.put(bucketName, key, value)            // 數據保存到緩存中
}

那麼查詢一個Key的流程就是這樣的：

1. 根據key，在索引中查到Revision（利用BTree）
2. 根據Revision，在Backend的緩存中查找
3. 若緩存中不符合條件，在BlotDB中查找（Blot本身的索引）

這裏有一個細節，Backend的緩存中是怎麼查找的呢？緩存的結構以下

type bucketBuffer struct {  
    buf []kv  

    used int  
}

數據是保存在一個數組中，每次查找都須要遍歷數組嗎？不是的
每次寫事務提交後都會將本次寫操做的緩存merge到讀緩存上

func merge() {
    ...
    sort.Stable(bb)  
  
    // remove duplicates, using only newest update  
    widx := 0  
    for ridx := 1; ridx < bb.used; ridx++ {  
       if !bytes.Equal(bb.buf[ridx].key, bb.buf[widx].key) {  
          widx++  
       }  
       bb.buf[widx] = bb.buf[ridx]  
    }  
    bb.used = widx + 1
}

merge會將全部key排序，而且去重。也就是說緩存中的key始終是有序的。
因此查找的時候就能夠用二分法了。

func (bb *bucketBuffer) Range(key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {  
    f := func(i int) bool { return bytes.Compare(bb.buf[i].key, key) >= 0 }  
    idx := sort.Search(bb.used, f)  
    if idx < 0 {  
      return nil, nil  
    }  
    if len(endKey) == 0 {  
      if bytes.Equal(key, bb.buf[idx].key) {  
         keys = append(keys, bb.buf[idx].key)  
         vals = append(vals, bb.buf[idx].val)  
      }  
      return keys, vals  
    }  
    if bytes.Compare(endKey, bb.buf[idx].key) <= 0 {  
      return nil, nil  
    }  
    for i := idx; i < bb.used && int64(len(keys)) < limit; i++ {  
      if bytes.Compare(endKey, bb.buf[i].key) <= 0 {  
         break  
    }  
      keys = append(keys, bb.buf[i].key)  
      vals = append(vals, bb.buf[i].val)  
    }  
    return keys, vals  
}

查找先用二分法找到key所在的id，而後從id開始遍歷，找到key~endKey之間的數據。

---

如今已經介紹了store中的存儲用的數據結構。大體結構以下

Range時

Put時