mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
etcdserver: use same ReadView for read-only txns
A read-only txn isn't serialized by raft, but it uses a fresh read txn for every mvcc access prior to executing its request ops. If a write txn modifies the keys matching the read txn's comparisons, the read txn may return inconsistent results. To fix, use the same read-only mvcc txn for the duration of the etcd txn. Probably gets a modest txn speedup as well since there are fewer read txn allocations.
This commit is contained in:
parent
da48f1feaf
commit
d173b09a1b
@ -319,33 +319,36 @@ func (a *applierV3backend) Range(txn mvcc.TxnRead, r *pb.RangeRequest) (*pb.Rang
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (a *applierV3backend) Txn(rt *pb.TxnRequest) (*pb.TxnResponse, error) {
|
func (a *applierV3backend) Txn(rt *pb.TxnRequest) (*pb.TxnResponse, error) {
|
||||||
ok := true
|
isWrite := !isTxnReadonly(rt)
|
||||||
for _, c := range rt.Compare {
|
txn := mvcc.NewReadOnlyTxnWrite(a.s.KV().Read())
|
||||||
if _, ok = a.applyCompare(c); !ok {
|
|
||||||
break
|
reqs, ok := a.compareToOps(txn, rt)
|
||||||
|
if isWrite {
|
||||||
|
if err := a.checkRequestPut(txn, reqs); err != nil {
|
||||||
|
txn.End()
|
||||||
|
return nil, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if err := checkRequestRange(txn, reqs); err != nil {
|
||||||
var reqs []*pb.RequestOp
|
txn.End()
|
||||||
if ok {
|
|
||||||
reqs = rt.Success
|
|
||||||
} else {
|
|
||||||
reqs = rt.Failure
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := a.checkRequestPut(reqs); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
if err := a.checkRequestRange(reqs); err != nil {
|
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
resps := make([]*pb.ResponseOp, len(reqs))
|
resps := make([]*pb.ResponseOp, len(reqs))
|
||||||
|
txnResp := &pb.TxnResponse{
|
||||||
|
Responses: resps,
|
||||||
|
Succeeded: ok,
|
||||||
|
Header: &pb.ResponseHeader{},
|
||||||
|
}
|
||||||
|
|
||||||
// When executing the operations of txn, etcd must hold the txn lock so
|
// When executing mutable txn ops, etcd must hold the txn lock so
|
||||||
// readers do not see any intermediate results.
|
// readers do not see any intermediate results. Since writes are
|
||||||
// TODO: use Read txn if only Ranges
|
// serialized on the raft loop, the revision in the read view will
|
||||||
txn := a.s.KV().Write()
|
// be the revision of the write txn.
|
||||||
|
if isWrite {
|
||||||
|
txn.End()
|
||||||
|
txn = a.s.KV().Write()
|
||||||
|
}
|
||||||
for i := range reqs {
|
for i := range reqs {
|
||||||
resps[i] = a.applyUnion(txn, reqs[i])
|
resps[i] = a.applyUnion(txn, reqs[i])
|
||||||
}
|
}
|
||||||
@ -355,23 +358,25 @@ func (a *applierV3backend) Txn(rt *pb.TxnRequest) (*pb.TxnResponse, error) {
|
|||||||
}
|
}
|
||||||
txn.End()
|
txn.End()
|
||||||
|
|
||||||
txnResp := &pb.TxnResponse{}
|
|
||||||
txnResp.Header = &pb.ResponseHeader{}
|
|
||||||
txnResp.Header.Revision = rev
|
txnResp.Header.Revision = rev
|
||||||
txnResp.Responses = resps
|
|
||||||
txnResp.Succeeded = ok
|
|
||||||
return txnResp, nil
|
return txnResp, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// applyCompare applies the compare request.
|
func (a *applierV3backend) compareToOps(rv mvcc.ReadView, rt *pb.TxnRequest) ([]*pb.RequestOp, bool) {
|
||||||
// It returns the revision at which the comparison happens. If the comparison
|
for _, c := range rt.Compare {
|
||||||
// succeeds, the it returns true. Otherwise it returns false.
|
if !applyCompare(rv, c) {
|
||||||
func (a *applierV3backend) applyCompare(c *pb.Compare) (int64, bool) {
|
return rt.Failure, false
|
||||||
rr, err := a.s.KV().Range(c.Key, nil, mvcc.RangeOptions{})
|
}
|
||||||
rev := rr.Rev
|
}
|
||||||
|
return rt.Success, true
|
||||||
|
}
|
||||||
|
|
||||||
|
// applyCompare applies the compare request.
|
||||||
|
// If the comparison succeeds, it returns true. Otherwise, returns false.
|
||||||
|
func applyCompare(rv mvcc.ReadView, c *pb.Compare) bool {
|
||||||
|
rr, err := rv.Range(c.Key, nil, mvcc.RangeOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return rev, false
|
return false
|
||||||
}
|
}
|
||||||
var ckv mvccpb.KeyValue
|
var ckv mvccpb.KeyValue
|
||||||
if len(rr.KVs) != 0 {
|
if len(rr.KVs) != 0 {
|
||||||
@ -383,7 +388,7 @@ func (a *applierV3backend) applyCompare(c *pb.Compare) (int64, bool) {
|
|||||||
// We can treat non-existence as the empty set explicitly, such that
|
// We can treat non-existence as the empty set explicitly, such that
|
||||||
// even a key with a value of length 0 bytes is still a real key
|
// even a key with a value of length 0 bytes is still a real key
|
||||||
// that was written that way
|
// that was written that way
|
||||||
return rev, false
|
return false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -415,23 +420,15 @@ func (a *applierV3backend) applyCompare(c *pb.Compare) (int64, bool) {
|
|||||||
|
|
||||||
switch c.Result {
|
switch c.Result {
|
||||||
case pb.Compare_EQUAL:
|
case pb.Compare_EQUAL:
|
||||||
if result != 0 {
|
return result == 0
|
||||||
return rev, false
|
|
||||||
}
|
|
||||||
case pb.Compare_NOT_EQUAL:
|
case pb.Compare_NOT_EQUAL:
|
||||||
if result == 0 {
|
return result != 0
|
||||||
return rev, false
|
|
||||||
}
|
|
||||||
case pb.Compare_GREATER:
|
case pb.Compare_GREATER:
|
||||||
if result != 1 {
|
return result > 0
|
||||||
return rev, false
|
|
||||||
}
|
|
||||||
case pb.Compare_LESS:
|
case pb.Compare_LESS:
|
||||||
if result != -1 {
|
return result < 0
|
||||||
return rev, false
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
return rev, true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *applierV3backend) applyUnion(txn mvcc.TxnWrite, union *pb.RequestOp) *pb.ResponseOp {
|
func (a *applierV3backend) applyUnion(txn mvcc.TxnWrite, union *pb.RequestOp) *pb.ResponseOp {
|
||||||
@ -771,7 +768,7 @@ func (s *kvSortByValue) Less(i, j int) bool {
|
|||||||
return bytes.Compare(s.kvs[i].Value, s.kvs[j].Value) < 0
|
return bytes.Compare(s.kvs[i].Value, s.kvs[j].Value) < 0
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *applierV3backend) checkRequestPut(reqs []*pb.RequestOp) error {
|
func (a *applierV3backend) checkRequestPut(rv mvcc.ReadView, reqs []*pb.RequestOp) error {
|
||||||
for _, requ := range reqs {
|
for _, requ := range reqs {
|
||||||
tv, ok := requ.Request.(*pb.RequestOp_RequestPut)
|
tv, ok := requ.Request.(*pb.RequestOp_RequestPut)
|
||||||
if !ok {
|
if !ok {
|
||||||
@ -783,7 +780,7 @@ func (a *applierV3backend) checkRequestPut(reqs []*pb.RequestOp) error {
|
|||||||
}
|
}
|
||||||
if preq.IgnoreValue || preq.IgnoreLease {
|
if preq.IgnoreValue || preq.IgnoreLease {
|
||||||
// expects previous key-value, error if not exist
|
// expects previous key-value, error if not exist
|
||||||
rr, err := a.s.KV().Range(preq.Key, nil, mvcc.RangeOptions{})
|
rr, err := rv.Range(preq.Key, nil, mvcc.RangeOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -801,7 +798,7 @@ func (a *applierV3backend) checkRequestPut(reqs []*pb.RequestOp) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *applierV3backend) checkRequestRange(reqs []*pb.RequestOp) error {
|
func checkRequestRange(rv mvcc.ReadView, reqs []*pb.RequestOp) error {
|
||||||
for _, requ := range reqs {
|
for _, requ := range reqs {
|
||||||
tv, ok := requ.Request.(*pb.RequestOp_RequestRange)
|
tv, ok := requ.Request.(*pb.RequestOp_RequestRange)
|
||||||
if !ok {
|
if !ok {
|
||||||
@ -812,10 +809,10 @@ func (a *applierV3backend) checkRequestRange(reqs []*pb.RequestOp) error {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
if greq.Revision > a.s.KV().Rev() {
|
if greq.Revision > rv.Rev() {
|
||||||
return mvcc.ErrFutureRev
|
return mvcc.ErrFutureRev
|
||||||
}
|
}
|
||||||
if greq.Revision < a.s.KV().FirstRev() {
|
if greq.Revision < rv.FirstRev() {
|
||||||
return mvcc.ErrCompacted
|
return mvcc.ErrCompacted
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Loading…
x
Reference in New Issue
Block a user