Skip to content

Commit

Permalink
etcdserver: use linearizableReadNotify for txn
Browse files Browse the repository at this point in the history
  • Loading branch information
xiang90 committed Sep 28, 2016
1 parent 1e1dd24 commit ef5dd26
Showing 1 changed file with 50 additions and 0 deletions.
50 changes: 50 additions & 0 deletions etcdserver/v3_server.go
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,42 @@ func (s *EtcdServer) DeleteRange(ctx context.Context, r *pb.DeleteRangeRequest)
}

func (s *EtcdServer) Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error) {
// TODO: remove this checking when we release etcd 3.2
if s.ClusterVersion() == nil || s.ClusterVersion().LessThan(newRangeClusterVersion) {
return s.legacyTxn(ctx, r)
}

if isTxnReadonly(r) {
if !isTxnSerializable(r) {
err := s.linearizableReadNotify(ctx)
if err != nil {
return nil, err
}
}
var resp *pb.TxnResponse
var err error
chk := func(ai *auth.AuthInfo) error {
return checkTxnAuth(s.authStore, ai, r)
}
get := func() { resp, err = s.applyV3Base.Txn(r) }
if serr := s.doSerialize(ctx, chk, get); serr != nil {
return nil, serr
}
return resp, err
}
// TODO: readonly Txn do not need to go through raft
result, err := s.processInternalRaftRequest(ctx, pb.InternalRaftRequest{Txn: r})
if err != nil {
return nil, err
}
if result.err != nil {
return nil, result.err
}
return result.resp.(*pb.TxnResponse), nil
}

// TODO: remove this func when we release etcd 3.2
func (s *EtcdServer) legacyTxn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error) {
if isTxnSerializable(r) {
var resp *pb.TxnResponse
var err error
Expand Down Expand Up @@ -202,6 +238,20 @@ func isTxnSerializable(r *pb.TxnRequest) bool {
return true
}

func isTxnReadonly(r *pb.TxnRequest) bool {
for _, u := range r.Success {
if r := u.GetRequestRange(); r == nil {
return false
}
}
for _, u := range r.Failure {
if r := u.GetRequestRange(); r == nil {
return false
}
}
return true
}

func (s *EtcdServer) Compact(ctx context.Context, r *pb.CompactionRequest) (*pb.CompactionResponse, error) {
result, err := s.processInternalRaftRequestOnce(ctx, pb.InternalRaftRequest{Compaction: r})
if r.Physical && result != nil && result.physc != nil {
Expand Down

0 comments on commit ef5dd26

Please sign in to comment.