From 3527b3b224040159dc22bc94510c067f1927d17d Mon Sep 17 00:00:00 2001 From: Benjamin Wang Date: Fri, 17 Jan 2025 19:16:12 +0000 Subject: [PATCH 1/3] Enhance method (*serveCtx) serve to wait for all goroutines to complete before it returns Signed-off-by: Benjamin Wang --- server/embed/serve.go | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/server/embed/serve.go b/server/embed/serve.go index c888e3b1cfb..5615a0c2120 100644 --- a/server/embed/serve.go +++ b/server/embed/serve.go @@ -68,6 +68,8 @@ type serveCtx struct { serviceRegister func(*grpc.Server) serversC chan *servers closeOnce sync.Once + + wg sync.WaitGroup } type servers struct { @@ -182,13 +184,17 @@ func (sctx *serveCtx) serve( server = m.Serve httpl := m.Match(cmux.HTTP1()) + sctx.wg.Add(1) go func(srvhttp *http.Server, tlsLis net.Listener) { + defer sctx.wg.Done() errHandler(srvhttp.Serve(tlsLis)) }(srv, httpl) if grpcEnabled { grpcl := m.Match(cmux.HTTP2()) + sctx.wg.Add(1) go func(gs *grpc.Server, l net.Listener) { + defer sctx.wg.Done() errHandler(gs.Serve(l)) }(gs, grpcl) } @@ -237,7 +243,7 @@ func (sctx *serveCtx) serve( TLSConfig: tlscfg, ErrorLog: logger, // do not log user error } - if err := configureHTTPServer(srv, s.Cfg); err != nil { + if err = configureHTTPServer(srv, s.Cfg); err != nil { sctx.lg.Error("Configure https server failed", zap.Error(err)) return err } @@ -248,11 +254,13 @@ func (sctx *serveCtx) serve( } else { server = m.Serve - tlsl, err := transport.NewTLSListener(m.Match(cmux.Any()), tlsinfo) - if err != nil { - return err + tlsl, tlsErr := transport.NewTLSListener(m.Match(cmux.Any()), tlsinfo) + if tlsErr != nil { + return tlsErr } + sctx.wg.Add(1) go func(srvhttp *http.Server, tlsl net.Listener) { + defer sctx.wg.Done() errHandler(srvhttp.Serve(tlsl)) }(srv, tlsl) } @@ -265,7 +273,11 @@ func (sctx *serveCtx) serve( ) } - return server() + err = server() + sctx.close() + // ensure all goroutines, which are created by this method, to complete before this method returns. + sctx.wg.Wait() + return err } func configureHTTPServer(srv *http.Server, cfg config.ServerConfig) error { @@ -334,7 +346,9 @@ func (sctx *serveCtx) registerGateway(dial func(ctx context.Context) (*grpc.Clie return nil, err } } + sctx.wg.Add(1) go func() { + defer sctx.wg.Done() <-ctx.Done() if cerr := conn.Close(); cerr != nil { sctx.lg.Warn( From 86a31702c4a54829787edd87e7d3d9ae1f10c801 Mon Sep 17 00:00:00 2001 From: Benjamin Wang Date: Fri, 17 Jan 2025 19:16:47 +0000 Subject: [PATCH 2/3] Ensure all goroutines created by StartEtcd to complete before closing the errc Signed-off-by: Benjamin Wang --- server/embed/etcd.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/server/embed/etcd.go b/server/embed/etcd.go index f5fa2c2c692..fe2450401de 100644 --- a/server/embed/etcd.go +++ b/server/embed/etcd.go @@ -607,7 +607,9 @@ func (e *Etcd) servePeers() { // start peer servers in a goroutine for _, pl := range e.Peers { + e.wg.Add(1) go func(l *peerListener) { + defer e.wg.Done() u := l.Addr().String() e.cfg.logger.Info( "serving peer traffic", @@ -774,7 +776,9 @@ func (e *Etcd) serveClients() { // start client servers in each goroutine for _, sctx := range e.sctxs { + e.wg.Add(1) go func(s *serveCtx) { + defer e.wg.Done() e.errHandler(s.serve(e.Server, &e.cfg.ClientTLSInfo, mux, e.errHandler, e.grpcGatewayDial(splitHTTP), splitHTTP, gopts...)) }(sctx) } @@ -859,7 +863,9 @@ func (e *Etcd) serveMetrics() (err error) { return err } e.metricsListeners = append(e.metricsListeners, ml) + e.wg.Add(1) go func(u url.URL, ln net.Listener) { + defer e.wg.Done() e.cfg.logger.Info( "serving metrics", zap.String("address", u.String()), @@ -872,9 +878,6 @@ func (e *Etcd) serveMetrics() (err error) { } func (e *Etcd) errHandler(err error) { - e.wg.Add(1) - defer e.wg.Done() - if err != nil { e.GetLogger().Error("setting up serving from embedded etcd failed.", zap.Error(err)) } From 201568ac45057ab1aaabf588f80d6823c769f371 Mon Sep 17 00:00:00 2001 From: Benjamin Wang Date: Tue, 21 Jan 2025 15:46:01 +0000 Subject: [PATCH 3/3] add commment to clarify the etcd shutting down workflow Signed-off-by: Benjamin Wang --- server/embed/etcd.go | 37 +++++++++++++++++++++++++++++++++---- server/embed/serve.go | 12 ++++++++++-- 2 files changed, 43 insertions(+), 6 deletions(-) diff --git a/server/embed/etcd.go b/server/embed/etcd.go index fe2450401de..fb610e0f42d 100644 --- a/server/embed/etcd.go +++ b/server/embed/etcd.go @@ -79,12 +79,23 @@ type Etcd struct { Server *etcdserver.EtcdServer - cfg Config - stopc chan struct{} - errc chan error + cfg Config + // closeOnce is to ensure `stopc` is closed only once, no matter + // how many times the Close() method is called. closeOnce sync.Once - wg sync.WaitGroup + // stopc is used to notify the sub goroutines not to send + // any errors to `errc`. + stopc chan struct{} + // errc is used to receive error from sub goroutines (including + // client handler, peer handler and metrics handler). It's closed + // after all these sub goroutines exit (checked via `wg`). Writers + // should avoid writing after `stopc` is closed by selecting on + // reading from `stopc`. + errc chan error + + // wg is used to track the lifecycle of all sub goroutines created by `StartEtcd`. + wg sync.WaitGroup } type peerListener struct { @@ -388,6 +399,24 @@ func (e *Etcd) Config() Config { // Close gracefully shuts down all servers/listeners. // Client requests will be terminated with request timeout. // After timeout, enforce remaning requests be closed immediately. +// +// The rough workflow to shut down etcd: +// 1. close the `stopc` channel, so that all error handlers (child +// goroutines) won't send back any errors anymore; +// 2. stop the http and grpc servers gracefully, within request timeout; +// 3. close all client and metrics listeners, so that etcd server +// stops receiving any new connection; +// 4. call the cancel function to close the gateway context, so that +// all gateway connections are closed. +// 5. stop etcd server gracefully, and ensure the main raft loop +// goroutine is stopped; +// 6. stop all peer listeners, so that it stops receiving peer connections +// and messages (wait up to 1-second); +// 7. wait for all child goroutines (i.e. client handlers, peer handlers +// and metrics handlers) to exit; +// 8. close the `errc` channel to release the resource. Note that it's only +// safe to close the `errc` after step 7 above is done, otherwise the +// child goroutines may send errors back to already closed `errc` channel. func (e *Etcd) Close() { fields := []zap.Field{ zap.String("name", e.cfg.Name), diff --git a/server/embed/serve.go b/server/embed/serve.go index 5615a0c2120..ad7307666b0 100644 --- a/server/embed/serve.go +++ b/server/embed/serve.go @@ -61,14 +61,22 @@ type serveCtx struct { insecure bool httpOnly bool + // ctx is used to control the grpc gateway. Terminate the grpc gateway + // by calling `cancel` when shutting down the etcd. ctx context.Context cancel context.CancelFunc userHandlers map[string]http.Handler serviceRegister func(*grpc.Server) - serversC chan *servers - closeOnce sync.Once + // serversC is used to receive the http and grpc server objects (created + // in `serve`), both of which will be closed when shutting down the etcd. + // Close it when `serve` returns or when etcd fails to bootstrap. + serversC chan *servers + // closeOnce is to ensure `serversC` is closed only once. + closeOnce sync.Once + + // wg is used to track the lifecycle of all sub goroutines created by `serve`. wg sync.WaitGroup }