etcd 操作 boltdb 的优化实现

发表于 3年以前 | 总阅读数：543 次

etcd 底层存储使用 boltdb[1] 引擎，boltdb 本身支持事务，同时允许单个写事务与多个读事务，这其中用到了 COW 思想。

boltdb 是 lmdb 的实现，从压测图可以看到，除了读操作，其它各种指标均低于其它引擎，现在 boltdb 也成为了 etcd 的瓶颈之一，性能差，同时事务并发是全局的，并不像 innodb 那样支持 row 行并发

那么，当初 etcd 为什么选择 boltdb 作为 backend 呢？我猜是没得选，用 go 写的单机引擎，还支持 tx 事务的也就这一个，本次分享来看一下 etcd 优化 boltdb 操作的历史实现。

整体架构

上图是 etcd 架构，client 写入数据经过 raft 模块确认后，由 ApplierV3 模块写入 MVCC 层。数据先写入内存表示的 KV index 中，然后再写入 backend, 即 boltdb

从官方文档[2] 中可以看到，boltdb 提供了读写事务的 api

func (db *DB) Update(fn func(*Tx) error) error {
 t, err := db.Begin(true)
  ......
 // If an error is returned from the function then rollback and return error.
 err = fn(t)
  ......
 return t.Commit()
}

但是这个操作太重了，每次写事务 Commit 都会导致磁盘写入，还可能导致 b+tree 重新 rebalance, 操作非常 expensive ...

为了权衡，需要将写操作 batch 到一起，尽可能的提高性能，由于需要额外的操作，etcd 自己封装了一层 batch_tx, 并没有直接使用 boltdb 底层提供的 Batch 函数

原始版本

这一版本实现非常粗糙，简单总结：所有上层的读写操作都使用一个 batch_tx 来实现，也就是读写阻塞的，并没有充份发挥底层 boltdb 特性，来看一下 mvcc/kvstore.go 的 Range 实现

func (s *store) Range(key, end []byte, ro RangeOptions) (r *RangeResult, err error) {
 id := s.TxnBegin()
 kvs, count, rev, err := s.rangeKeys(key, end, ro.Limit, ro.Rev, ro.Count)
 s.txnEnd(id)
  ......
}

func (s *store) TxnBegin() int64 {
 s.mu.Lock()
 s.currentRev.sub = 0
 s.tx = s.b.BatchTx()
 s.tx.Lock()

 s.txnID = rand.Int63()
 return s.txnID
}

func (s *store) txnEnd(txnID int64) error {
  ......
 s.tx.Unlock()
  ......
 s.mu.Unlock()
 return nil
}

Txn 共同使用 backend 层的 BatchTx, 事务开启关闭有一把大锁 s.mu.Lock, 如果有 expensive operation 会导致其它操作阻塞。同时 BatchTx 为了安全，事务每写 10000 条或是超过 100ms, 均会 Commit 持久化并开启新的 BatchTx，供后续使用

读写分离

这个版本为了提高并发，读操作使用 TxnRead 接口，对应 Backend.ReadTx, 写操作使用 TxnWrite 接口，对应 backend.batchTxBuffered, 这两个 tx 封装都是带上 buffer 的，大大的提高了性能。这就是所谓的 N Read or 1 Write 并发。

但是实现还是有竞争和性能损失，比如TxnWrite 在每次提交时都会将写更改写回到 TxnRead 事务的 buffer 中，这会抢占 TxnRead 的锁

同时当 TxnWrite 写事务提交时，为了事务一致性，Commit 操作会回滚读事务 TxnRead

func (s *store) Read() TxnRead {
 s.mu.RLock()
 tx := s.b.ReadTx()
 s.revMu.RLock()
 tx.Lock()
 firstRev, rev := s.compactMainRev, s.currentRev
 s.revMu.RUnlock()
 return newMetricsTxnRead(&storeTxnRead{s, tx, firstRev, rev})
}

func (tr *storeTxnRead) End() {
 tr.tx.Unlock()
 tr.s.mu.RUnlock()
}

func (s *store) Write() TxnWrite {
 s.mu.RLock()
 tx := s.b.BatchTx()
 tx.Lock()
 tw := &storeTxnWrite{
  storeTxnRead: &storeTxnRead{s, tx, 0, 0},
  tx:           tx,
  beginRev:     s.currentRev,
  changes:      make([]mvccpb.KeyValue, 0, 4),
 }
 return newMetricsTxnWrite(tw)
}

func (tw *storeTxnWrite) End() {
  ......
 tw.tx.Unlock()
 if len(tw.changes) != 0 {
  tw.s.revMu.Unlock()
 }
 tw.s.mu.RUnlock()
}

上面就是 Txn 操作的接口，读可以做到并发，写还是串行的

func (t *batchTxBuffered) Unlock() {
 if t.pending != 0 {
  t.backend.readTx.mu.Lock()
  t.buf.writeback(&t.backend.readTx.buf)
  t.backend.readTx.mu.Unlock()
  if t.pending >= t.backend.batchLimit {
   t.commit(false)
  }
 }
 t.batchTx.Unlock()
}

上面是写事务 Unlock 时的操作，将更新的数据回写到 TxnRead 读事务的 buf 中

func (t *batchTxBuffered) unsafeCommit(stop bool) {
 if t.backend.readTx.tx != nil {
  if err := t.backend.readTx.tx.Rollback(); err != nil {
   plog.Fatalf("cannot rollback tx (%s)", err)
  }
  t.backend.readTx.buf.reset()
  t.backend.readTx.tx = nil
 }

 t.batchTx.commit(stop)

 if !stop {
  t.backend.readTx.tx = t.backend.begin(false)
 }
}

上面是写事务提交时的操作，需要回滚读事务，然后重新开启读事务 tx, 并重置内存 buffer

Fully Concurrent

Fully Concurrent[3] 连接是官方的 proposal

从上面的分析可以看到，底层还是有把大锁，如果有一个 expensive read, 比如 range 大范围的 key, 会导致写等待很久，所以为了实现真正的 N Read and 1 Write 的并发粒度，还有很多工作要作

创建一个新的类 ConcurrentReadTx, 每次读请求新建一个 TxnRead, copy 全局 ReadTx.Buf, 而不是复用
当 TxnWrite Unlock 时，需要写回 ReadTx.Buf, 而不是 ConcurrentReadTx.Buf, 所以不会受读影响
当 TxnWrite Commit 时，重新开启 ReadTx 即可，这样新来的 Read 读操作就会基于新的 ReadTx 派生生 ConcurrentReadTx。同时 Commit 也会开启一个 goroutine, 来确保所有的基于旧 ReadTx 的事务结束后回滚。

func (s *store) Read(trace *traceutil.Trace) TxnRead {
 s.mu.RLock()
 s.revMu.RLock()
 // backend holds b.readTx.RLock() only when creating the concurrentReadTx. After
 // ConcurrentReadTx is created, it will not block write transaction.
 tx := s.b.ConcurrentReadTx()
 tx.RLock() // RLock is no-op. concurrentReadTx does not need to be locked after it is created.
 firstRev, rev := s.compactMainRev, s.currentRev
 s.revMu.RUnlock()
 return newMetricsTxnRead(&storeTxnRead{s, tx, firstRev, rev, trace})
}

// 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 {
 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().
 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{
   buf:     b.readTx.buf.unsafeCopy(),
   txMu:    b.readTx.txMu,
   tx:      b.readTx.tx,
   buckets: b.readTx.buckets,
   txWg:    b.readTx.txWg,
  },
 }
}

ReadTx 锁，只会持有很短的时间，直到 unsafeCopy 结束，很轻量的

func (t *batchTxBuffered) Unlock() {
 if t.pending != 0 {
  t.backend.readTx.Lock() // blocks txReadBuffer for writing.
  t.buf.writeback(&t.backend.readTx.buf)
  t.backend.readTx.Unlock()
  if t.pending >= t.backend.batchLimit {
   t.commit(false)
  }
 }
 t.batchTx.Unlock()
}

Unlock 写回更新数据到 readTx.buf，锁时间也很短

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 tx
  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)
  t.backend.readTx.reset()
 }

 t.batchTx.commit(stop)

 if !stop {
  t.backend.readTx.tx = t.backend.begin(false)
 }
}

Commit 提交写的同时，直接开启新的读事务，同时开启 goroutine wait 所有读事务完成后，回滚老的读事务确保一致性。

这里面可以看到，batchTxBuffered 做 Unlock 时，将更改只回写到了全局的 ReadTx.Buf 中，其它读事务是看不到更改的。如果套用数据库 ACID 的隔离级别，这个类似 Read-Repeatable 的实现，而上面版本中能看到，那就类似 Read-Commited 的实现