Browse Source

concurrency: use current revisions for election

Watching from the leader's ModRevision could cause live-locking on
observe retry loops when the ModRevision is less than the compacted
revision. Instead, start watching the leader from at least the store
revision of the linearized read used to detect the current leader.

Fixes #7815
Anthony Romano 8 years ago
parent
commit
50f29bd661
1 changed files with 14 additions and 7 deletions
  1. 14 7
      clientv3/concurrency/election.go

+ 14 - 7
clientv3/concurrency/election.go

@@ -165,15 +165,14 @@ func (e *Election) observe(ctx context.Context, ch chan<- v3.GetResponse) {
 	client := e.session.Client()
 	client := e.session.Client()
 
 
 	defer close(ch)
 	defer close(ch)
-	lastRev := int64(0)
 	for {
 	for {
-		opts := append(v3.WithFirstCreate(), v3.WithRev(lastRev))
-		resp, err := client.Get(ctx, e.keyPrefix, opts...)
+		resp, err := client.Get(ctx, e.keyPrefix, v3.WithFirstCreate()...)
 		if err != nil {
 		if err != nil {
 			return
 			return
 		}
 		}
 
 
 		var kv *mvccpb.KeyValue
 		var kv *mvccpb.KeyValue
+		var hdr *pb.ResponseHeader
 
 
 		if len(resp.Kvs) == 0 {
 		if len(resp.Kvs) == 0 {
 			cctx, cancel := context.WithCancel(ctx)
 			cctx, cancel := context.WithCancel(ctx)
@@ -189,18 +188,27 @@ func (e *Election) observe(ctx context.Context, ch chan<- v3.GetResponse) {
 				// only accept PUTs; a DELETE will make observe() spin
 				// only accept PUTs; a DELETE will make observe() spin
 				for _, ev := range wr.Events {
 				for _, ev := range wr.Events {
 					if ev.Type == mvccpb.PUT {
 					if ev.Type == mvccpb.PUT {
-						kv = ev.Kv
+						hdr, kv = &wr.Header, ev.Kv
+						// may have multiple revs; hdr.rev = the last rev
+						// set to kv's rev in case batch has multiple PUTs
+						hdr.Revision = kv.ModRevision
 						break
 						break
 					}
 					}
 				}
 				}
 			}
 			}
 			cancel()
 			cancel()
 		} else {
 		} else {
-			kv = resp.Kvs[0]
+			hdr, kv = resp.Header, resp.Kvs[0]
+		}
+
+		select {
+		case ch <- v3.GetResponse{Header: hdr, Kvs: []*mvccpb.KeyValue{kv}}:
+		case <-ctx.Done():
+			return
 		}
 		}
 
 
 		cctx, cancel := context.WithCancel(ctx)
 		cctx, cancel := context.WithCancel(ctx)
-		wch := client.Watch(cctx, string(kv.Key), v3.WithRev(kv.ModRevision))
+		wch := client.Watch(cctx, string(kv.Key), v3.WithRev(hdr.Revision+1))
 		keyDeleted := false
 		keyDeleted := false
 		for !keyDeleted {
 		for !keyDeleted {
 			wr, ok := <-wch
 			wr, ok := <-wch
@@ -209,7 +217,6 @@ func (e *Election) observe(ctx context.Context, ch chan<- v3.GetResponse) {
 			}
 			}
 			for _, ev := range wr.Events {
 			for _, ev := range wr.Events {
 				if ev.Type == mvccpb.DELETE {
 				if ev.Type == mvccpb.DELETE {
-					lastRev = ev.Kv.ModRevision
 					keyDeleted = true
 					keyDeleted = true
 					break
 					break
 				}
 				}