mvcc | 南风未起｜云原生/Go技术分享博客

# 1.简介

# 1.1.MVCC

MVCC是Multi-Version Concurrency Control的缩写，是一种多版本并发控制方法，用于数据库系统中实现事务的隔离性。MVCC是一种乐观锁机制，通过保存数据的多个版本实现事务的隔离性。

# 1.2.数据版本

reversion是etcd中的一个概念，是一个int64类型的递增整数，用于标识etcd中的数据版本。etcd数据发生变化时，reversion就会递增。

# 删除数据
etcdctl del /test
1
# 查看 revision
etcdctl get / -wjson
# {"header":{"cluster_id":8735285696067307020,"member_id":7131777314758672153,"revision":16,"raft_term":4}}
# 刚才是 15 现在是 16
# 添加 /test2 数据
etcdctl put /test2 t3
OK
# 查看 revision
etcdctl get / -wjson
# {"header":{"cluster_id":8735285696067307020,"member_id":7131777314758672153,"revision":17,"raft_term":4}}

1
2
3
4
5
6
7
8
9
10
11
12
13

# 1.3.存储方式

etcd mvcc中会维护两个数据结构，分别是treeindex和boltdb。treeindex是一颗B树，存储key和reversion的映射关系，主要维护在内存；boltdb是一个key/value数据库，用于存储key和value之间的映射关系，主要维护在磁盘，用于持久化数据。MVCC模块接受请求后会划分为两个类别，分别是读事务ReadTxn和写事务WriteTxn，读事务负责处理range请求，写事务负责put/delete请求，读写事务都基于treeIndex和boltdb提供的能力实现KV的管理功能。

# 2.treeindex

# 2.1.keyIndex

treeindex中，数据的每个key是一个keyIndex结构，它保存了key和reversion之间的映射关系。

type keyIndex struct {
	key         []byte			// key的值
	modified    revision 		// 最后修改的main reversion
	generations []generation	// 保存key的若干历史版本,每代中包含对key的多次修改的版本列表
}

type revision struct {
	main int64					// 全局递增的主版本号,根据put/txn/delete事务递增,事务内的key main版本一致
	sub int64					// 一个事务内的子版本号,从0开始随事务内put/delete操作递增
}

type generation struct {
	ver     int64				// 当前key的修改次数
	created revision 			// generation结构创建时的版本号
	revs    []revision			// 每次修改key时reversion追加到此数组
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

keyIndex会保存key与reversion的索引信息，主要就存在generations。generations标识一个key创建到删除的整个生命周期。初次创建key时会生成第0代generation追加到generations，后续的修改操作向0代追加版本号。key删除时，会生成第1代generation，key不断经历创建、删除的过程generations就会有很多代。
treeIndex会以B树组织keyIndex节点，实现稳定的watch机制和事务隔离能力，树默认最大度为32，也就是叶子节点最大保存63个key，否则进行分裂及再平衡。B树中找到的keyIndex会进一步包装，以版本号作为boltdb key，实际的KV作为boltdb value保存到boltdb存储。
注意

由于B树存在于内存，etcd服务器重启后内存中的treeIndex数据会丢失。因此，etcd会采用一种称为WAL(Write-AHead Log)的机制持久化数据，修改键值对时，更改会被顺序追加到WAL日志，确保数据的持久性和一致性。etcd启动时，加载WAL日志文件就可以重构treeIndex对应B树。

# 2.3.reversion查询

执行get时，MVCC模块创建读事务TxnRead，etcd 3.4版本后支持ConcurrentReadTx，即并发读特性。MVCC读取数据时，先从treeIndex获取keyIndex拿到版本信息，根据key的版本号从boltdb查询数据。

func (ti *treeIndex) Get(key []byte, atRev int64) (modified, created revision,ver int64,err error) {
	keyi := &keyIndex{key: key}
	ti.RLock()
	defer ti.RUnlock()
    // 从B树获取对应的keyIndex
	if keyi = ti.keyIndex(keyi); keyi == nil {
		return revision{}, revision{}, 0, ErrRevisionNotFound
	}
    // 从keyIndex获取小于等于atRev的最接近reversion
	return keyi.get(ti.lg, atRev)
}

1
2
3
4
5
6
7
8
9
10
11

查询版本时，先找到key对应的最新keyIndex，查询时会先从当前节点的items中找，找不到取最接近key的items[i]递归向下找，直至查询到目标或返回nil。

// 从B树获取keyIndex信息
func (ti *treeIndex) keyIndex(keyi *keyIndex) *keyIndex {
	if item := ti.tree.Get(keyi); item != nil {
		return item.(*keyIndex)
	}
	return nil
}

// 获取KeyIndex
func (t *BTree) Get(key Item) Item {
	if t.root == nil {
		return nil
	}
	return t.root.get(key)
}

// 从当前节点找,
func (n *node) get(key Item) Item {
    // 二分查找小于key的最接近keyIndex
	i, found := n.items.find(key)
    // 找到返回
	if found {
		return n.items[i]
    // 从找到位置的子节点继续递归找
	} else if len(n.children) > 0 {
		return n.children[i].get(key)
	}
	return nil
}

// 二分查找keyIndex位置
func (s items) find(item Item) (index int, found bool) {
    // 找到第一个比item大的keyIndex
	i := sort.Search(len(s), func(i int) bool {
		return item.Less(s[i])
	})
    // 非最左侧且前一个元素大等于item时,说明找到item
	if i > 0 && !s[i-1].Less(item) {
		return i - 1, true
	}
    // 否则未找到
	return i, false
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

B树中二分找到keyIndex后，调用keyIndex.get()获取对应版本信息，实现上先查最新的generation，再从generation二分查找reversion。

func (ki *keyIndex) get(...atRev int64) (modified,created revision,ver int64,err error) {
	...
    // 找到key的generation
	g := ki.findGeneration(atRev)
	if g.isEmpty() {
		return revision{}, revision{}, 0, ErrRevisionNotFound
	}
	// 从generation中获取小于atRev的最接近reversion
	n := g.walk(func(rev revision) bool { return rev.main > atRev })
	if n != -1 {
		return g.revs[n], g.created, g.ver - int64(len(g.revs)-n-1), nil
	}
	return revision{}, revision{}, 0, ErrRevisionNotFound
}

// 获取mainRev≤rev且rev[last]＞rev的generation
func (ki *keyIndex) findGeneration(rev int64) *generation {
    // 获取最新的generation
	lastg := len(ki.generations) - 1
	cg := lastg
 
    // 找到包含rev版本的generation
	for cg >= 0 {
        // 最新的generation是对应删除(revs为空),回退到上一个generation
		if len(ki.generations[cg].revs) == 0 {
			cg--
			continue
		}
        // 获取当前可用generation
		g := ki.generations[cg]
        // 如果存在删除
		if cg != lastg {
            // 最后一个generation是空的,说明删除了key
            // 此时generation的最后一个reversion≤rev,说明是tombstone,用于标记删除
            // 此时的generation最后一个reversion＞rev,说明可能包含rev
			if tomb := g.revs[len(g.revs)-1].main; tomb <= rev {
				return nil
			}
		}
        // 最新的generation不是删除生成的,reversion≤rev,说明可能有rev版本
		if g.revs[0].main <= rev {
			return &ki.generations[cg]
		}
		cg--
	}
	return nil
}

// 从后向前遍历generation的版本
func (g *generation) walk(f func(rev revision) bool) int {
	l := len(g.revs)
	for i := range g.revs {
        // 当前版本大于指定版本,继续遍历
		ok := f(g.revs[l-i-1])
        // 当前版本≤指定版本,返回(选中的一定是rev)
		if !ok {
			return l - i - 1
		}
	}
	return -1
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

# 2.4.reversions更新

执行put时，MVCC先从treeIndex模块查询key的版本信息，生成新的递增版本后追加到对应keyIndex.generations[i]，然后将新版本作为key将数据存储到backend/boltdb。

func (ti *treeIndex) Put(key []byte, rev revision) {
	keyi := &keyIndex{key: key}
	ti.Lock()
	defer ti.Unlock()
    // 查找keyIndex
	item := ti.tree.Get(keyi)
	if item == nil {
        //基于当前最新版本更新keyIndex写入B树
		keyi.put(ti.lg, rev.main, rev.sub)
		ti.tree.ReplaceOrInsert(keyi)
		return
	}
    // 直接更新keyIndex信息
	okeyi := item.(*keyIndex)
	okeyi.put(ti.lg, rev.main, rev.sub)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

key对应的keyIndex不存在时，ReplaceOrInsert用于写入keyIndex。

func (t *BTree) ReplaceOrInsert(item Item) Item {
	...
    // 不存在根节点时,初始化写入keyIndex
	if t.root == nil {
		t.root = t.cow.newNode()
		t.root.items = append(t.root.items, item)
		t.length++
		return nil
    // 复制B树
	} else {
		t.root = t.root.mutableFor(t.cow)
        // 每个节点item容量超出,递归分裂
		if len(t.root.items) >= t.maxItems() {
			item2, second := t.root.split(t.maxItems() / 2)
			oldroot := t.root
			t.root = t.cow.newNode()
			t.root.items = append(t.root.items, item2)
			t.root.children = append(t.root.children, oldroot, second)
		}
	}
    // 写入keyIndex
	out := t.root.insert(item, t.maxItems())
	if out == nil {
		t.length++
	}
	return out
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 2.5.reversion删除

执行del命令时，etcd采用延迟删除，原理与put类似。不同在于，key对应的keyIndex最新的generation追加一个删除标识版本tombstone，同时generations追加空的generation作为del的代。

func (ti *treeIndex) Tombstone(key []byte, rev revision) error {
	keyi := &keyIndex{key: key}
	ti.Lock()
	defer ti.Unlock()
    // 查找keyIndex
	item := ti.tree.Get(keyi)
    // 未找到,说明删除元素不存在
	if item == nil {
		return ErrRevisionNotFound
	}
	ki := item.(*keyIndex)
    // 标记key删除
	return ki.tombstone(ti.lg, rev.main, rev.sub)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

tombstone会向最新的generation追加一个标识删除版本tombstone，然后追加一个空的generation。

func (ki *keyIndex) tombstone(lg *zap.Logger, main int64, sub int64) error {
	...
    // 最新的generation是空的或没有任何revs
	if ki.generations[len(ki.generations)-1].isEmpty() {
		return ErrRevisionNotFound
	}
    // 否则追加一个版本
	ki.put(lg, main, sub)
    // 追加一个空的generation
	ki.generations = append(ki.generations, generation{})
	keysGauge.Dec()
	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13

注意

1.key删除时会生成events，watch模块根据key的删除标识生成delete事件

2.重启etcd时，遍历boltdb key构建treeIndex内存树时，会检查哪些key删除，对应key索引生成tombstone标识

3.真正删除treeIndex中的索引对象、boltdb中的key是通过compactor压缩组件异步完成

4.基于etcd延迟删除原理，压缩组件未回收历史版本时，就可以从etcd中找回误删的数据

# 3.backend

# 3.1.backend

backend模块是一个基于boltdb实现的KV存储，支持事务读写及缓冲。backend会在内存维护已提交的事务，积累到一定程度才会批量提交事务，提高etcd整体的写入性能。由于存在维护于内存的事务中间态，因此持久化同样依赖WAL模块保证。另外，backend和treeIndex也是mvcc实现的关键部分。

// backend接口
type Backend interface {
	ReadTx() ReadTx												// 只读事务
	BatchTx() BatchTx											// 批量事务
	ConcurrentReadTx() ReadTx									// 并发读事务
	Snapshot() Snapshot											// 快照
	Hash(ignores map[IgnoreKey]struct{}) (uint32, error)
	Size() int64												// 存储引擎已分配的物理空间大小
	SizeInUse() int64											// 存储引擎已使用的空间大小
	OpenReadTxN() int64											// 活跃的只读事务数量
	Defrag() error												// 碎片整理相关
	ForceCommit()												// 提交批量读写事务
	Close() error
}

// backend实现
type backend struct {
	size int64									// 已分配的字节数
	sizeInUse int64								// 实际使用的字节数
	commits int64								// 提交次数
	openReadTxN int64							// 已打开的只读任务数量
	mu    sync.RWMutex							// 事务隔离锁
	...					
	db    *bolt.DB								// boltdb指针
	...
    batchTx       *batchTxBuffered				// 读写事务缓存
    ...
    txReadBufferCache txReadBufferCache			// 读事务缓存,readTx的镜像
}

func (b *backend) run() {
	defer close(b.donec)
	t := time.NewTimer(b.batchInterval)
	defer t.Stop()
	for {
		select {
		case <-t.C:
        // 停止前提交事务
		case <-b.stopc:
			b.batchTx.CommitAndStop()
			return
		}
        // 否则，每隔batchInterval批量提交一次事务
		if b.batchTx.safePending() != 0 {
			b.batchTx.Commit()
		}
		t.Reset(b.batchInterval)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

注意

backend是boltdb的进一步封装，原有基础上引入缓冲区，读写及事务能力实际还是boltdb提供

# 3.2.只读事务

ReadTx用于检索指定数据，主要有范围读和遍历两个方法。readTx和concurrentReadTx都是其具体实现，readTx是定义为只读事务，实现上会先从buffer缓存获取，缓存没有则进一步从boltdb获取。

type ReadTx interface {
	Lock()
	Unlock()
	RLock()
	RUnlock()
	// 从指定bucket范围查询数据
	UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte)
    // 遍历指定bucket,执行回调方法
	UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error
}

type readTx struct {
	mu  sync.RWMutex					// 读写锁,控制对txReadBuffer读缓存访问
	buf txReadBuffer					// 读缓存

	txMu    sync.RWMutex				// 事务读写锁,范围搜索时,控制对bucket访问
	tx      *bolt.Tx					// 底层存储引擎事务
	buckets map[string]*bolt.Bucket		// 底层引擎的bucket映射,key为bucket name
	...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

读缓冲区是由每个bucket的缓存构成，缓存形式是kv数组，这些kv其实是涉及修改但未提交到boltdb的数据。

type txReadBuffer struct {
	txBuffer								// bucket缓存
	bufVersion uint64						// 缓存版本
}

type txBuffer struct {
	buckets map[string]*bucketBuffer		// bucket的kv缓存
}

type bucketBuffer struct {
	buf []kv								// kv缓存
	used int								// 元素数量
}

type kv struct {
	key []byte
	val []byte
}

// 二分查找缓存kv
func (txr *txReadBuffer) Range(bucketName, key, endKey []byte, limit int64) ([][]byte, [][]byte) {
    // 获取bucket关联KV映射,然后二分查找
	if b := txr.buckets[string(bucketName)]; b != nil {
		return b.Range(key, endKey, limit)
	}
	return nil, nil
}

func (bb *bucketBuffer) Range(key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte) {
    // 二分查找key
	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
	}
    // 非范围查询,返回key对应数据
	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
	}
    // 范围查询,判断起始数据key是否溢出endKey
	if bytes.Compare(endKey, bb.buf[idx].key) <= 0 {
		return nil, nil
	}
    // 记录key~endKey之间数据
	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
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

查询数据时，先从buf缓存获取，不满足时尝试从buckets缓存拿到bucket，找不到bucket时基于boltdb.Bucket()读取，然后从bucket获取数据，新获取的bucket会在buckets映射存一份。

func (rt *readTx) UnsafeRange(bucketName, key, endKey []byte, limit int64) ([][]byte, [][]byte) {
    // endKey是nil,非范围查询
	if endKey == nil {
		// forbid duplicates for single keys
		limit = 1
	}
	...
    // txReadBuffer缓存查找
	keys, vals := rt.buf.Range(bucketName, key, endKey, limit)
	if int64(len(keys)) == limit {
		return keys, vals
	}
	// 读锁限制下缓存获取bucket
	bn := string(bucketName)
	rt.txMu.RLock()
	bucket, ok := rt.buckets[bn]
	rt.txMu.RUnlock()
	if !ok {
        // buckets缓存没找到,可能其他协程写锁修改
        // 写锁限制下boltdb二分递归查找bucket指针
        // 找到会向buckets缓存一份
		rt.txMu.Lock()
		bucket = rt.tx.Bucket(bucketName)
		rt.buckets[bn] = bucket
		rt.txMu.Unlock()
	}
	// 对应bucket没找到,KV不存在
	if bucket == nil {
		return keys, vals
	}
	rt.txMu.Lock()
    // 初始化cursor
	c := bucket.Cursor()
	rt.txMu.Unlock()
	// 基于cursor二分查找KV
	k2, v2 := unsafeRange(c, key, endKey, limit-int64(len(keys)))
    // 这里不需要去重,buf缓存的是当前事务未提交的修改
	return append(k2, keys...), append(v2, vals...)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

unsafeRange会利用cursor.seek()二分查到key的位置，然后移动到nextKey与endKey比较，记录范围内容的所有数据，最终从buf缓冲获取的数据和boltdb获取的数据会进行合并，合并的原因在于buf缓冲的是未提交到底层的修改数据，对boltdb还不可见。

func unsafeRange(c *bolt.Cursor, key, endKey []byte, limit int64) (keys [][]byte, vs [][]byte) {
	// 范围查询或非范围查询处理
	var isMatch func(b []byte) bool
	if len(endKey) > 0 {
		isMatch = func(b []byte) bool { return bytes.Compare(b, endKey) < 0 }
	} else {
		isMatch = func(b []byte) bool { return bytes.Equal(b, key) }
		limit = 1
	}

    // 利用游标查询key~endKey数据
	for ck, cv := c.Seek(key); ck != nil && isMatch(ck); ck, cv = c.Next() {
		vs = append(vs, cv)
		keys = append(keys, ck)
		if limit == int64(len(keys)) {
			break
		}
	}
	return keys, vs
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 3.3.并发读事务

当readTx查询较慢时，mu针对txReadBuffer的读锁迟迟未释放，这种情况支持并发读。但batchTx完成时，会申请写锁将txWriteBuffer同步到txReadBuffer；batchInterval到期将batchTx批量提交重置读写事务时，会申请写锁阻塞新的只读事务；快照生成时，也会提交批量事务，此时申请写锁重置只读事务，这些情况都会阻塞。

// batchTx完成
func (t *batchTxBuffered) Unlock() {
    if t.pending != 0 {
        t.backend.readTx.Lock() 					// 读缓存加锁
        t.buf.writeback(&t.backend.readTx.buf)	// txWriteBuffer同步到txReadBuffer
        t.backend.readTx.Unlock()					// 读缓存解锁
        // 达到批量提交限制,提交事务
        if t.pending >= t.backend.batchLimit {
            t.commit(false)
        }
    }
    t.batchTx.Unlock()
}

// batchInterval到期
func (t *batchTxBuffered) commit(stop bool) {
    t.backend.readTx.Lock()			// 阻塞新的只读事务
    t.unsafeCommit(stop)				// 提交数据
    t.backend.readTx.Unlock()
}

func (t *batchTxBuffered) unsafeCommit(stop bool) {
	if t.backend.readTx.tx != nil {
		// 等待所有只读事务完成
		go func(tx *bolt.Tx, wg *sync.WaitGroup) {
			wg.Wait()
            // 回滚,清理资源,空闲页回收
			if err := tx.Rollback(); err != nil {  ...	}
		}(t.backend.readTx.tx, t.backend.readTx.txWg)
        // readTx重置
		t.backend.readTx.reset()
	}
	// 读写事务提交
	t.batchTx.commit(stop)
	// 非停止,启动新的只读事务备用
	if !stop {
		t.backend.readTx.tx = t.backend.begin(false)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

读写事务提交时与只读事务查询之间的阻塞情况很常见，慢查询带来的阻塞问题相当于一个etcd节点阻塞在查询请求，降低了etcd节点的处理能力。concurrentReadTx其实就是为了解决readTx慢查询场景阻塞问题出现的，concurrentReadTx基于拷贝非共享的思路，复用readTx的底层能力，拷贝新的缓冲区，将N read 1 write问题变为真正的读写并发。

type concurrentReadTx struct {
	buf     txReadBuffer				// 缓冲区
	txMu    *sync.RWMutex				// 事务读写锁
	tx      *bolt.Tx					// 底层存储引擎事务
	buckets map[string]*bolt.Bucket		// bucket缓存
	txWg    *sync.WaitGroup				// 执行组,记录所有正在执行的只读事务和并发读事务,用于重置或更新只读事务
}

func (b *backend) ConcurrentReadTx() ReadTx {
    // 更新backend对应readTx的镜像缓存
	...
    // 初始化concurrentReadTx,拷贝缓冲区时mu上读锁
	return &concurrentReadTx{
		buf:     *buf,					// readTx缓冲拷贝		
		tx:      b.readTx.tx,			// 底层存储引擎事务
		txMu:    &b.readTx.txMu,		// 缓冲读写锁
		buckets: b.readTx.buckets,		// bucket缓存
		txWg:    b.readTx.txWg,			// waitGroup,用来控制只读事务及并发事务执行
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

注意

concurrentReadTx并发读事务的UnsafeRange实现与只读事务一致，只是缓冲区buf是只读事务缓冲区的拷贝。

# 3.4.批量事务

batchTx对应读写事务，定义了增删改查的方法，同时提供batchTx和batchTxBuffered的实现。由于BatchTx定义的一系列读写方法不是安全的，因此使用时需自动控制并发。

type BatchTx interface {
	ReadTx														// 记录的只读事务,用于查询、缓存同步
	UnsafeCreateBucket(name []byte)								// 创建bucket
	UnsafePut(bucketName []byte, key []byte, value []byte)		// 修改数据
	UnsafeSeqPut(bucketName []byte, key []byte, value []byte)	// 修改数据
	UnsafeDelete(bucketName []byte, key []byte)					// 删除数据
	Commit()													// 提交及开启新事务
	CommitAndStop()												// 提交及不开启新事务
}

1
2
3
4
5
6
7
8
9

batchTx基于boltdb的事务进一步封装，batchTxBuffered作为写缓冲区，变更记录会保存在写缓冲区，事务提交时变更同步到读缓冲区，后者其实也是backend使用的读写事务。

type batchTx struct {
	sync.Mutex			// 互斥锁
	tx      *bolt.Tx	// 底层存储引擎事务
	backend *backend	// 关联的backend实例
	pending int			// 待提交的指令数量
}

type txWriteBuffer struct {
	txBuffer			// 缓冲
	seq bool			// 记录是否修改过数据
}

type batchTxBuffered struct {
	batchTx
	buf txWriteBuffer
}

// 批量事务修改数据
func (t *batchTxBuffered) UnsafePut(bucketName []byte, key []byte, value []byte) {
    // 修改
	t.batchTx.UnsafePut(bucketName, key, value)
    // 追加到txWriteBuffer
	t.buf.put(bucketName, key, value)
}

func (t *batchTx) UnsafePut(bucketName []byte, key []byte, value []byte) {
	t.unsafePut(bucketName, key, value, false)
}

// 基于boltdb的bucket操作数据
func (t *batchTx) unsafePut(bucketName []byte, key []byte, value []byte, seq bool) {
    // 根据meta.root查询或创建bucket
	bucket := t.tx.Bucket(bucketName)
    ...
	if seq {
		// seq为true时增大填充百分比,延迟页面分裂或合并
		bucket.FillPercent = 0.9
	}
    // 基于bucket新增或修改数据
	if err := bucket.Put(key, value); err != nil {
		...
	}
    // 记录修改数
	t.pending++
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

注意

批量事务涉及到的方法不再过多赘述，其实底层都是基于boltdb的bucket找到数据、修改数据再put回去，唯一注意的是，这里涉及到的修改会保存一份到txWriteBuffer，同时事务暂时不会提交，直到达到批量条件会统一提交。

# 4.kvstore

# 4.1.store

kvstore真正意思上提供KV存储，其内部包装了抽象的读写方法ReadView、WriteView，隔离事务与非事务操作的读写锁mu，内存B树索引index和后端存储backend/boltdb。

// 基础KV能力接口
type KV interface {
	ReadView															// 读视图接口,定义查询相关方法
	WriteView															// 写视图接口,定义修改操作方法
	Read(trace *traceutil.Trace) TxnRead								// 创建只读事务

	Write(trace *traceutil.Trace) TxnWrite								// 创建读写事务

    Hash() (hash uint32, revision int64, err error)						// 计算后端存储哈希值(数据校验)

	Compact(trace *traceutil.Trace, rev int64) (<-chan struct{}, error)	// 压缩历史版本数据

	Commit()															// 提交事务,基于backend持久化
    
	Restore(b backend.Backend) error									// 从后端恢复数据
	Close() error														// 关闭相关管道
}

// 包装WatchStream的watchable接口
type Watchable interface {
	NewWatchStream() WatchStream										// 创建watchStream的方法
}

// 包装KV基础存储及Watch能力的watchKV接口
type WatchableKV interface {
	KV																	// KV基础存储能力
	Watchable															// watch能力
}

// 扩充一致性索引支持的watchKV,用于RAFT
type ConsistentWatchableKV interface {
	WatchableKV															// 基础存储+watch能力
	ConsistentIndex() uint64											// 一致性索引支持
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

KV是存储的抽象，定义了包括读写、压缩、事务获取等基础能力，WatchableKV扩展KV功能，增加了watch支持，``ConsistentWatchableKV基于WatchableKV进一步增加了一致性索引支持。etcd提供服务基于ConsistentWatchableKV的实现，通过多层包装的复合能力提供KV`存储、事务、监听及数据一致功能。

type store struct {
	ReadView										// 读视图,定义查询接口
	WriteView										// 写视图,定义修改接口

	consistentIndex uint64							// raft一致性索引
	...
    
	mu sync.RWMutex									// 隔离事务与非事务操作的锁

	ig ConsistentIndexGetter						// 一致性索引获取器

    b       backend.Backend							// 后端存储模块(boltdb)
	kvindex index									// B树索引
	...
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

store是基础形式的KV实现，watchableStore继承store实现的基础KV方法，扩展watch能力，ConsistentWatchableKV继承watchableStore进一步扩展一致性索引能力，整体为etcd提供对外能力。

etcdserver (ConsistentWatchableKV)
│
├── 1. 处理Raft一致性、线性读等集群层逻辑
│
└── 2. watchableStore(WatchableKV)
       │
	   ├── 2.1. store结构体,继承所有KV方法
       │     │
       │     ├── readTxn(实现TxnRead,用于快照读)
       │     │
       │     └── batchTx(实现TxnWrite,用于批量写)
       │
	   │── 2.2. watchStream(watchable接口,获取watch接口实现)
       │
       └── 2.3. watchStoreTxn(扩展batchTx,添加事件生成能力)
             │
             └── batchTx提交时自动触发事件通知

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

注意

1.store实现的方法都基于readTxn和writeTxn实现，包括readView和writeView规定的读写方法

2.read和write获取读写事务的方法是核心，操作前都是先获取读写事务，再基于事务操作B树索引及backend落库

3.writeTxn比较关键的实现是end方法，该方法定义了事务结束前动作，store实现基于backend.batchTx持久化数据，watchstore则进一步包装，store.end前基于watchStream通知事件

# 3.2.read

store.read可以获取读事务storeTxnRead，其封装了backend.ReadTx，创建时加锁锁定当前事务版本号，backend.ReadTxn的重置由backendInterval统一进行。

type storeTxnRead struct {
	s  *store				// 底层存储实现
	tx backend.ReadTx		// backend读事务
    firstRev int64			// first rev(压缩版本号)
	rev      int64			// cur rev
}

// 初始化读事务
func (s *store) Read(trace *traceutil.Trace) TxnRead {
	s.mu.RLock()
	s.revMu.RLock()
	// 基于backend初始化读事务
	tx := s.b.ConcurrentReadTx()
	tx.RLock() 
    // 记录压缩版本和当前版本
	firstRev, rev := s.compactMainRev, s.currentRev
	s.revMu.RUnlock()
    // 返回包装的读事务对象
	return newMetricsTxnRead(&storeTxnRead{s, tx, firstRev, rev, trace})
}

// 结束读事务
func (tr *storeTxnRead) End() {
   tr.tx.RUnlock()
   tr.s.mu.RUnlock()
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# 3.3.查询

读操作的处理函数由storeTxnRead.rangeKeys完成，先从B树索引查询最新版本，然后从backend查询数据。backend查询过程已经介绍，基于concurrentReadTx先从buf缓存获取，进一步从boltdb.bucket查询，最后将两者结果合并。

func (tr *storeTxnRead) rangeKeys(key, end []byte, curRev int64, ro RangeOptions) (*RangeResult, error) {
	rev := ro.Rev
	...
    // B树查询版本
	revpairs, total := tr.s.kvindex.Revisions(key, end, rev, int(ro.Limit))
	...
	limit := int(ro.Limit)
	if limit <= 0 || limit > len(revpairs) {
		limit = len(revpairs)
	}
	kvs := make([]mvccpb.KeyValue, limit)
	revBytes := newRevBytes()
    // 遍历版本
	for i, revpair := range revpairs[:len(kvs)] {
        // 解析版本
		revToBytes(revpair, revBytes)
        // 从backend读数据
		_, vs := tr.tx.UnsafeRange(keyBucketName, revBytes, nil, 0)
        // 读出的数据追加到kvs
		if err := kvs[i].Unmarshal(vs[0]); err != nil {
			...
		}
	}
	return &RangeResult{KVs: kvs, Count: total, Rev: curRev}, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 3.4.write

store.write可以获取写事务storeTxnWrite，写事务包装了readTxn和backend.batchTx，写之前会使用最新的版本号读取最新修改，写成功后记录递增后的版本号。由于backend的读写事务按照一定策略批量提交，所以这里的storeTxnWrite.End()只是做一些收尾工作。

type storeTxnWrite struct {
	storeTxnRead				// 写事务,用于读取最新状态
	tx backend.BatchTx			// backend写事务
	beginRev int64				// 开始版本
	changes  []mvccpb.KeyValue	// KV变化内容
}

// 获取写事务
func (s *store) Write(trace *traceutil.Trace) TxnWrite {
	s.mu.RLock()
    // backend写事务初始化
	tx := s.b.BatchTx()
	tx.Lock()
	tw := &storeTxnWrite{
		storeTxnRead: storeTxnRead{s, tx, 0, 0, trace},
		tx:           tx,
		beginRev:     s.currentRev,
		changes:      make([]mvccpb.KeyValue, 0, 4),
	}
	return newMetricsTxnWrite(tw)
}

// 写事务结束
func (tw *storeTxnWrite) End() {
	// 当前数据发生变化,保存
	if len(tw.changes) != 0 {
		tw.s.saveIndex(tw.tx)
		tw.s.revMu.Lock()
		tw.s.currentRev++
	}
	tw.tx.Unlock()
	if len(tw.changes) != 0 {
		tw.s.revMu.Unlock()
	}
	tw.s.mu.RUnlock()
}

func (s *store) saveIndex(tx backend.BatchTx) {
	if s.ig == nil {
		return
	}
	bs := s.bytesBuf8
	ci := s.ig.ConsistentIndex()
	binary.BigEndian.PutUint64(bs, ci)
	// 基于backend的存储能力修改数据
	tx.UnsafePut(metaBucketName, consistentIndexKeyName, bs)
    // 更新一致性索引
	atomic.StoreUint64(&s.consistentIndex, ci)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

注意

watchstore的end方法实现略有不同，其对store实现的end进一步包装，保存数据前调用notify通知注册的watcher事件变更。

# 3.5.修改

storeTxnWrite.put用于更新数据，更新前基于treeIndex查询当前更新版本信息，然后向backend写入数据，再更新B树索引，这里的写入指的是内存中写入，此时还未真正提交到boltdb。

func (tw *storeTxnWrite) Put(key, value []byte, lease lease.LeaseID) int64 {
	tw.put(key, value, lease)
	return tw.beginRev + 1
}

func (tw *storeTxnWrite) put(key, value []byte, leaseID lease.LeaseID) {
	rev := tw.beginRev + 1
	// B树查当前版本
	_, created, ver, err := tw.s.kvindex.Get(key, rev)
	...
	ibytes := newRevBytes()
	idxRev := revision{main: rev, sub: int64(len(tw.changes))}
    // 解析要写入的版本
	revToBytes(idxRev, ibytes)
	ver = ver + 1
	kv := mvccpb.KeyValue{
		Key:            key,
		Value:          value,
		CreateRevision: c,
		ModRevision:    rev,
		Version:        ver,
		Lease:          int64(leaseID),
	}
	// 解析要写入的内容
	d, err := kv.Marshal()
	...
    // 向backend写入数据
	tw.tx.UnsafeSeqPut(keyBucketName, ibytes, d)
    // 更新B树索引
	tw.s.kvindex.Put(key, idxRev)
    // 记录变化内容
	tw.changes = append(tw.changes, kv)
    // 删除旧租约
	if oldLease != lease.NoLease {
		err = tw.s.le.Detach(oldLease, []lease.LeaseItem{{Key: string(key)}})
		...
	}
    // 补充新租约
	if leaseID != lease.NoLease {
		...
		err = tw.s.le.Attach(leaseID, []lease.LeaseItem{{Key: string(key)}})
		...
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44