Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

clientv3: cancel watches proactively on client context cancellation #11850

Merged
merged 1 commit into from
May 7, 2020
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
clientv3: cancel watches proactively on client context cancellation
Currently, watch cancel requests are only sent to the server after a
message comes through on a watch where the client has cancelled. This
means that cancelled watches that don't receive any new messages are
never cancelled; they persist for the lifetime of the client stream.
This has negative connotations for locking applications where a watch
may observe a key which might never change again after cancellation,
leading to many accumulating watches on the server.

By cancelling proactively, in most cases we simply move the cancel
request to happen earlier, and additionally we solve the case where the
cancel request would never be sent.

Fixes #9416
Heavy inspiration drawn from the solutions proposed there.
  • Loading branch information
jackkleeman committed May 7, 2020
commit 87aa5a9e60dd9da84de1ab10b2d1562d703dba72
18 changes: 17 additions & 1 deletion clientv3/watch.go
Original file line number Diff line number Diff line change
@@ -624,8 +624,9 @@ func (w *watchGrpcStream) run() {
},
}
req := &pb.WatchRequest{RequestUnion: cr}
lg.Info("sending watch cancel request for failed dispatch", zap.Int64("watch-id", pbresp.WatchId))
if err := wc.Send(req); err != nil {
lg.Warningf("error when sending request: %v", err)
lg.Warning("failed to send watch cancel request", zap.Int64("watch-id", pbresp.WatchId), zap.Error(err))
}
}

@@ -649,6 +650,21 @@ func (w *watchGrpcStream) run() {
return

case ws := <-w.closingc:
if ws.id != -1 {
// client is closing an established watch; close it on the server proactively instead of waiting
// to close when the next message arrives
cancelSet[ws.id] = struct{}{}
cr := &pb.WatchRequest_CancelRequest{
CancelRequest: &pb.WatchCancelRequest{
WatchId: ws.id,
},
}
req := &pb.WatchRequest{RequestUnion: cr}
lg.Info("sending watch cancel request for closed watcher", zap.Int64("watch-id", ws.id))
if err := wc.Send(req); err != nil {
jackkleeman marked this conversation as resolved.
Show resolved Hide resolved
lg.Warning("failed to send watch cancel request", zap.Int64("watch-id", ws.id), zap.Error(err))
}
}
w.closeSubstream(ws)
delete(closing, ws)
// no more watchers on this stream, shutdown
32 changes: 32 additions & 0 deletions integration/v3_watch_test.go
Original file line number Diff line number Diff line change
@@ -1213,3 +1213,35 @@ func TestV3WatchWithPrevKV(t *testing.T) {
}
}
}

// TestV3WatchCancellation ensures that watch cancellation frees up server resources.
func TestV3WatchCancellation(t *testing.T) {
clus := NewClusterV3(t, &ClusterConfig{Size: 1})
defer clus.Terminate(t)

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

cli := clus.RandClient()

// increment watcher total count and keep a stream open
cli.Watch(ctx, "/foo")
jackkleeman marked this conversation as resolved.
Show resolved Hide resolved

for i := 0; i < 1000; i++ {
ctx, cancel := context.WithCancel(ctx)
cli.Watch(ctx, "/foo")
cancel()
}

// Wait a little for cancellations to take hold
time.Sleep(3 * time.Second)

minWatches, err := clus.Members[0].Metric("etcd_debugging_mvcc_watcher_total")
if err != nil {
t.Fatal(err)
}

if minWatches != "1" {
t.Fatalf("expected one watch, got %s", minWatches)
}
}