mvcc: address comments

mvcc/backend/backend.go

@@ -49,6 +49,7 @@ var (
 )
 
 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.
@@ -200,7 +201,7 @@ func (b *backend) ReadTx() ReadTx { return b.readTx }
 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.
+	// 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{
@@ -516,9 +517,10 @@ func (b *backend) begin(write bool) *bolt.Tx {
 
 	size := tx.Size()
 	db := tx.DB()
+	stats := db.Stats()
 	atomic.StoreInt64(&b.size, size)
-	atomic.StoreInt64(&b.sizeInUse, size-(int64(db.Stats().FreePageN)*int64(db.Info().PageSize)))
-	atomic.StoreInt64(&b.openReadTxN, int64(db.Stats().OpenTxN))
+	atomic.StoreInt64(&b.sizeInUse, size-(int64(stats.FreePageN)*int64(db.Info().PageSize)))
+	atomic.StoreInt64(&b.openReadTxN, int64(stats.OpenTxN))
 
 	return tx
 }

mvcc/backend/batch_tx.go

@@ -308,7 +308,16 @@ 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 waitAndRollback(t.backend.readTx.tx, t.backend.readTx.txWg, t.backend.lg)
+		go func(tx *bolt.Tx, wg *sync.WaitGroup) {
+			wg.Wait()
+			if err := tx.Rollback(); err != nil {
+				if t.backend.lg != nil {
+					t.backend.lg.Fatal("failed to rollback tx", zap.Error(err))
+				} else {
+					plog.Fatalf("cannot rollback tx (%s)", err)
+				}
+			}
+		}(t.backend.readTx.tx, t.backend.readTx.txWg)
 		t.backend.readTx.reset()
 	}
 
@@ -319,17 +328,6 @@ func (t *batchTxBuffered) unsafeCommit(stop bool) {
 	}
 }
 
-func waitAndRollback(tx *bolt.Tx, wg *sync.WaitGroup, lg *zap.Logger) {
-	wg.Wait()
-	if err := tx.Rollback(); err != nil {
-		if lg != nil {
-			lg.Fatal("failed to rollback tx", zap.Error(err))
-		} else {
-			plog.Fatalf("cannot rollback tx (%s)", err)
-		}
-	}
-}
-
 func (t *batchTxBuffered) UnsafePut(bucketName []byte, key []byte, value []byte) {
 	t.batchTx.UnsafePut(bucketName, key, value)
 	t.buf.put(bucketName, key, value)

mvcc/backend/read_tx.go

@@ -132,13 +132,17 @@ type concurrentReadTx struct {
 	buf     txReadBuffer
 	txMu    *sync.RWMutex
 	tx      *bolt.Tx
-	buckets map[string]*bolt.Bucket // note: A map value is a pointer
+	buckets map[string]*bolt.Bucket
 	txWg    *sync.WaitGroup
 }
 
-func (rt *concurrentReadTx) Lock()    {}
-func (rt *concurrentReadTx) Unlock()  {}
-func (rt *concurrentReadTx) RLock()   {}
+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() }
 
 func (rt *concurrentReadTx) UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error {

mvcc/kvstore_txn.go

@@ -35,6 +35,7 @@ func (s *store) Read() TxnRead {
 	// 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})
@@ -48,7 +49,7 @@ func (tr *storeTxnRead) Range(key, end []byte, ro RangeOptions) (r *RangeResult,
 }
 
 func (tr *storeTxnRead) End() {
-	tr.tx.RUnlock()
+	tr.tx.RUnlock() // RUnlock signals the end of concurrentReadTx.
 	tr.s.mu.RUnlock()
 }