diff --git a/client/v3/watch.go b/client/v3/watch.go index 46adce7ba84d..5b52c44784d6 100644 --- a/client/v3/watch.go +++ b/client/v3/watch.go @@ -47,6 +47,11 @@ const ( watcherStreamBufWarningInterval = 5 * time.Second watcherStreamBufWarningThreshold = 100 * time.Millisecond + + // Watch cancel/progress requests are best-effort and can be dropped when queue is full, + // but watch create requests must be kept, otherwise the resuming queue can stall indefinitely. + sendBestEffortChanBufLen = 128 + sendCreateChanBufLen = 16 ) type Event = mvccpb.Event @@ -167,13 +172,18 @@ type watchGRPCStream struct { // substreams holds all active watchers on this grpc stream substreams map[int64]*watcherStream - // resuming holds all resuming watchers on this grpc stream + // resuming holds pending watchers to be created for the first time, + // and disconnected watchers that wait to reconnect. resuming []*watcherStream // reqc sends a watch request from Watch() to the main goroutine reqc chan watchStreamRequest // respc receives data from the watch client respc chan *pb.WatchResponse + // sendCreatec holds create requests to send to the grpc stream + sendCreatec chan *pb.WatchRequest + // sendBestEffortc holds cancel/progress requests to send to the grpc stream + sendBestEffortc chan *pb.WatchRequest // donec closes to broadcast shutdown donec chan struct{} // errc transmits errors from grpc Recv to the watch stream reconnect logic @@ -182,6 +192,8 @@ type watchGRPCStream struct { closingc chan *watcherStream // wg is Done when all substream goroutines have exited wg sync.WaitGroup + // sendWg ensures there's only one sendLoop goroutine running + sendWg sync.WaitGroup // resumec closes to signal that all substreams should begin resuming resumec chan struct{} @@ -240,6 +252,8 @@ type watcherStream struct { closing bool // id is the registered watch id on the grpc stream id int64 + // key is the key in the KV store this watcher is watching + key string // buf holds all events received from etcd but not yet consumed by the client buf []*WatchResponse @@ -510,7 +524,6 @@ func (w *watchGRPCStream) closeSubstream(ws *watcherStream) { // run is the root of the goroutines for managing a watcher client func (w *watchGRPCStream) run() { - var wc pb.Watch_WatchClient var closeErr error // substreams marked to close but goroutine still running; needed for @@ -541,7 +554,8 @@ func (w *watchGRPCStream) run() { }() // start a stream with the etcd grpc server - if wc, closeErr = w.newWatchClient(); closeErr != nil { + if closeErr = w.newWatchClient(); closeErr != nil { + w.lg.Debug("creating new watch client failed. watchGRPCStream exiting now", zap.Error(closeErr)) return } @@ -555,12 +569,12 @@ func (w *watchGRPCStream) run() { case req := <-w.reqc: switch wreq := req.(type) { case *watchRequest: - outc := make(chan WatchResponse, 1) // TODO: pass custom watch ID? ws := &watcherStream{ + key: wreq.key, initReq: *wreq, id: InvalidWatchID, - outc: outc, + outc: make(chan WatchResponse, 1), // unbuffered so resumes won't cause repeat events recvc: make(chan *WatchResponse), } @@ -576,14 +590,10 @@ func (w *watchGRPCStream) run() { w.resuming = append(w.resuming, ws) if len(w.resuming) == 1 { // head of resume queue, can register a new watcher - if err := wc.Send(ws.initReq.toPB()); err != nil { - w.lg.Debug("error when sending request", zap.Error(err)) - } + w.enqueueCreateReq(ws.initReq.toPB()) } case *progressRequest: - if err := wc.Send(wreq.toPB()); err != nil { - w.lg.Debug("error when sending request", zap.Error(err)) - } + w.enqueueBestEffortReq(wreq.toPB()) } // new events from the watch client @@ -609,9 +619,7 @@ func (w *watchGRPCStream) run() { } if ws := w.nextResume(); ws != nil { - if err := wc.Send(ws.initReq.toPB()); err != nil { - w.lg.Debug("error when sending request", zap.Error(err)) - } + w.enqueueCreateReq(ws.initReq.toPB()) } // reset for next iteration @@ -656,26 +664,29 @@ func (w *watchGRPCStream) run() { }, } req := &pb.WatchRequest{RequestUnion: cr} - w.lg.Debug("sending watch cancel request for failed dispatch", zap.Int64("watch-id", pbresp.WatchId)) - if err := wc.Send(req); err != nil { - w.lg.Debug("failed to send watch cancel request", zap.Int64("watch-id", pbresp.WatchId), zap.Error(err)) - } + w.enqueueBestEffortReq(req) } // watch client failed on Recv; spawn another if possible case err := <-w.errc: + // Drain all remaining unsent requests, otherwise these outdated messages + // will be picked up by the next newly created sendLoop. Client and server + // might not agree on what's the key for watchID=1, etc. + close(w.sendBestEffortc) + close(w.sendCreatec) + if isHaltErr(w.ctx, err) || errors.Is(ContextError(w.ctx, err), v3rpc.ErrNoLeader) { closeErr = err return } backoff = w.backoffIfUnavailable(backoff, err) - if wc, closeErr = w.newWatchClient(); closeErr != nil { + w.sendWg.Wait() + + if closeErr = w.newWatchClient(); closeErr != nil { return } if ws := w.nextResume(); ws != nil { - if err := wc.Send(ws.initReq.toPB()); err != nil { - w.lg.Debug("error when sending request", zap.Error(err)) - } + w.enqueueCreateReq(ws.initReq.toPB()) } cancelSet = make(map[int64]struct{}) @@ -699,10 +710,7 @@ func (w *watchGRPCStream) run() { }, } req := &pb.WatchRequest{RequestUnion: cr} - w.lg.Debug("sending watch cancel request for closed watcher", zap.Int64("watch-id", ws.id)) - if err := wc.Send(req); err != nil { - w.lg.Debug("failed to send watch cancel request", zap.Int64("watch-id", ws.id), zap.Error(err)) - } + w.enqueueBestEffortReq(req) } } } @@ -766,11 +774,12 @@ func (w *watchGRPCStream) unicastResponse(wr *WatchResponse, watchID int64) bool return true } -// serveWatchClient forwards messages from the grpc stream to run() -func (w *watchGRPCStream) serveWatchClient(wc pb.Watch_WatchClient) { +// recvLoop forwards watch responses from the grpc stream to run() +func (w *watchGRPCStream) recvLoop(wc pb.Watch_WatchClient) { for { resp, err := wc.Recv() if err != nil { + w.lg.Debug("failed to receive watch response from gRPC stream", zap.Error(err)) select { case w.errc <- err: case <-w.donec: @@ -785,6 +794,50 @@ func (w *watchGRPCStream) serveWatchClient(wc pb.Watch_WatchClient) { } } +// sendLoop sends watch requests (create/cancel/progress) to the grpc stream +func (w *watchGRPCStream) sendLoop(wc pb.Watch_WatchClient) { + for { + select { + case req, ok := <-w.sendCreatec: + if !ok { + return + } + if err := wc.Send(req); err != nil { + return + } + case req, ok := <-w.sendBestEffortc: + if !ok { + return + } + if err := wc.Send(req); err != nil { + return + } + case <-w.donec: + return + case <-w.ctx.Done(): + return + } + } +} + +func (w *watchGRPCStream) enqueueCreateReq(req *pb.WatchRequest) { + select { + case w.sendCreatec <- req: + case <-w.donec: + case <-w.ctx.Done(): + } +} + +func (w *watchGRPCStream) enqueueBestEffortReq(req *pb.WatchRequest) { + select { + case w.sendBestEffortc <- req: + case <-w.donec: + case <-w.ctx.Done(): + default: + w.lg.Warn("dropping outgoing watch request due to send queue full") + } +} + // serveSubstream forwards watch responses from run() to the subscriber func (w *watchGRPCStream) serveSubstream(ws *watcherStream, resumec chan struct{}) { if ws.closing { @@ -831,7 +884,8 @@ func (w *watchGRPCStream) serveSubstream(ws *watcherStream, resumec chan struct{ ws.buf = ws.buf[1:] case wr, ok := <-ws.recvc: if !ok { - // shutdown from closeSubstream + // either the entire client side watchGRPCStream is shutting down or this + // substream has been canceled. return } @@ -925,7 +979,7 @@ func (w *watchGRPCStream) newWatcherStreamBufLogger(ws *watcherStream, now func( }) } -func (w *watchGRPCStream) newWatchClient() (pb.Watch_WatchClient, error) { +func (w *watchGRPCStream) newWatchClient() error { // mark all substreams as resuming close(w.resumec) w.resumec = make(chan struct{}) @@ -963,12 +1017,19 @@ func (w *watchGRPCStream) newWatchClient() (pb.Watch_WatchClient, error) { } if err != nil { - return nil, v3rpc.Error(err) + return v3rpc.Error(err) } - // receive data from new grpc stream - go w.serveWatchClient(wc) - return wc, nil + // receive from and send to the new grpc stream + w.sendCreatec = make(chan *pb.WatchRequest, sendCreateChanBufLen) + w.sendBestEffortc = make(chan *pb.WatchRequest, sendBestEffortChanBufLen) + w.sendWg.Add(1) + go func() { + defer w.sendWg.Done() + w.sendLoop(wc) + }() + go w.recvLoop(wc) + return nil } func (w *watchGRPCStream) waitCancelSubstreams(stopc <-chan struct{}) <-chan struct{} { @@ -1071,6 +1132,7 @@ func (wr *watchRequest) toPB() *pb.WatchRequest { Filters: wr.filters, PrevKv: wr.prevKV, Fragment: wr.fragment, + WatchId: AutoWatchID, } cr := &pb.WatchRequest_CreateRequest{CreateRequest: req} return &pb.WatchRequest{RequestUnion: cr} diff --git a/server/etcdserver/api/v3rpc/watch.go b/server/etcdserver/api/v3rpc/watch.go index 1fe1cbe13c33..aa4e8d58fa26 100644 --- a/server/etcdserver/api/v3rpc/watch.go +++ b/server/etcdserver/api/v3rpc/watch.go @@ -116,12 +116,9 @@ func SetProgressReportInterval(newTimeout time.Duration) { progressReportIntervalMu.Unlock() } -// We send ctrl response inside the read loop. We do not want -// send to block read, but we still want ctrl response we sent to -// be serialized. Thus we use a buffered chan to solve the problem. -// A small buffer should be OK for most cases, since we expect the -// ctrl requests are infrequent. -const ctrlStreamBufLen = 16 +// ctrlStream buffers watch control (create / cancel) responses produced by +// recvLoop while sendLoop is busy. +const ctrlStreamBufLen = 128 // serverWatchStream is an etcd server side stream. It receives requests // from client side gRPC stream. It receives watch events from mvcc.WatchStream,