POC for caching range response

etcdserver/api/v3rpc/watch.go

@@ -404,7 +404,7 @@ func (sws *serverWatchStream) sendLoop() {
 					opt := mvcc.RangeOptions{Rev: evs[i].Kv.ModRevision - 1}
 					r, err := sws.watchable.Range(evs[i].Kv.Key, nil, opt)
 					if err == nil && len(r.KVs) != 0 {
-						events[i].PrevKv = &(r.KVs[0])
+						events[i].PrevKv = r.KVs[0]
 					}
 				}
 			}

etcdserver/apply.go

@@ -218,7 +218,7 @@ func (a *applierV3backend) Put(txn mvcc.TxnWrite, p *pb.PutRequest) (resp *pb.Pu
 	}
 	if p.PrevKv {
 		if rr != nil && len(rr.KVs) != 0 {
-			resp.PrevKv = &rr.KVs[0]
+			resp.PrevKv = rr.KVs[0]
 		}
 	}
 
@@ -245,7 +245,7 @@ func (a *applierV3backend) DeleteRange(txn mvcc.TxnWrite, dr *pb.DeleteRangeRequ
 		if rr != nil {
 			resp.PrevKvs = make([]*mvccpb.KeyValue, len(rr.KVs))
 			for i := range rr.KVs {
-				resp.PrevKvs[i] = &rr.KVs[i]
+				resp.PrevKvs[i] = rr.KVs[i]
 			}
 		}
 	}
@@ -346,7 +346,7 @@ func (a *applierV3backend) Range(ctx context.Context, txn mvcc.TxnRead, r *pb.Ra
 		if r.KeysOnly {
 			rr.KVs[i].Value = nil
 		}
-		resp.Kvs[i] = &rr.KVs[i]
+		resp.Kvs[i] = rr.KVs[i]
 	}
 	trace.Step("assemble the response")
 	return resp, nil
@@ -463,7 +463,7 @@ func applyCompare(rv mvcc.ReadView, c *pb.Compare) bool {
 			// nil == empty string in grpc; no way to represent missing value
 			return false
 		}
-		return compareKV(c, mvccpb.KeyValue{})
+		return compareKV(c, &mvccpb.KeyValue{})
 	}
 	for _, kv := range rr.KVs {
 		if !compareKV(c, kv) {
@@ -473,7 +473,7 @@ func applyCompare(rv mvcc.ReadView, c *pb.Compare) bool {
 	return true
 }
 
-func compareKV(c *pb.Compare, ckv mvccpb.KeyValue) bool {
+func compareKV(c *pb.Compare, ckv *mvccpb.KeyValue) bool {
 	var result int
 	rev := int64(0)
 	switch c.Target {
@@ -868,7 +868,7 @@ func (a *quotaApplierV3) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantRes
 	return resp, err
 }
 
-type kvSort struct{ kvs []mvccpb.KeyValue }
+type kvSort struct{ kvs []*mvccpb.KeyValue }
 
 func (s *kvSort) Swap(i, j int) {
 	t := s.kvs[i]
@@ -1019,7 +1019,7 @@ func pruneKVs(rr *mvcc.RangeResult, isPrunable func(*mvccpb.KeyValue) bool) {
 	j := 0
 	for i := range rr.KVs {
 		rr.KVs[j] = rr.KVs[i]
-		if !isPrunable(&rr.KVs[i]) {
+		if !isPrunable(rr.KVs[i]) {
 			j++
 		}
 	}

mvcc/kv.go

@@ -28,7 +28,7 @@ type RangeOptions struct {
 }
 
 type RangeResult struct {
-	KVs   []mvccpb.KeyValue
+	KVs   []*mvccpb.KeyValue
 	Rev   int64
 	Count int
 }

mvcc/kv_test.go

@@ -634,7 +634,7 @@ func TestKVRestore(t *testing.T) {
 		b, tmpPath := backend.NewDefaultTmpBackend()
 		s := NewStore(zap.NewExample(), b, &lease.FakeLessor{}, nil, StoreConfig{})
 		tt(s)
-		var kvss [][]mvccpb.KeyValue
+		var kvss [][]*mvccpb.KeyValue
 		for k := int64(0); k < 10; k++ {
 			r, _ := s.Range([]byte("a"), []byte("z"), RangeOptions{Rev: k})
 			kvss = append(kvss, r.KVs)
@@ -652,7 +652,7 @@ func TestKVRestore(t *testing.T) {
 
 		// wait for possible compaction to finish
 		testutil.WaitSchedule()
-		var nkvss [][]mvccpb.KeyValue
+		var nkvss [][]*mvccpb.KeyValue
 		for k := int64(0); k < 10; k++ {
 			r, _ := ns.Range([]byte("a"), []byte("z"), RangeOptions{Rev: k})
 			nkvss = append(nkvss, r.KVs)

mvcc/kvstore.go

@@ -89,6 +89,9 @@ type store struct {
 
 	ig ConsistentIndexGetter
 
+	cache map[revision]mvccpb.KeyValue
+	clock sync.RWMutex
+
 	b       backend.Backend
 	kvindex index
 
@@ -125,6 +128,8 @@ func NewStore(lg *zap.Logger, b backend.Backend, le lease.Lessor, ig ConsistentI
 		b:       b,
 		ig:      ig,
 		kvindex: newTreeIndex(lg),
+		cache: map[revision]mvccpb.KeyValue{},
+		clock: sync.RWMutex{},
 
 		le: le,
 

mvcc/kvstore_txn.go

@@ -20,6 +20,7 @@ import (
 	"go.etcd.io/etcd/mvcc/mvccpb"
 	"go.etcd.io/etcd/pkg/traceutil"
 	"go.uber.org/zap"
+	"strconv"
 )
 
 type storeTxnRead struct {
@@ -140,9 +141,16 @@ func (tr *storeTxnRead) rangeKeys(key, end []byte, curRev int64, ro RangeOptions
 		limit = len(revpairs)
 	}
 
-	kvs := make([]mvccpb.KeyValue, limit)
+	kvs := make([]*mvccpb.KeyValue, limit)
 	revBytes := newRevBytes()
 	for i, revpair := range revpairs[:len(kvs)] {
+		tr.s.clock.RLock()
+		v, exists := tr.s.cache[revpair]
+		tr.s.clock.RUnlock()
+		if exists {
+			kvs[i] = &v
+			continue
+		}
 		revToBytes(revpair, revBytes)
 		_, vs := tr.tx.UnsafeRange(keyBucketName, revBytes, nil, 0)
 		if len(vs) != 1 {
@@ -156,7 +164,8 @@ func (tr *storeTxnRead) rangeKeys(key, end []byte, curRev int64, ro RangeOptions
 				plog.Fatalf("range cannot find rev (%d,%d)", revpair.main, revpair.sub)
 			}
 		}
-		if err := kvs[i].Unmarshal(vs[0]); err != nil {
+		x := mvccpb.KeyValue{}
+		if err := x.Unmarshal(vs[0]); err != nil {
 			if tr.s.lg != nil {
 				tr.s.lg.Fatal(
 					"failed to unmarshal mvccpb.KeyValue",
@@ -165,6 +174,12 @@ func (tr *storeTxnRead) rangeKeys(key, end []byte, curRev int64, ro RangeOptions
 			} else {
 				plog.Fatalf("cannot unmarshal event: %v", err)
 			}
+		} else {
+			kvs[i] = &x
+			tr.s.clock.Lock()
+			tr.s.cache[revpair] = x
+			tr.s.clock.Unlock()
+			plog.Warning("caching " + strconv.Itoa(i))
 		}
 	}
 	tr.trace.Step("range keys from bolt db")

tools/benchmark/cmd/put.go

@@ -112,7 +112,7 @@ func putFunc(cmd *cobra.Command, args []string) {
 			} else {
 				binary.PutVarint(k, int64(rand.Intn(keySpaceSize)))
 			}
-			requests <- v3.OpPut(string(k), v)
+			requests <- v3.OpPut("a"+string(k), v)
 		}
 		close(requests)
 	}()

tools/benchmark/cmd/range.go

@@ -57,6 +57,7 @@ func rangeFunc(cmd *cobra.Command, args []string) {
 	}
 
 	k := args[0]
+
 	end := ""
 	if len(args) == 2 {
 		end = args[1]