server: make applier use ReadTx() in Txn() instead of ConcurrentReadTx()

server/etcdserver/apply.go

@@ -336,7 +336,7 @@ func (a *applierV3backend) Range(ctx context.Context, txn mvcc.TxnRead, r *pb.Ra
  resp.Header = &pb.ResponseHeader{}
 
  if txn == nil {
- txn = a.s.kv.Read(trace)
+ txn = a.s.kv.Read(mvcc.ConcurrentReadTxMode, trace)
  defer txn.End()
  }
 
@@ -434,7 +434,15 @@ func (a *applierV3backend) Txn(ctx context.Context, rt *pb.TxnRequest) (*pb.TxnR
  ctx = context.WithValue(ctx, traceutil.TraceKey, trace)
  }
  isWrite := !isTxnReadonly(rt)
- txn := mvcc.NewReadOnlyTxnWrite(a.s.KV().Read(trace))
+
+ // When the transaction contains write operations, we use ReadTx instead of
+ // ConcurrentReadTx to avoid extra overhead of copying buffer.
+ var txn mvcc.TxnWrite
+ if isWrite {
+ txn = mvcc.NewReadOnlyTxnWrite(a.s.KV().Read(mvcc.SharedBufReadTxMode, trace))
+ } else {
+ txn = mvcc.NewReadOnlyTxnWrite(a.s.KV().Read(mvcc.ConcurrentReadTxMode, trace))
+ }
 
  var txnPath []bool
  trace.StepWithFunction(

server/mvcc/kv.go

@@ -100,12 +100,21 @@ func (trw *txnReadWrite) Changes() []mvccpb.KeyValue { return nil }
 
 func NewReadOnlyTxnWrite(txn TxnRead) TxnWrite { return &txnReadWrite{txn} }
 
+type ReadTxMode uint32
+
+const (
+ // Use ConcurrentReadTx and the txReadBuffer is copied
+ ConcurrentReadTxMode = ReadTxMode(1)
+ // Use backend ReadTx and txReadBuffer is not copied
+ SharedBufReadTxMode = ReadTxMode(2)
+)
+
 type KV interface {
  ReadView
  WriteView
 
  // Read creates a read transaction.
- Read(trace *traceutil.Trace) TxnRead
+ Read(mode ReadTxMode, trace *traceutil.Trace) TxnRead
 
  // Write creates a write transaction.
  Write(trace *traceutil.Trace) TxnWrite

server/mvcc/kv_test.go

@@ -50,7 +50,7 @@ var (
  return kv.Range(context.TODO(), key, end, ro)
  }
  txnRangeFunc = func(kv KV, key, end []byte, ro RangeOptions) (*RangeResult, error) {
- txn := kv.Read(traceutil.TODO())
+ txn := kv.Read(ConcurrentReadTxMode, traceutil.TODO())
  defer txn.End()
  return txn.Range(context.TODO(), key, end, ro)
  }

server/mvcc/kv_view.go

@@ -24,19 +24,19 @@ import (
 type readView struct{ kv KV }
 
 func (rv *readView) FirstRev() int64 {
- tr := rv.kv.Read(traceutil.TODO())
+ tr := rv.kv.Read(ConcurrentReadTxMode, traceutil.TODO())
  defer tr.End()
  return tr.FirstRev()
 }
 
 func (rv *readView) Rev() int64 {
- tr := rv.kv.Read(traceutil.TODO())
+ tr := rv.kv.Read(ConcurrentReadTxMode, traceutil.TODO())
  defer tr.End()
  return tr.Rev()
 }
 
 func (rv *readView) Range(ctx context.Context, key, end []byte, ro RangeOptions) (r *RangeResult, err error) {
- tr := rv.kv.Read(traceutil.TODO())
+ tr := rv.kv.Read(ConcurrentReadTxMode, traceutil.TODO())
  defer tr.End()
  return tr.Range(ctx, key, end, ro)
 }

server/mvcc/kvstore_test.go

@@ -658,7 +658,7 @@ func TestConcurrentReadNotBlockingWrite(t *testing.T) {
  s.Put([]byte("foo"), []byte("bar"), lease.NoLease)
 
  // readTx simulates a long read request
- readTx1 := s.Read(traceutil.TODO())
+ readTx1 := s.Read(ConcurrentReadTxMode, traceutil.TODO())
 
  // write should not be blocked by reads
  done := make(chan struct{}, 1)
@@ -673,7 +673,7 @@ func TestConcurrentReadNotBlockingWrite(t *testing.T) {
  }
 
  // readTx2 simulates a short read request
- readTx2 := s.Read(traceutil.TODO())
+ readTx2 := s.Read(ConcurrentReadTxMode, traceutil.TODO())
  ro := RangeOptions{Limit: 1, Rev: 0, Count: false}
  ret, err := readTx2.Range(context.TODO(), []byte("foo"), nil, ro)
  if err != nil {
@@ -756,7 +756,7 @@ func TestConcurrentReadTxAndWrite(t *testing.T) {
  mu.Lock()
  wKVs := make(kvs, len(committedKVs))
  copy(wKVs, committedKVs)
- tx := s.Read(traceutil.TODO())
+ tx := s.Read(ConcurrentReadTxMode, traceutil.TODO())
  mu.Unlock()
  // get all keys in backend store, and compare with wKVs
  ret, err := tx.Range(context.TODO(), []byte("\x00000000"), []byte("\xffffffff"), RangeOptions{})

server/mvcc/kvstore_txn.go

@@ -34,12 +34,18 @@ type storeTxnRead struct {
  trace *traceutil.Trace
 }
 
-func (s *store) Read(trace *traceutil.Trace) TxnRead {
+func (s *store) Read(mode ReadTxMode, trace *traceutil.Trace) TxnRead {
+ var tx backend.ReadTx
  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()
+ if mode == ConcurrentReadTxMode {
+ tx = s.b.ConcurrentReadTx()
+ } else {
+ tx = s.b.ReadTx()
+ }
+
  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()