mvcc: allow large concurrent reads under light write workload

internal/mvcc/backend/backend.go

@@ -310,11 +310,7 @@ func (b *backend) defrag() error {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	// block concurrent read requests while resetting tx
-	b.readTx.mu.Lock()
-	defer b.readTx.mu.Unlock()
-
-	b.batchTx.unsafeCommit(true)
+	b.batchTx.commit(true)
 	b.batchTx.tx = nil
 
 	tmpdb, err := bolt.Open(b.db.Path()+".tmp", 0600, boltOpenOptions)

internal/mvcc/backend/batch_tx.go

@@ -98,6 +98,7 @@ func unsafeRange(c *bolt.Cursor, key, endKey []byte, limit int64) (keys [][]byte
 		isMatch = func(b []byte) bool { return bytes.Equal(b, key) }
 		limit = 1
 	}
+
 	for ck, cv := c.Seek(key); ck != nil && isMatch(ck); ck, cv = c.Next() {
 		vs = append(vs, cv)
 		keys = append(keys, ck)
@@ -154,7 +155,7 @@ func (t *batchTx) Unlock() {
 	t.Mutex.Unlock()
 }
 
-func (t *batchTx) commit(stop bool) {
+func (t *batchTx) commit(stop bool) bool {
 	// commit the last tx
 	if t.tx != nil {
 		if t.pending == 0 && !stop {
@@ -169,7 +170,7 @@ func (t *batchTx) commit(stop bool) {
 			db := t.tx.DB()
 			atomic.StoreInt64(&t.backend.size, size)
 			atomic.StoreInt64(&t.backend.sizeInUse, size-(int64(db.Stats().FreePageN)*int64(db.Info().PageSize)))
-			return
+			return false
 		}
 
 		start := time.Now()
@@ -187,6 +188,7 @@ func (t *batchTx) commit(stop bool) {
 	if !stop {
 		t.tx = t.backend.begin(true)
 	}
+	return true
 }
 
 type batchTxBuffered struct {
@@ -231,22 +233,25 @@ func (t *batchTxBuffered) CommitAndStop() {
 }
 
 func (t *batchTxBuffered) commit(stop bool) {
+	flushed := t.batchTx.commit(stop)
+	if !flushed {
+		return
+	}
+
 	// all read txs must be closed to acquire boltdb commit rwlock
 	t.backend.readTx.mu.Lock()
 	defer t.backend.readTx.mu.Unlock()
 	t.unsafeCommit(stop)
 }
 
-func (t *batchTxBuffered) unsafeCommit(stop bool) {
+func (t *batchTxBuffered) unsafeReadTxCommit(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.reset()
 	}
 
-	t.batchTx.commit(stop)
-
 	if !stop {
 		t.backend.readTx.tx = t.backend.begin(false)
 	}

internal/mvcc/kvstore_test.go

@@ -638,6 +638,8 @@ func TestTxnBlockBackendForceCommit(t *testing.T) {
 	s := NewStore(b, &lease.FakeLessor{}, nil)
 	defer os.Remove(tmpPath)
 
+	// Put a key into the store so that force commit can take effect.
+	s.Put([]byte("foo"), []byte("bar"), lease.NoLease)
 	txn := s.Read()
 
 	done := make(chan struct{})