vitalif
/
etcd
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

grpcproxy: update cache based on txn response 

Fixes more hangs in TestSTMConflict.
release-3.1
Anthony Romano 2016-11-08 21:21:44 -08:00
parent dbb692e50f
commit a4dcceb8aa
2 changed files with 24 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
proxy/grpcproxy/cache/store.go vendored
View File

@ -134,14 +134,14 @@ func (c *cache) Invalidate(key, endkey []byte) {
}
ivs = c.cachedRanges.Stab(ivl)
c.cachedRanges.Delete(ivl)
for _, iv := range ivs {
keys := iv.Val.([]string)
for _, key := range keys {
c.lru.Remove(key)
}
}
// delete after removing all keys since it is destructive to 'ivs'
c.cachedRanges.Delete(ivl)
}
// Compact invalidate all caching response before the given rev.

22
proxy/grpcproxy/kv.go
View File

@ -70,6 +70,22 @@ func (p *kvProxy) DeleteRange(ctx context.Context, r *pb.DeleteRangeRequest) (*p
return (*pb.DeleteRangeResponse)(resp.Del()), err
}
func (p *kvProxy) txnToCache(reqs []*pb.RequestOp, resps []*pb.ResponseOp) {
for i := range resps {
switch tv := resps[i].Response.(type) {
case *pb.ResponseOp_ResponsePut:
p.cache.Invalidate(reqs[i].GetRequestPut().Key, nil)
case *pb.ResponseOp_ResponseDeleteRange:
rdr := reqs[i].GetRequestDeleteRange()
p.cache.Invalidate(rdr.Key, rdr.RangeEnd)
case *pb.ResponseOp_ResponseRange:
req := reqs[i].GetRequestRange()
req.Serializable = true
p.cache.Add(req, tv.ResponseRange)
}
}
}
func (p *kvProxy) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error) {
txn := p.kv.Txn(ctx)
cmps := make([]clientv3.Cmp, len(r.Compare))
@ -97,6 +113,12 @@ func (p *kvProxy) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, e
for _, cmp := range r.Compare {
p.cache.Invalidate(cmp.Key, nil)
}
// update any fetched keys
if resp.Succeeded {
p.txnToCache(r.Success, resp.Responses)
} else {
p.txnToCache(r.Failure, resp.Responses)
}
return (*pb.TxnResponse)(resp), nil
}