clientv3: simplify watcher synchronization #6525

heyitsanthony · 2016-09-26T23:45:41Z

Was more complicated than it needed to be.

Also fixes clobbering id's on resume and losing watcher channels when watchers are disconnected and canceled.

/cc @hongchaodeng

gyuho · 2016-09-27T13:13:12Z

Also fixes clobbering id's on

Do we need to backport this then?

heyitsanthony · 2016-09-27T16:17:51Z

@gyuho probably, it fixes the new test case

gyuho · 2016-09-27T16:24:50Z

clientv3/watch.go

@@ -284,6 +284,9 @@ func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) Watch
 	if ok {
 		select {
 		case ret := <-retc:
+			if ret == nil {


When does this happen? Can we add some comment? Thanks!

this happens if retc is closed, I changed it to ret, ok := <-retc so it's clearer that's what it's checking

xiang90 · 2016-09-27T17:20:25Z

clientv3/watch.go

@@ -118,8 +118,8 @@ type watchGrpcStream struct {

 	// mu protects the streams map
 	mu sync.RWMutex


it seems that this mu also protects closeErr?

xiang90 · 2016-09-27T17:20:53Z

clientv3/watch.go

@@ -118,8 +118,8 @@ type watchGrpcStream struct {

 	// mu protects the streams map
 	mu sync.RWMutex
-	// streams holds all active watchers
-	streams map[int64]*watcherStream
+	// substreams holds all active watchers


holds all active gRPC streams for watchers?

That's in watcher.streams, this tracks the watch id's that are in a single grpc stream (hence the rename from streams to substreams, using "streams" twice is super confusing). Will clarify the comments a bit.

xiang90 · 2016-09-27T17:21:12Z

clientv3/watch.go

-	w.mu.Lock()
-	w.streams[ws.id] = ws
-	w.mu.Unlock()
+	w.substreams[ws.id] = ws


why do we drop the lock here?

xiang90 · 2016-09-27T17:21:31Z

clientv3/watch.go

-	}
-	w.mu.Unlock()
-	return empty
+	delete(w.substreams, ws.id)


why do we drop the lock here?

xiang90 · 2016-09-27T17:22:21Z

clientv3/watch.go

+			w.closeStream(ws)
+		}
+
+		w.owner.mu.Lock()


probably add a delStream func on owner?

xiang90 · 2016-09-27T17:22:52Z

clientv3/watch.go

-	w.mu.RLock()
-	defer w.mu.RUnlock()
-	ws, ok := w.streams[pbresp.WatchId]
+	ws, ok := w.substreams[pbresp.WatchId]


why do we drop the lock here?

xiang90 · 2016-09-27T17:35:15Z

@heyitsanthony OK. After read the whole thing, I understand that we want to serialize the closing event in the single run routine. Can we update the description of the mutex then?

heyitsanthony · 2016-09-28T22:30:23Z

there's still some intrinsic raceiness between the resume path and stream cancelation, going to rework this a little bit more so the substream goroutines don't have to reason about resume

heyitsanthony · 2016-09-30T18:47:10Z

Tore up the watcher code from the roots. The old code's resume path was totally broken and could livelock in some cases. Now, all watch registrations go through a resuming queue on watchGrpcStream. On reconnect, all watcherStream goroutines are stopped so the resuming revision is stable, then restarted and requeued on resuming. Overall the design is much more convincing than the old mess.

PTAL /cc @xiang90 @gyuho

gyuho

LGTM.

TestWatchWithRequireLeader fails?

Defer to @xiang90

Thanks!

gyuho · 2016-09-30T18:51:10Z

clientv3/watch.go

+	donec chan struct{}
+	// closing is set to true when stream should be scheduled to shutdown.
+	closing bool
+	// id is the registered watch id for on the grpc stream


s/for on/on/?

right, will fix

gyuho · 2016-09-30T18:53:26Z

clientv3/watch.go

@@ -314,12 +320,7 @@ func (w *watcher) Close() (err error) {
 }

 func (w *watchGrpcStream) Close() (err error) {
-	w.mu.Lock()


Q. Why do we remove this now? Was this causing any problem?

The sharing was making it unnecessarily difficult to reason about the watcherStream teardown path so now watcherStream will post itself to watchGrpcStream.closingc and watchGrpcStream.run handles the final removal of watcherStream resources.

xiang90 · 2016-10-03T06:01:06Z

clientv3/watch.go

+		w.resumec = make(chan struct{})
+		w.joinSubstreams()
+		for _, ws := range w.substreams {
+			ws.id = -1


do we need to update the donec of wc as what we do for the resuming streams at line 654?

yes, but ws is appended into resuming and the following loop over resuming picks it up, so it's unnecessary here

xiang90 · 2016-10-03T06:02:25Z

clientv3/watch.go

+// streams are marked as nil in the queue since the head must wait for its inflight registration.
+func (w *watchGrpcStream) nextResume() *watcherStream {
+	for len(w.resuming) != 0 {
+		if w.resuming[0] != nil {


is it possible to handle the nil case outside the next resume? it seems like we can always remove the abandoned resuming at the caller side, no?

It's possible, but the caller side would have to reimplement this loop every time. I'd rather have it done in one place.

xiang90 · 2016-10-03T06:14:50Z

clientv3/watch.go

@@ -283,7 +288,10 @@ func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) Watch
 	// receive channel
 	if ok {
 		select {
-		case ret := <-retc:
+		case ret, ok := <-wr.retc:


can we add some comments for this? it is not clear to me why do we need to call watch recursively.

i failed to figure out where we might close retc.

it doesn't any more; removed the close case

xiang90 · 2016-10-03T21:31:52Z

LGTM. Fix CI?

Was more complicated than it needed to be and didn't really work in the first place. Restructured watcher registation to use a queue.

gyuho added the backport/v3.3 label Sep 27, 2016

gyuho reviewed Sep 27, 2016

View reviewed changes

xiang90 reviewed Sep 27, 2016

View reviewed changes

clientv3/watch.go

w.mu.Lock()

w.streams[ws.id] = ws

w.mu.Unlock()

w.substreams[ws.id] = ws

Copy link

Contributor

xiang90 Sep 27, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we drop the lock here?

xiang90 reviewed Sep 27, 2016

View reviewed changes

clientv3/watch.go

}

w.mu.Unlock()

return empty

delete(w.substreams, ws.id)

Copy link

Contributor

xiang90 Sep 27, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we drop the lock here?

xiang90 reviewed Sep 27, 2016

View reviewed changes

clientv3/watch.go

w.closeStream(ws)

}

w.owner.mu.Lock()

Copy link

Contributor

xiang90 Sep 27, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

probably add a delStream func on owner?

xiang90 reviewed Sep 27, 2016

View reviewed changes

heyitsanthony added the WIP label Sep 28, 2016

heyitsanthony force-pushed the watcher-disconn branch from 40f82e0 to a6606b1 Compare September 30, 2016 18:35

heyitsanthony removed the WIP label Sep 30, 2016

gyuho approved these changes Sep 30, 2016

View reviewed changes

heyitsanthony force-pushed the watcher-disconn branch 3 times, most recently from 6a3b548 to a64d996 Compare September 30, 2016 23:50

xiang90 reviewed Oct 3, 2016

View reviewed changes

heyitsanthony force-pushed the watcher-disconn branch from a64d996 to a8aaefa Compare October 3, 2016 19:26

Anthony Romano added 2 commits October 3, 2016 16:56

integration: test canceling watchers when disconnected

9ce398f

clientv3: simplify watch synchronization

5b50658

Was more complicated than it needed to be and didn't really work in the first place. Restructured watcher registation to use a queue.

heyitsanthony force-pushed the watcher-disconn branch from a8aaefa to 5b50658 Compare October 4, 2016 17:49

heyitsanthony merged commit 19122b4 into etcd-io:master Oct 4, 2016

gyuho mentioned this pull request Oct 4, 2016

*: backporting clientv3 watcher change to release-3.0 branch #6581

Closed

heyitsanthony deleted the watcher-disconn branch October 4, 2016 23:16

gyuho removed the backport/v3.3 label Nov 22, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

clientv3: simplify watcher synchronization #6525

clientv3: simplify watcher synchronization #6525

heyitsanthony commented Sep 26, 2016

gyuho commented Sep 27, 2016

heyitsanthony commented Sep 27, 2016

gyuho Sep 27, 2016

heyitsanthony Sep 28, 2016

xiang90 Sep 27, 2016

xiang90 Sep 27, 2016

heyitsanthony Sep 28, 2016

xiang90 Sep 27, 2016

xiang90 Sep 27, 2016

xiang90 Sep 27, 2016

xiang90 Sep 27, 2016

xiang90 commented Sep 27, 2016

heyitsanthony commented Sep 28, 2016

heyitsanthony commented Sep 30, 2016

gyuho left a comment

gyuho Sep 30, 2016

heyitsanthony Sep 30, 2016

gyuho Sep 30, 2016

heyitsanthony Sep 30, 2016

xiang90 Oct 3, 2016

heyitsanthony Oct 3, 2016

xiang90 Oct 3, 2016

xiang90 Oct 3, 2016

heyitsanthony Oct 3, 2016

xiang90 Oct 3, 2016

xiang90 Oct 3, 2016

xiang90 Oct 3, 2016

heyitsanthony Oct 3, 2016

xiang90 commented Oct 3, 2016

		@@ -118,8 +118,8 @@ type watchGrpcStream struct {

		// mu protects the streams map
		mu sync.RWMutex

clientv3: simplify watcher synchronization #6525

clientv3: simplify watcher synchronization #6525

Conversation

heyitsanthony commented Sep 26, 2016

gyuho commented Sep 27, 2016

heyitsanthony commented Sep 27, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xiang90 commented Sep 27, 2016

heyitsanthony commented Sep 28, 2016

heyitsanthony commented Sep 30, 2016

gyuho left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xiang90 commented Oct 3, 2016