etcd Backend存储引擎实现原理

ゝ一纸荒年。 2022-12-20 11:10 158阅读 0赞

### 文章目录 ###

*  etcd KV存储层原理
 *   *  Backend 存储引擎
     *   *  读事务ReadTX
         *   *  并发读事务ConcurrentReadTx
             *  读事务的range查询
         *  读写事务 batchTx
         *  batchTx的批量提交
     *  MVCC索引与存储实现
     *   *  Revision概念
         *  keyIndex 和 treeIndex
         *   *  keyIndex
             *  treeIndex
             *  索引的恢复
         *  store存储的实现

# etcd KV存储层原理 #

要理解etcd存储层的原理，需要有一些背景：

1.  对于读操作，不需要走Leader的共识算法，直接通过读事务读取存储层的KV；
2.  对于写操作，首先需要通过raft集群的共识算法，共识算法达成一致之后，会apply写操作日志，也就是进入etcd的KV存储层
3.  etcd的KV存储层是基于BoltDB这个KV存储的嵌入式数据库，BoltDB本身已经实现了事务的隔离性、原子性、持久化、一致性，并提供了并发的单写+多读。但是BoltDB的写是基于COW(copy on write)技术的，这会影响写性能的吞吐量，此外BoltDB也提供了batch写事务以提升写事务吞吐。etcd的 `mvcc/backend` 层是基于 BoltDB 提供的并发1写+N读能力做了优化以提升写吞吐。

etcd KV存储层分为两部分，一部分是etcd实现的基于revision的MVCC层([https://github.com/etcd-io/etcd/tree/master/server/mvcc][https_github.com_etcd-io_etcd_tree_master_server_mvcc])；另一部分是通过 backend 这一后端设计([https://github.com/etcd-io/etcd/tree/master/server/mvcc/backend][https_github.com_etcd-io_etcd_tree_master_server_mvcc_backend])，封装了KV存储引擎的细节，为上层提供了一个统一的接口。

## Backend 存储引擎 ##

Backend封装了一层通用的存储接口：

type Backend interface { 
    	// ReadTx returns a read transaction. It is replaced by ConcurrentReadTx in the main data path, see #10523.
    	ReadTx() ReadTx
    	BatchTx() BatchTx
    	// ConcurrentReadTx returns a non-blocking read transaction.
    	ConcurrentReadTx() ReadTx
    
    	Snapshot() Snapshot
    	Hash(ignores map[IgnoreKey]struct{ }) (uint32, error)
    	// Size returns the current size of the backend physically allocated.
    	// The backend can hold DB space that is not utilized at the moment,
    	// since it can conduct pre-allocation or spare unused space for recycling.
    	// Use SizeInUse() instead for the actual DB size.
    	Size() int64
    	// SizeInUse returns the current size of the backend logically in use.
    	// Since the backend can manage free space in a non-byte unit such as
    	// number of pages, the returned value can be not exactly accurate in bytes.
    	SizeInUse() int64
    	// OpenReadTxN returns the number of currently open read transactions in the backend.
    	OpenReadTxN() int64
    	Defrag() error
    	ForceCommit()
    	Close() error
    }

这里只解释最核心主流程，重点关注`ReadTx() ReadTx` 和 `BatchTx() BatchTx`接口：

*  `ReadTx() ReadTx` 创建了一个读事务；
 *  `BatchTx() BatchTx` 创建了一个读写事务，这里的batch的含义也就是写事务的批量提交意思。

etcd默认是基于 BoltDB存储实现了Backend接口。backend结构体定义如下：

// backend是etcd基于BoltDB实现的KV存储
    // BoltDB 天然支持一写多读并发事务；
    // backend 基于BoltDB封装了事务，提升了并发读的性能
    type backend struct { 
    	// size and commits are used with atomic operations so they must be
    	// 64-bit aligned, otherwise 32-bit tests will crash
    
    	// size is the number of bytes allocated in the backend
    	size int64
    	// sizeInUse is the number of bytes actually used in the backend
    	sizeInUse int64
    	// commits counts number of commits since start
    	commits int64
    	// openReadTxN is the number of currently open read transactions in the backend
    	openReadTxN int64
    
    	// 这里的锁也是隔离下面的db对象；
    	// 正常的创建bolt.DB事务只需要读锁；
    	// 但是做 defrag 时候需要写锁隔离
    	mu sync.RWMutex
    	db *bolt.DB
    
    	// 默认100ms
    	batchInterval time.Duration
    	// 默认defaultBatchLimit = 10000
    	batchLimit int
    	// backend 执行写事务的对象
    	batchTx *batchTxBuffered
    
    	// backend 执行只读事务的对象
    	readTx *readTx
    
    	stopc chan struct{ }
    	donec chan struct{ }
    
    	lg *zap.Logger
    }

从封装的对象就可以看出，backend是使用 `*bolt.DB` 作为存储引擎。

### 读事务ReadTX ###

Backend接口返回读事务的有两个接口：

*  `ReadTx() ReadTx`:
 *  `ConcurrentReadTx() ReadTx`: 在主流程中都是使用的这个并发读事务。

ReadTx接口定义：

type ReadTx interface { 
    	Lock()
    	Unlock()
    	RLock()
    	RUnlock()
    
    	UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte)
    	UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error
    }

不难看出外部对读事务的使用流程:

ReadTx.RLock()
    ReadTx.UnsafeRange()
    ReadTx.RUnlock()

#### 并发读事务ConcurrentReadTx ####

// ConcurrentReadTx creates and returns a new ReadTx, which:
    // A) creates and keeps a copy of backend.readTx.txReadBuffer,
    // B) references the boltdb read Tx (and its bucket cache) of current batch interval.
    func (b *backend) ConcurrentReadTx() ReadTx { 
    	// 这里需要读 readTx 的buffer， 所以需要读锁
    	// 这里的锁占用时间是很低的
    	b.readTx.RLock()
    	defer b.readTx.RUnlock()
    	// prevent boltdb read Tx from been rolled back until store read Tx is done. Needs to be called when holding readTx.RLock().
    	// 用于 ConcurrentReadTx 的，增加一个并发的ReadTx
    	// 在concurrentReadTx.RUnlock()时候会释放这个waiter信号
    	b.readTx.txWg.Add(1)
    	// TODO: might want to copy the read buffer lazily - create copy when A) end of a write transaction B) end of a batch interval.
    	return &concurrentReadTx{ 
    		baseReadTx: baseReadTx{ 
    			// copy一份backend的readTx.buf, 这样就可以不用持有readTx.mu对buffer的保护，从而提升读的性能
    			// 这里就是空间换时间(锁的竞争)
    			buf:     b.readTx.buf.unsafeCopy(),
    			txMu:    b.readTx.txMu,
    			tx:      b.readTx.tx,
    			buckets: b.readTx.buckets,
    			txWg:    b.readTx.txWg,
    		},
    	}
    }

backend通过ConcurrentReadTx()接口创建了一个并发读事务，创建流程表明了几个很重要的点：

*  这里需要持有backend.readTx的读锁，创建完concurrentReadTx对象之后读锁会很快释放，整体对读锁的占用非常轻
 *  新创建的concurrentReadTx对象会对backend.readTx里面的缓存做一次全拷贝，(这里的缓存与batch批量提交有关，后面将batchTx会讲），这样读事务对buf的访问就是独立的，在一个concurrentReadTx内不需要任何加锁。

看一下concurrentReadTx的一些实现：

type concurrentReadTx struct { 
    	baseReadTx
    }
    
    func (rt *concurrentReadTx) Lock()   { }
    func (rt *concurrentReadTx) Unlock() { }
    
    // RLock is no-op. concurrentReadTx does not need to be locked after it is created.
    func (rt *concurrentReadTx) RLock() { }
    
    // RUnlock signals the end of concurrentReadTx.
    func (rt *concurrentReadTx) RUnlock() {  rt.txWg.Done() }

可以看到，就是因为concurrentReadTx前面的一次buf的拷贝操作，所以每个concurrentReadTx的加锁全部是空操作。

这里我们对比非并发的readTx的实现：

// Base type for readTx and concurrentReadTx to eliminate duplicate functions between these
    // baseReadTx的访问是并发的，所以需要读写锁来保护。
    type baseReadTx struct { 
    	// mu protects accesses to the txReadBuffer
    	// 写事务执行End时候，需要获取这个写锁，然后把写事务的更新写到 baseReadTx 的buffer里面；
    	// 创建 concurrentReadTx 时候，需要获取读锁，因为需要拷贝buffer
    	mu  sync.RWMutex
    	buf txReadBuffer
    
    	// TODO: group and encapsulate {txMu, tx, buckets, txWg}, as they share the same lifecycle.
    	// txMu protects accesses to buckets and tx on Range requests.
    	// 这个读写锁是保护下面的tx和buckets
    	txMu    *sync.RWMutex
    	tx      *bolt.Tx
    	buckets map[string]*bolt.Bucket
    	// txWg protects tx from being rolled back at the end of a batch interval until all reads using this tx are done.
    	txWg *sync.WaitGroup
    }
    
    type readTx struct { 
    	baseReadTx
    }
    
    func (rt *readTx) Lock()    {  rt.mu.Lock() }
    func (rt *readTx) Unlock()  {  rt.mu.Unlock() }
    func (rt *readTx) RLock()   {  rt.mu.RLock() }
    func (rt *readTx) RUnlock() {  rt.mu.RUnlock() }

从上面对比可以看到，readTx的加锁过程，是真正意义的加锁，因为并发访问readTx的buf的时候，是需要加锁的。

有关`concurrentReadTx` 优化的设计可以参考相关PR的设计文档：[etcd: Fully Concurrent Reads Design Proposal][etcd_ Fully Concurrent Reads Design Proposal]

这里的优化是针对一些大range查询场景：比如某些大range的查询在占用读事务ReadTx的读锁之后，会占用较长时间。但是对于batchTx这个写事务来说，每次写事务ReadTx提交需要更新读事务ReadTx的buf，所以这里需要竞争ReadTx的读写锁的写锁，但是写锁的获取会一直被读锁block住，因为前面的读锁一直得不到释放，这个场景下会造成写事务超时。针对这个场景，这里提出了`concurrentReadTx` 的优化，通过拷贝的形式，将读锁的占用缩小到创建`concurrentReadTx` ，这样无论读使用的range有多大，都不会block 读锁的竞争，因为每个`concurrentReadTx` 的buf都是基于拷贝私有的。

#### 读事务的range查询 ####

不管是`concurrenctReadTx`还是`readTx`，基本的range操作都是基于其继承的`baseReadTx`实现的：

func (baseReadTx *baseReadTx) UnsafeRange(bucketName, key, endKey []byte, limit int64) ([][]byte, [][]byte) { 
    	if endKey == nil { 
    		// forbid duplicates for single keys
    		limit = 1
    	}
    	if limit <= 0 { 
    		limit = math.MaxInt64
    	}
    	if limit > 1 && !bytes.Equal(bucketName, safeRangeBucket) { 
    		panic("do not use unsafeRange on non-keys bucket")
    	}
    
    	// tips: 优先找：local buffer cache
    	// 读缓存的时候不需要加锁，因为这里
    	keys, vals := baseReadTx.buf.Range(bucketName, key, endKey, limit)
    	if int64(len(keys)) == limit { 
    		return keys, vals
    	}
    
    	// find/cache bucket
    	// tips: 然后找持久化存储
    	bn := string(bucketName)
    	baseReadTx.txMu.RLock()
    	bucket, ok := baseReadTx.buckets[bn]
    	baseReadTx.txMu.RUnlock()
    	lockHeld := false
    	if !ok { 
    		baseReadTx.txMu.Lock()
    		lockHeld = true
    		bucket = baseReadTx.tx.Bucket(bucketName)
    		baseReadTx.buckets[bn] = bucket
    	}
    
    	// ignore missing bucket since may have been created in this batch
    	if bucket == nil { 
    		if lockHeld { 
    			baseReadTx.txMu.Unlock()
    		}
    		return keys, vals
    	}
    	if !lockHeld { 
    		baseReadTx.txMu.Lock()
    		lockHeld = true
    	}
    	c := bucket.Cursor()
    	baseReadTx.txMu.Unlock()
    
    	// 从持久化存储里面查找剩下的kv
    	k2, v2 := unsafeRange(c, key, endKey, limit-int64(len(keys)))
    	return append(k2, keys...), append(v2, vals...)
    }

Range的流程也很好理解：

1.  首先从baseReadTx的buf里面查询，如果从buf里面已经拿到了足够的KV(入参里面有限制range查询的最大数量)，那么就直接返回拿到的KVs;
2.  如果buf里面的KV不足以满足要求，那么这里就会利用 BoltDB的读事务接口去BoltDB 里面查询KV，然后返回。

### 读写事务 batchTx ###

Backend接口`BatchTx() BatchTx` 返回的实际上就是backend.batchTx。 还是先看接口定义：

type BatchTx interface { 
    	ReadTx
    	UnsafeCreateBucket(name []byte)
    	UnsafePut(bucketName []byte, key []byte, value []byte)
    	UnsafeSeqPut(bucketName []byte, key []byte, value []byte)
    	UnsafeDelete(bucketName []byte, key []byte)
    	// Commit commits a previous tx and begins a new writable one.
    	Commit()
    	// CommitAndStop commits the previous tx and does not create a new one.
    	CommitAndStop()
    }

除了batchTx还有一个带缓存的batchTxBuffered，默认也是使用batchTxBuffered：

type batchTxBuffered struct { 
    	batchTx
    	buf txWriteBuffer
    }
    // batchTx是写事务的封装
    // 这里的访问都需要提前获取排他锁
    type batchTx struct { 
    	// 这里的锁是保证 batchTx 的排他，也就是单线程
    	// 所以对下面属性的访问是都是在锁的保护下。
    	sync.Mutex
    	tx      *bolt.Tx
    	backend *backend
    
    	pending int
    }

这里的读写事务之所以叫batch开头，是因为etcd存储层有做一定的优化，默认情况下，写事务不是执行结束后，不会马上提交到BoltDB数据库，而是通过一个默认定时100ms的ticker去触发批量提交，以提升写的吞吐量。但是上层对写事务结束之后需要能够马上读到，所以这里有一个buffer去缓存提交到BoltDB之前的数据，当然这里的buffer需要和读事务的buf一致，所以也存在一个回写到读事务的buf的过程。

这里我们基于使用BatchTx最基本的流程来分析代码：

BatchTx.Lock()
    BatchTx.UnsafePut()
    BatchTx.UnLock()

（1）首先加锁：

func (t *batchTx) Lock() { 
    	t.Mutex.Lock()
    }

前面说过，etcd的存储层对写是单线程的，所以这里也就是通过Mutex这个排他锁，保证了batchTx的事务的排他性。

（2）Put操作：

func (t *batchTxBuffered) UnsafePut(bucketName []byte, key []byte, value []byte) { 
    	t.batchTx.UnsafePut(bucketName, key, value)
    	t.buf.put(bucketName, key, value)
    }
    
    // UnsafePut must be called holding the lock on the tx.
    func (t *batchTx) UnsafePut(bucketName []byte, key []byte, value []byte) { 
    	t.unsafePut(bucketName, key, value, false)
    }
    
    func (t *batchTx) unsafePut(bucketName []byte, key []byte, value []byte, seq bool) { 
    	bucket := t.tx.Bucket(bucketName)
    	if bucket == nil { 
    		t.backend.lg.Fatal(
    			"failed to find a bucket",
    			zap.String("bucket-name", string(bucketName)),
    		)
    	}
    	if seq { 
    		// it is useful to increase fill percent when the workloads are mostly append-only.
    		// this can delay the page split and reduce space usage.
    		bucket.FillPercent = 0.9
    	}
    	if err := bucket.Put(key, value); err != nil { 
    		t.backend.lg.Fatal(
    			"failed to write to a bucket",
    			zap.String("bucket-name", string(bucketName)),
    			zap.Error(err),
    		)
    	}
    	t.pending++
    }

这里要注意的点是：

*  我们实际上是调用batchTxBuffered的`UnsafePut`函数；
 *  这里会基于BoltDB 的写事务将put的KV写到BoltDB(还没有commit)
 *  此外最重要的是这里会同步的写batchTxBuffered的缓存buf

（3）最后释放排它锁

func (t *batchTxBuffered) Unlock() { 
    	// 表示当前 batch 已经有写事务结束但是还没有提交
    	if t.pending != 0 { 
    		// 获取读事务的写锁，然后把写事务的KV更新同步到读事务的缓存Buffer中
    		t.backend.readTx.Lock() // blocks txReadBuffer for writing.
    		t.buf.writeback(&t.backend.readTx.buf)
    		t.backend.readTx.Unlock()
    		// 如果当前的写事务的更新操作的次数已经超过了betch的上限，就将当前pending的
    		// 所有写操作一次提交到BoltDB数据库
    		if t.pending >= t.backend.batchLimit { 
    			t.commit(false)
    		}
    	}
    	t.batchTx.Unlock()
    }

由于batchTxBuffered的实现使用了buf，对于上次来说，调用了UnLock就表示写操作已经执行成功，所以之后的读事务能够马上读取到最新数据。所以这里必须要把写事务的buf刷新回读事务中。

需要注意的是：

1.  `t.backend.readTx.Lock()` 因为会写读事务的buf，所以这里会获取读事务的写锁。但是如果此时如果存在range特别大的读事务正在执行又没有concurrenctReadTx, 那么读锁就不会被释放，倒是这里获取读事务的阻塞过长，而导致写事务超时。这里也是为什么会有`concurrenctReadTx`这个优化的原因。
2.  `t.buf.writeback(&t.backend.readTx.buf)`就是实际执行把写事务中的buf写回读事务中的buf的逻辑。
3.  还有一点，backend 里面也有一些保护操作，如果积累pending的写事务超过了限制也会触发写事务的批量触发，实际执行写事务提交的就是函数：`batchTxBuffered.commit()`

### batchTx的批量提交 ###

前面我们说了 backend 对于写事务的优化是通过buffer + 批量提交来实现的。

这里回到创建 backend实例的地方：

func newBackend(bcfg BackendConfig) *backend { 
    	......
    	// 通过mmap打开BoltDB
    	db, err := bolt.Open(bcfg.Path, 0600, bopts)
    	if err != nil { 
    		bcfg.Logger.Panic("failed to open database", zap.String("path", bcfg.Path), zap.Error(err))
    	}
    
    	// In future, may want to make buffering optional for low-concurrency systems
    	// or dynamically swap between buffered/non-buffered depending on workload.
    	b := &backend{ 
    		db: db,
    
    		batchInterval: bcfg.BatchInterval,
    		batchLimit:    bcfg.BatchLimit,
    		
    		readTx: &readTx{ 
    			baseReadTx: baseReadTx{ 
    				buf: txReadBuffer{ 
    					txBuffer: txBuffer{ make(map[string]*bucketBuffer)},
    				},
    				buckets: make(map[string]*bolt.Bucket),
    				txWg:    new(sync.WaitGroup),
    				txMu:    new(sync.RWMutex),
    			},
    		},
    
    		stopc: make(chan struct{ }),
    		donec: make(chan struct{ }),
    
    		lg: bcfg.Logger,
    	}
    	b.batchTx = newBatchTxBuffered(b)
    	go b.run()
    	return b
    }

上面初始化过程有几点需要注意：

*  创建readTx和batchTx实例时候，其内部的tx并没有被初始化；
 *  `newBatchTxBuffered()` 函数里面会通过一次Commit()操作，来重置readTx和batchTx内部的BoltDB的tx实例。

实现异步批量提交就在 `go b.run()` 里面做的：

func (b *backend) run() { 
    	defer close(b.donec)
    	t := time.NewTimer(b.batchInterval)
    	defer t.Stop()
    	for { 
    		select { 
    		case <-t.C:
    		case <-b.stopc:
    			// 关闭了Backend才会触发
    			b.batchTx.CommitAndStop()
    			return
    		}
    		// 默认每100ms执行一次自动提交
    		if b.batchTx.safePending() != 0 { 
    			// 如果当前周期内存在pending的写事务就提交事务
    			b.batchTx.Commit()
    		}
    		t.Reset(b.batchInterval)
    	}
    }

这里的逻辑很清晰了，就是一个timer定时默认100ms执行一次`batchTx.Commit()`，当然如果当前100ms周期内没有写事务，也就不会执行提交了。

具体我们看下实现:

func (t *batchTxBuffered) Commit() { 
    	t.Lock()
    	t.commit(false)
    	t.Unlock()
    }
    
    func (t *batchTxBuffered) commit(stop bool) { 
    	// all read txs must be closed to acquire boltdb commit rwlock
    	// 这里是个读写锁，会等待所有读锁关闭，然后竞争到独占的写锁
    	// 写事务的提交需要等所有读事务的锁释放
    	t.backend.readTx.Lock()
    	t.unsafeCommit(stop)
    	// 释放读事务的写锁
    	t.backend.readTx.Unlock()
    }
    
    func (t *batchTxBuffered) unsafeCommit(stop bool) { 
    	// 处理读事务
    	if t.backend.readTx.tx != nil { 
    		// wait all store read transactions using the current boltdb tx to finish,
    		// then close the boltdb txs
    		// rollback 读事务
    		go func(tx *bolt.Tx, wg *sync.WaitGroup) { 
    			wg.Wait()
    			if err := tx.Rollback(); err != nil { 
    				t.backend.lg.Fatal("failed to rollback tx", zap.Error(err))
    			}
    		}(t.backend.readTx.tx, t.backend.readTx.txWg)
    		// 每次写事务的batch提交都需要清空 读事务的buffer
    		t.backend.readTx.reset()
    	}
    
    	// batch提交写事务
    	t.batchTx.commit(stop)
    
    	if !stop { 
    		// 重新创建读事务
    		t.backend.readTx.tx = t.backend.begin(false)
    	}
    }

大概需要做的事情如下:

1.  首先需要获取batchTxBuffered的排他锁，确保提交过程中不能有并发写事务执行；
2.  也需要获取读事务的读写锁中的写锁，这里也就是阻止了新的读事务创建；
3.  在`unsafeCommit()`函数里面提交异步协程，等待当前所有的正在执行的读事务执行完毕，然后将所有读事务回滚掉(BoltDB只读事务不涉及写，所以必须回滚掉，我估计就是一些资源释放。)
4.  然后将读事务的BoltDB的tx置为nil;
5.  执行 `t.batchTx.commit(stop)`, 里面实际上也就是调用了 BoltDB.tx这个batchTx的提交，将写事务提交到BlotDB里面。然后给backend.batchTx创建了新的写事务。
6.  最后再给backend的读事务创建一个全新的读事务。

到此写事务的批量提交也就是完成了，这里有几点细节需要再强调一些：

1.  Backend的写事务是通过排他锁保证单线程执行的；
2.  写事务的批量提交是基于Buffer机制的，这里就是buffer还没提交机器就挂了，也不会丢数据，因为还有WAL日志呢；
3.  写事务的批量提交时候，读事务其实也是被block的，也就是也无法创建新的读事务。但是不影响前面已经创建好的正在执行中的读事务，因为BoltDB 是支持单写事务+N个读事务并发执行的。

整体来说，Backend实现的性能强劲地方最核心几个点：

1.  批量提交BatchTx写事务+buffer机制；
2.  基于copy实现的读事务正在完全的并发；

## MVCC索引与存储实现 ##

前面我们分析了Backend的实现，并且知道Backend封装了一套接口，提供了高性能的并发执行1写事务 + 完全并发的N个读事务能力。

etcd的应用数据存储层，在Backend的基础上又封装了一层MVCC([https://github.com/etcd-io/etcd/tree/master/server/mvcc][https_github.com_etcd-io_etcd_tree_master_server_mvcc])。这层MVCC的存储能力是基于Backend实现的：

1.  封装了一层通用的Read/Write 接口，屏蔽了底层存储引擎BoltDB；
2.  每个KV都有revision的概念，事务的revision是递增的。基于BTree实现了一套索引机制。检索过程是通过 Key，可以找到这个key生命周期中的所有revision。而BoltDB 里面存储的实际上是Revision -> Value的映射。
3.  对于etcd来说，它的多版本中的“版本”实际上是指revision；
4.  基于Revision 和 Backend, 又实现了Watch机制；

### Revision概念 ###

etcd存储数据时，实际存储引擎中存放的key并不是我们实际PUT的KV中的K，而是以数据的revision作为key，value中存放的是数据的KV。

revision由2部分组成：\{main revision, sub revision\}。每次事务main revision都会递增，同一个事务中，每次操作sub revision都会自增1。这两个结合就能保证每次key唯一而且是递增的。  
![在这里插入图片描述][20201112234247350.png_pic_center]

举个例子：比如通过批量接口两次更新两对键值，第一次写入数据时，写入<key1,value1>和<key2,value2>，在Etcd这边的存储看来，存放的数据就是这样的：

revision={1,0}, key=key1, value=value1
     revision={1,1}, key=key2, value=value2

而在第二次更新写入数据<key1,update1>和<key2,update2>后，存储中又记录（注意不是覆盖前面的数据）了以下数据：

revision={2,0}, key=key1, value=update1
    revision={2,1}, key=key2, value=update2

对于客户端来说，每次操作的时候是根据Key来进行操作的，所以这里就需要一个Key映射到当前revision的操作了，为了做到这个映射关系，Etcd引入了一个内存中的Btree索引，整个操作过程如下面的流程所示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70_pic_center]  
查询时，先通过内存中的btree索引来查询该key对应的keyIndex结构体，然后再根据这个结构体才能去boltdb中查询真实的数据返回。

### keyIndex 和 treeIndex ###

#### keyIndex ####

keyIndex相关实现可参考源码：[https://github.com/etcd-io/etcd/blob/master/server/mvcc/key\_index.go][https_github.com_etcd-io_etcd_blob_master_server_mvcc_key_index.go]

keyIndex存储了一个key在backend中从第一次PUT开始的所有生命周期中的所有revision记录。

keyIndex结构体：

// A revision indicates modification of the key-value space.
    // The set of changes that share same main revision changes the key-value space atomically.
    type revision struct { 
    	// main is the main revision of a set of changes that happen atomically.
    	main int64
    	// sub is the sub revision of a change in a set of changes that happen atomically. Each change has different increasing sub revision in that set.
    	sub int64
    }
    // generation contains multiple revisions of a key.
    type generation struct { 
    	ver     int64
    	created revision // when the generation is created (put in first revision).
    	revs    []revision
    }
    
    type keyIndex struct { 
    	key         []byte
    	modified    revision // the main rev of the last modification
    	generations []generation
    }

*  key：存储数据真实的键。
 *  modified：最后一次修改该键对应的revision。
 *  generations：generation数组。

可以认为每个generation对应一个数据从创建到删除的过程。每次删除key的操作，都会导致一个generation最后添加一个tombstone记录，然后创建一个新的空generation记录添加到generations数组中。

generation结构体存放以下数据：

*  ver：当前generation中存放了多少次修改，其实就是revs数组的大小-1（因为需要去掉tombstone）。
 *  created：创建该generation时的revision。
 *  revs：存放该generation中存放的revision数组。

以下图来说明keyIndex结构体  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70_pic_center 1]  
如上图所示，存放的键为test的keyIndex结构。

它的generations数组有两条记录，其中generations\[0\]在revision 1.0时创建，当revision2.1的时候进行tombstone操作，因此该generation的created是1.0；对应的generations\[1\]在revision3.3时创建，紧跟着就做了tombstone操作。所以该keyIndex.modifiled成员存放的是3.3，因为这是这条数据最后一次被修改的revision。

一个已经被tombstone的generation是可以被删除的，如果整个generations数组都已经被删除空了，那么整个keyIndex记录也可以被删除了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70_pic_center 2]  
如上图所示，keyIndex.compact(n)函数可以对keyIndex数据进行压缩操作，将删除满足main revision < n的数据。

*  compact(2)：找到了generations\[0\]的1.0 revision的数据进行了删除。
 *  compact(3)：找到了generations\[0\]的2.1 revision的数据进行了删除，此时由于generations\[0\]已经没有数据了，所以这一整个generation被删除，原先的generations\[1\]变成了generations\[0\]。
 *  compact(4)：找到了generations\[0\]的3.3 revision的数据进行了删除。由于所有的generation数据都被删除了，此时这个keyIndex数据可以删除了。

#### treeIndex ####

Etcd中使用treeIndex来在内存中存放keyIndex数据信息，这样就可以快速的根据输入的key定位到对应的keyIndex。

treeIndex使用开源的[github.com/google/btree][github.com_google_btree]来在内存中存储btree索引信息，因为用的是外部库，所以不打算就这部分做解释。而如果很清楚了前面keyIndex结构，其实这部分很好理解。

所有的操作都以key做为参数进行操作，treeIndex使用btree根据key查找到对应的keyIndex，再进行相关的操作，最后重新写入到btree中。

#### 索引的恢复 ####

TODO

### store存储的实现 ###

前面我们说过，mvcc的存储store实际上是对 backend存储引擎的封装；

先看store里面对KV接口的定义：

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
    }

该接口有两个实现：

1.  mvcc/kvstore,go\#store，store结构体实现了KV接口，提供基本的KV操作；
2.  mvcc/watchable\_store,go\#watchableStore，watchableStore是基于store提供的KV操作，封装了一层watch的机制。

这里我们重点介绍store基本KV实现。

在etcdServer模块里面，对mvcc模块基本调用模式是：

txn := KV.Write(trace)
    txn.Put()
    txn.Range()
    ...
    // 结束事务
    txn.End()

下面我们就基于这个基本调用模式来分析源码：

首先看一下mvcc层基本的事务接口：

type ReadView interface { 
    	FirstRev() int64
    	Rev() int64
    	Range(key, end []byte, ro RangeOptions) (r *RangeResult, err error)
    }
    
    // 只读的事务
    type TxnRead interface { 
    	ReadView
    	// End marks the transaction is complete and ready to commit.
    	End()
    }
    
    type WriteView interface { 
    	DeleteRange(key, end []byte) (n, rev int64)
    	Put(key, value []byte, lease lease.LeaseID) (rev int64)
    }
    
    type TxnWrite interface { 
    	TxnRead
    	WriteView
    	// Changes gets the changes made since opening the write txn.
    	Changes() []mvccpb.KeyValue
    }

可以看到整个接口继承是比较复杂的。

（1）`KV.Write() TxnWrite` 获取一些写事务

func (s *store) Write(trace *traceutil.Trace) TxnWrite { 
    	s.mu.RLock()
    	// 获取 backend 当前的写事务对象
    	tx := s.b.BatchTx()
    	// 获取写事务的独占锁，也就是同一时间，只能存在一个写事务，
    	// Tips: 这里需要加锁，然后确保写事务的执行是单线程的
    	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)
    }

可以看到这里的本质是创建了一个`storeTxnWrite`，当然这里会获取backend的BatchTx的排他锁，表明写事务的独占的。

（2）TxnWrite的PUT方法

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
    	c := rev
    	oldLease := lease.NoLease
    
    	// if the key exists before, use its previous created and
    	// get its previous leaseID
    	_, created, ver, err := tw.s.kvindex.Get(key, rev)
    	if err == nil { 
    		c = created.main
    		oldLease = tw.s.le.GetLease(lease.LeaseItem{ Key: string(key)})
    	}
    	tw.trace.Step("get key's previous created_revision and leaseID")
    	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()
    	if err != nil { 
    		tw.storeTxnRead.s.lg.Fatal(
    			"failed to marshal mvccpb.KeyValue",
    			zap.Error(err),
    		)
    	}
    
    	tw.trace.Step("marshal mvccpb.KeyValue")
    	tw.tx.UnsafeSeqPut(keyBucketName, ibytes, d)
    	tw.s.kvindex.Put(key, idxRev)
    	tw.changes = append(tw.changes, kv)
    	tw.trace.Step("store kv pair into bolt db")
    
    	if oldLease != lease.NoLease { 
    		if tw.s.le == nil { 
    			panic("no lessor to detach lease")
    		}
    		err = tw.s.le.Detach(oldLease, []lease.LeaseItem{ { Key: string(key)}})
    		if err != nil { 
    			tw.storeTxnRead.s.lg.Error(
    				"failed to detach old lease from a key",
    				zap.Error(err),
    			)
    		}
    	}
    	if leaseID != lease.NoLease { 
    		if tw.s.le == nil { 
    			panic("no lessor to attach lease")
    		}
    		err = tw.s.le.Attach(leaseID, []lease.LeaseItem{ { Key: string(key)}})
    		if err != nil { 
    			panic("unexpected error from lease Attach")
    		}
    	}
    	tw.trace.Step("attach lease to kv pair")
    }

这里的PUT逻辑其实也比较清晰：

1.  将当前KV PUT操作的revision存入Btree索引中；
2.  调用backend的batchTx的接口写入事务中；
3.  如果当前PUT操作绑定了lease， 那attach一个异步lease处理协程。
4.  保存当前操作KV entity到storeTxnWrite的changes列表中。

（3）TxnWrite的End方法以表明完成事务

func (tw *storeTxnWrite) End() { 
    	// only update index if the txn modifies the mvcc state.
    	if len(tw.changes) != 0 { 
    		tw.s.saveIndex(tw.tx)
    		// hold revMu lock to prevent new read txns from opening until writeback.
    		tw.s.revMu.Lock()
    		tw.s.currentRev++
    	}
    	// 这里需要释放 backend.BatchTx 的独占锁
    	// 释放之后，其余的线程就可以再来获取写事务对象
    	tw.tx.Unlock()
    	if len(tw.changes) != 0 { 
    		tw.s.revMu.Unlock()
    	}
    	// 释放store的读锁
    	tw.s.mu.RUnlock()
    }

这里很重要：

1.  首先这个写事务有KV的changes操作，需要调用`tw.s.saveIndex(tw.tx)`更新存储的一致性索引，并做持久化，这里的一致性索引是用于Etcd存储引擎幂等判断。
2.  持有锁，并更新自增主版本号；

参考：  
[etcd多版本并发控制][etcd]  
[高可用分布式存储 etcd 的实现原理][etcd 1]  
[Etcd存储的实现][Etcd]

[https_github.com_etcd-io_etcd_tree_master_server_mvcc]: https://github.com/etcd-io/etcd/tree/master/server/mvcc
[https_github.com_etcd-io_etcd_tree_master_server_mvcc_backend]: https://github.com/etcd-io/etcd/tree/master/server/mvcc/backend
[etcd_ Fully Concurrent Reads Design Proposal]: https://docs.google.com/document/d/1V9UuL8BnpZF2xFpHhvE1lO2hWvtdiLml3Ukr-m4jcdI/edit#
[20201112234247350.png_pic_center]: /images/20221120/a92c086e5519459491f173c7b80ca2c4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/b921d260087c4d51965f8272ab4674de.png
[https_github.com_etcd-io_etcd_blob_master_server_mvcc_key_index.go]: https://github.com/etcd-io/etcd/blob/master/server/mvcc/key_index.go
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201113000256730.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE=,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201113000342935.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4NTMyNjE=,size_16,color_FFFFFF,t_70#pic_center
[github.com_google_btree]: https://github.com/google/btree
[etcd]: https://www.lixueduan.com/post/etcd/06-why-mvcc/
[etcd 1]: https://draveness.me/etcd-introduction/
[Etcd]: https://www.codedump.info/post/20181125-etcd-server/