From 9d15705c165313c23a1364ab32880846d36de93d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 16 Apr 2026 16:11:01 -0700 Subject: [PATCH] fix(mini): shut down admin/s3/webdav/filer before volume/master on Ctrl+C (#9112) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(mini): shut down admin/s3/webdav/filer before volume/master on Ctrl+C Interrupts fired grace hooks in registration order, so master (started first) shut down before its clients, producing heartbeat-canceled errors and masterClient reconnection noise during weed mini shutdown. Admin/s3/ webdav had no interrupt hooks at all and were killed at os.Exit. - grace: execute interrupt hooks in LIFO (defer-style) order so later- started services tear down first. - filer: consolidate the three separate interrupt hooks (gRPC / HTTP / DB) into one that runs in order, so filer shutdown stays correct independent of FIFO/LIFO semantics. - mini: add MiniClientsShutdownCtx (separate from test-facing MiniClusterCtx) plus an OnMiniClientsShutdown helper. Admin, S3, WebDAV and the maintenance worker observe it; runMini registers a cancel hook after startup so under LIFO it fires first and waits up to 10s on a WaitGroup for those services to drain before filer, volume, and master shut down. Resulting order on Ctrl+C: admin/s3/webdav/worker -> filer (gRPC -> HTTP -> DB) -> volume -> master. * refactor(mini): group mini-client shutdown into one state struct The first pass spread the shutdown plumbing across three globals (MiniClientsShutdownCtx, miniClientsWg, cancelMiniClients) and two ctx-derivation sites (OnMiniClientsShutdown and startMiniAdminWithWorker). Group into a private miniClientsState (ctx/cancel/wg) rebuilt per runMini invocation, and chain its ctx from MiniClusterCtx so clients only observe one signal. Tests that cancel MiniClusterCtx still trigger client shutdown via parent-child propagation. - resetMiniClients() installs fresh state at the top of runMini, so in-process test reruns don't inherit stale ctx/wg. - onMiniClientsShutdown(fn) replaces the exported OnMiniClientsShutdown and only observes one ctx. - trackMiniClient() replaces the manual wg.Add/Done dance for the admin goroutine. - miniClientsCtx() gives the admin startup a ctx without re-deriving. - triggerMiniClientsShutdown(timeout) is the interrupt hook body. No behaviour change; existing tests pass. * refactor: generalize shutdown ctx as an option, not a mini-specific helper Several service files (s3, webdav, filer, master, volume) observed the mini-specific MiniClusterCtx or called onMiniClientsShutdown directly. That leaked mini orchestration into code that also runs under weed s3, weed webdav, weed filer, weed master, and weed volume standalone. Replace with a generic `shutdownCtx context.Context` field on each service's Options struct. When non-nil, the server watches it and shuts down gracefully; when nil (standalone), the shutdown path is a no-op. Mini wires the contexts up from a single place (runMini): - miniMasterOptions/miniOptions.v/miniFilerOptions.shutdownCtx = MiniClusterCtx (drives test-triggered teardown) - miniS3Options/miniWebDavOptions.shutdownCtx = miniClientsCtx() (drives Ctrl+C teardown before filer/volume/master) All knowledge of MiniClusterCtx now lives in mini.go. * fix(mini): stop worker before clients ctx so admin shutdown isn't blocked Symptom on Ctrl+C of a clean weed mini: mini's Shutting down admin/s3/ webdav hook sat for 10s then logged "timed out". Admin had started its shutdown but was blocked inside StopWorkerGrpcServer's GracefulStop, waiting for the still-connected worker stream. That in turn left filer clients connected and cascaded into filer's own 10s gRPC graceful-stop timeout. Two causes, both fixed: 1. worker.Stop() deadlocked on clean shutdown. It sent ActionStop (which makes managerLoop `break out` and exit), then called getTaskLoad() which sends to the same unbuffered cmd channel — no receiver, hangs forever. Reorder Stop() to snapshot the admin client and drain tasks BEFORE sending ActionStop, and call Disconnect() via the local snapshot afterwards. 2. Worker's taskRequestLoop raced with Disconnect(): RequestTask reads from c.incoming, which Disconnect closes, yielding a nil response and a panic on response.Message. Handle the closed channel explicitly. 3. Mini now has a preCancel phase (beforeMiniClientsShutdown) that runs synchronously BEFORE the clients ctx is cancelled. Register worker shutdown there so admin's worker-gRPC GracefulStop finds the worker already disconnected and returns immediately, instead of waiting on a stream that is about to close anyway. Observed shutdown of a clean mini: admin/s3/webdav down in <10ms; full process exit in ~11s (the remaining 10s is a pre-existing filer gRPC graceful-stop timeout, not cascaded from the clients tier). * feat(mini): cap filer gRPC graceful stop at 1s under weed mini Full weed mini shutdown was ~11s on a clean exit, dominated by the filer's default 10s gRPC GracefulStop timeout while background SubscribeLocalMetadata streams drained. Expose the timeout as a FilerOptions.gracefulStopTimeout field (default 10s for standalone weed filer) and set it to 1s in mini. Clean weed mini shutdown now takes ~2s. --- weed/command/filer.go | 47 ++++--- weed/command/master.go | 9 +- weed/command/mini.go | 192 +++++++++++++++++++++++++---- weed/command/s3.go | 18 +-- weed/command/volume.go | 10 +- weed/command/webdav.go | 8 +- weed/util/grace/signal_handling.go | 5 +- weed/worker/client.go | 6 +- weed/worker/worker.go | 30 +++-- 9 files changed, 261 insertions(+), 64 deletions(-) diff --git a/weed/command/filer.go b/weed/command/filer.go index 0798417f3..08f4b3399 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -84,6 +84,13 @@ type FilerOptions struct { tusBasePath *string certProvider certprovider.Provider s3ConfigFile *string // optional path to static S3 identity config + // shutdownCtx, when non-nil, tells startFiler to gracefully shut down its + // HTTP/gRPC servers once the ctx is cancelled. Used by integration tests + // and by weed mini; nil for standalone weed filer. + shutdownCtx context.Context + // gracefulStopTimeout caps how long startFiler waits for gRPC graceful + // stop before forcing the server to stop. Zero means the default of 10s. + gracefulStopTimeout time.Duration } func init() { @@ -446,8 +453,12 @@ func (fo *FilerOptions) startFiler() { go grpcS.Serve(grpcL) pb.ServeGrpcOnLocalSocket(grpcS, grpcPort) - // Register graceful shutdown for gRPC server to wait for active RPCs - grace.OnInterrupt(func() { + // Helper to gracefully stop the gRPC server, waiting for active RPCs. + gracefulTimeout := fo.gracefulStopTimeout + if gracefulTimeout <= 0 { + gracefulTimeout = 10 * time.Second + } + stopGrpcServer := func() { glog.V(0).Infof("Gracefully stopping gRPC server") stopped := make(chan struct{}) go func() { @@ -457,11 +468,11 @@ func (fo *FilerOptions) startFiler() { select { case <-stopped: glog.V(0).Infof("gRPC server stopped gracefully") - case <-time.After(10 * time.Second): - glog.V(0).Infof("gRPC server graceful stop timed out, forcing stop") + case <-time.After(gracefulTimeout): + glog.V(0).Infof("gRPC server graceful stop timed out after %s, forcing stop", gracefulTimeout) grpcS.Stop() } - }) + } var socketServer *http.Server if runtime.GOOS != "windows" { @@ -528,8 +539,12 @@ func (fo *FilerOptions) startFiler() { } httpS := newHttpServer(defaultMux, tlsConfig) - // Register shutdown hooks: stop all HTTP servers, then close filer database + // Register a single shutdown hook that runs the steps in the correct order: + // stop accepting new gRPC/HTTP requests, then close the filer database. + // Combining them into one hook keeps ordering intact regardless of how + // grace fires interrupt hooks (FIFO vs LIFO). grace.OnInterrupt(func() { + stopGrpcServer() glog.V(0).Infof("Gracefully stopping all HTTP servers") shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() @@ -542,13 +557,12 @@ func (fo *FilerOptions) startFiler() { if err := httpS.Shutdown(shutdownCtx); err != nil { glog.Warningf("HTTPS server shutdown: %v", err) } + shutdownFiler() }) - grace.OnInterrupt(shutdownFiler) - if MiniClusterCtx != nil { - ctx := MiniClusterCtx + if fo.shutdownCtx != nil { go func() { - <-ctx.Done() + <-fo.shutdownCtx.Done() httpS.Shutdown(context.Background()) grpcS.Stop() }() @@ -570,8 +584,12 @@ func (fo *FilerOptions) startFiler() { } httpS := newHttpServer(defaultMux, nil) - // Register shutdown hooks: stop all HTTP servers, then close filer database + // Register a single shutdown hook that runs the steps in the correct order: + // stop accepting new gRPC/HTTP requests, then close the filer database. + // Combining them into one hook keeps ordering intact regardless of how + // grace fires interrupt hooks (FIFO vs LIFO). grace.OnInterrupt(func() { + stopGrpcServer() glog.V(0).Infof("Gracefully stopping all HTTP servers") shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() @@ -584,13 +602,12 @@ func (fo *FilerOptions) startFiler() { if err := httpS.Shutdown(shutdownCtx); err != nil { glog.Warningf("HTTP server shutdown: %v", err) } + shutdownFiler() }) - grace.OnInterrupt(shutdownFiler) - if MiniClusterCtx != nil { - ctx := MiniClusterCtx + if fo.shutdownCtx != nil { go func() { - <-ctx.Done() + <-fo.shutdownCtx.Done() httpS.Shutdown(context.Background()) grpcS.Stop() }() diff --git a/weed/command/master.go b/weed/command/master.go index 189a38a35..940aea680 100644 --- a/weed/command/master.go +++ b/weed/command/master.go @@ -74,6 +74,10 @@ type MasterOptions struct { telemetryEnabled *bool debug *bool debugPort *int + // shutdownCtx, when non-nil, tells startMaster to shut down once the ctx + // is cancelled. Used by integration tests and by weed mini; nil for + // standalone weed master. + shutdownCtx context.Context } func init() { @@ -336,9 +340,8 @@ func startMaster(masterOption MasterOptions, masterWhiteList []string) { ms.Topo.HashicorpRaft.LeadershipTransfer() } }) - ctx := MiniClusterCtx - if ctx != nil { - <-ctx.Done() + if masterOption.shutdownCtx != nil { + <-masterOption.shutdownCtx.Done() ms.Shutdown() grpcS.Stop() } else { diff --git a/weed/command/mini.go b/weed/command/mini.go index 56d8e2934..9b25ae33e 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -9,6 +9,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -66,6 +67,123 @@ var ( MiniClusterCtx context.Context ) +// miniClientsState orchestrates graceful shutdown of admin/s3/webdav/worker on +// weed mini Ctrl+C, BEFORE filer/volume/master tear down. It is rebuilt on +// each runMini invocation so in-process test reruns see fresh state. +// +// The ctx chains from MiniClusterCtx so cancelling MiniClusterCtx (how tests +// tear down the cluster) also triggers the client-shutdown path. +// +// Shutdown has two phases: +// 1. preCancelFns run synchronously in registration order (worker disconnect, +// etc.) so downstream servers don't block on handlers that are about to +// close anyway. +// 2. ctx is cancelled and the shutdown hook waits on wg for admin/s3/webdav +// goroutines (registered via onMiniClientsShutdown / trackMiniClient) to +// drain. +type miniClientsState struct { + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + preCancelMu sync.Mutex + preCancelFns []func() +} + +var miniClients *miniClientsState + +// resetMiniClients installs a fresh client-shutdown state chained from +// MiniClusterCtx. Called once at the top of runMini; any goroutines from a +// prior invocation keep their old state via closure and are unaffected. +func resetMiniClients() { + parent := context.Background() + if MiniClusterCtx != nil { + parent = MiniClusterCtx + } + s := &miniClientsState{} + s.ctx, s.cancel = context.WithCancel(parent) + miniClients = s +} + +// onMiniClientsShutdown runs fn when mini shutdown is triggered, and tracks +// it so the interrupt hook can wait for it to drain. No-op outside mini. +func onMiniClientsShutdown(fn func()) { + s := miniClients + if s == nil { + return + } + s.wg.Add(1) + go func() { + defer s.wg.Done() + <-s.ctx.Done() + fn() + }() +} + +// trackMiniClient registers an externally-managed goroutine (one that +// observes miniClientsCtx() itself) so the interrupt hook waits for it. +// The caller invokes the returned done func when the goroutine exits. +func trackMiniClient() (done func()) { + s := miniClients + if s == nil { + return func() {} + } + s.wg.Add(1) + return s.wg.Done +} + +// beforeMiniClientsShutdown registers fn to run synchronously BEFORE the +// clients ctx is cancelled. Use for cleanup that must complete before +// downstream servers (e.g., the admin worker-gRPC) start waiting on clients. +func beforeMiniClientsShutdown(fn func()) { + s := miniClients + if s == nil { + return + } + s.preCancelMu.Lock() + defer s.preCancelMu.Unlock() + s.preCancelFns = append(s.preCancelFns, fn) +} + +// miniClientsCtx returns the shutdown context for mini clients, or +// context.Background() if not running inside weed mini. +func miniClientsCtx() context.Context { + s := miniClients + if s == nil { + return context.Background() + } + return s.ctx +} + +// triggerMiniClientsShutdown runs preCancel fns synchronously, cancels the +// clients ctx, and waits up to timeout for tracked goroutines to finish. +// Called from the OnInterrupt hook. +func triggerMiniClientsShutdown(timeout time.Duration) { + s := miniClients + if s == nil { + return + } + glog.V(0).Infof("Shutting down admin/s3/webdav ...") + s.preCancelMu.Lock() + fns := s.preCancelFns + s.preCancelFns = nil + s.preCancelMu.Unlock() + for _, fn := range fns { + fn() + } + s.cancel() + done := make(chan struct{}) + go func() { + s.wg.Wait() + close(done) + }() + select { + case <-done: + glog.V(0).Infof("admin/s3/webdav shut down") + case <-time.After(timeout): + glog.V(0).Infof("timed out waiting for admin/s3/webdav to shut down") + } +} + func init() { cmdMini.Run = runMini // break init cycle } @@ -869,6 +987,21 @@ func runMini(cmd *Command, args []string) bool { miniWhiteList := util.StringSplit(*miniWhiteListOption, ",") + // Install a fresh clients-shutdown context (chained from MiniClusterCtx) + // before any service starts. + resetMiniClients() + + // Master/volume/filer observe MiniClusterCtx so tests that cancel it + // tear those services down too. On Ctrl+C they rely on their own + // OnInterrupt hooks (see grace's LIFO ordering). + miniMasterOptions.shutdownCtx = MiniClusterCtx + miniOptions.v.shutdownCtx = MiniClusterCtx + miniFilerOptions.shutdownCtx = MiniClusterCtx + // Mini is a small/dev setup with short-lived RPCs; cap the filer's + // gRPC graceful-stop at 1s so Ctrl+C returns quickly instead of sitting + // on the default 10s waiting for background subscription streams. + miniFilerOptions.gracefulStopTimeout = 1 * time.Second + // Start all services with proper dependency coordination // This channel will be closed when all services are fully ready allServicesReady := make(chan struct{}) @@ -877,6 +1010,13 @@ func runMini(cmd *Command, args []string) bool { // Wait for all services to be fully running before printing welcome message <-allServicesReady + // Register the clients-shutdown interrupt hook AFTER all services have + // registered theirs. Under grace's LIFO firing, this hook runs FIRST on + // Ctrl+C so admin/s3/webdav drain before filer/volume/master tear down. + grace.OnInterrupt(func() { + triggerMiniClientsShutdown(10 * time.Second) + }) + // Print welcome message after all services are running printWelcomeMessage() @@ -921,17 +1061,27 @@ func startMiniServices(miniWhiteList []string, allServicesReady chan struct{}) { // Wait for filer to be ready waitForServiceReady("Filer", *miniFilerOptions.port, bindIp) - // Start S3 and WebDAV in parallel (both depend on filer) + // Start S3 and WebDAV in parallel (both depend on filer). Each observes + // miniClientsCtx so it shuts down first on Ctrl+C, tracked via + // trackMiniClient so runMini's interrupt hook can wait for them. if *miniEnableS3 { - go startMiniService("S3", func() { - startS3Service() - }, *miniS3Options.port) + miniS3Options.shutdownCtx = miniClientsCtx() + done := trackMiniClient() + go func() { + defer done() + startMiniService("S3", startS3Service, *miniS3Options.port) + }() } if *miniEnableWebDAV { - go startMiniService("WebDAV", func() { - miniWebDavOptions.startWebDav() - }, *miniWebDavOptions.port) + miniWebDavOptions.shutdownCtx = miniClientsCtx() + done := trackMiniClient() + go func() { + defer done() + startMiniService("WebDAV", func() { + miniWebDavOptions.startWebDav() + }, *miniWebDavOptions.port) + }() } // Wait for services to be ready @@ -999,12 +1149,8 @@ func startS3Service() { func startMiniAdminWithWorker(allServicesReady chan struct{}) { defer close(allServicesReady) // Ensure channel is always closed on all paths - var ctx context.Context - if MiniClusterCtx != nil { - ctx = MiniClusterCtx - } else { - ctx = context.Background() - } + // Admin shuts down when mini clients shutdown is triggered. + ctx := miniClientsCtx() // Determine bind IP for health checks bindIp := getBindIp() @@ -1044,8 +1190,12 @@ func startMiniAdminWithWorker(allServicesReady chan struct{}) { *miniAdminOptions.dataDir = filepath.Join(*miniDataFolders, "admin") } - // Start admin server in background + // Start admin server in background. trackMiniClient lets the Ctrl+C + // handler wait for startAdminServer's graceful shutdown before filer/ + // volume/master tear down. + done := trackMiniClient() go func() { + defer done() var icebergPort int if miniS3Options.portIceberg != nil { icebergPort = *miniS3Options.portIceberg @@ -1189,13 +1339,13 @@ func startMiniWorker(workerDir string) { // Metrics server is already started in the main init function above, so no need to start it again here - // Start the worker - if MiniClusterCtx != nil { - go func() { - <-MiniClusterCtx.Done() - workerInstance.Stop() - }() - } + // Stop the worker BEFORE the clients ctx is cancelled. Otherwise admin's + // internal worker gRPC GracefulStop (called during admin.Shutdown) would + // wait for the worker stream to close, blocking the whole mini shutdown + // by ~10s and cascading into filer's own gRPC graceful stop timeout. + beforeMiniClientsShutdown(func() { + workerInstance.Stop() + }) err = workerInstance.Start() if err != nil { glog.Fatalf("Failed to start worker: %v", err) diff --git a/weed/command/s3.go b/weed/command/s3.go index 962bb2d05..dd97eab46 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -74,6 +74,11 @@ type S3Options struct { externalUrl *string defaultFileMode *string cacheSizeMB *int64 + // shutdownCtx, when non-nil, tells startS3Server/startIcebergServer to + // gracefully shut down their HTTP/gRPC servers once the ctx is cancelled. + // Used by weed mini to orchestrate an ordered shutdown; nil for standalone + // weed s3. + shutdownCtx context.Context } func init() { @@ -453,10 +458,9 @@ func (s3opt *S3Options) startS3Server() bool { }() } httpS := newHttpServer(router, tlsConfig) - if MiniClusterCtx != nil { - ctx := MiniClusterCtx + if s3opt.shutdownCtx != nil { go func() { - <-ctx.Done() + <-s3opt.shutdownCtx.Done() httpS.Shutdown(context.Background()) grpcS.Stop() }() @@ -495,9 +499,9 @@ func (s3opt *S3Options) startS3Server() bool { }() } httpS := newHttpServer(router, nil) - if MiniClusterCtx != nil { + if s3opt.shutdownCtx != nil { go func() { - <-MiniClusterCtx.Done() + <-s3opt.shutdownCtx.Done() httpS.Shutdown(context.Background()) grpcS.Stop() }() @@ -531,9 +535,9 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer) { glog.V(0).Infof("Start Iceberg REST Catalog Server at http://%s", listenAddress) httpS := newHttpServer(icebergRouter, nil) - if MiniClusterCtx != nil { + if s3opt.shutdownCtx != nil { go func() { - <-MiniClusterCtx.Done() + <-s3opt.shutdownCtx.Done() httpS.Shutdown(context.Background()) }() } diff --git a/weed/command/volume.go b/weed/command/volume.go index aa456a0ad..f5366964a 100644 --- a/weed/command/volume.go +++ b/weed/command/volume.go @@ -1,6 +1,7 @@ package command import ( + "context" "fmt" "net/http" httppprof "net/http/pprof" @@ -75,6 +76,10 @@ type VolumeServerOptions struct { ldbTimeout *int64 debug *bool debugPort *int + // shutdownCtx, when non-nil, tells startVolumeServer to shut down once the + // ctx is cancelled. Used by integration tests and by weed mini; nil for + // standalone weed volume. + shutdownCtx context.Context } func init() { @@ -327,11 +332,10 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v stopChan <- true }) - ctx := MiniClusterCtx - if ctx != nil { + if v.shutdownCtx != nil { select { case <-stopChan: - case <-ctx.Done(): + case <-v.shutdownCtx.Done(): shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer) } } else { diff --git a/weed/command/webdav.go b/weed/command/webdav.go index cf6b0bee6..6332832d9 100644 --- a/weed/command/webdav.go +++ b/weed/command/webdav.go @@ -36,6 +36,10 @@ type WebDavOption struct { cacheDir *string cacheSizeMB *int64 maxMB *int + // shutdownCtx, when non-nil, tells startWebDav to gracefully shut down the + // HTTP server once the ctx is cancelled. Used by weed mini; nil for + // standalone weed webdav. + shutdownCtx context.Context } func init() { @@ -137,9 +141,9 @@ func (wo *WebDavOption) startWebDav() bool { glog.Fatalf("WebDav Server listener on %s error: %v", listenAddress, err) } - if MiniClusterCtx != nil { + if wo.shutdownCtx != nil { go func() { - <-MiniClusterCtx.Done() + <-wo.shutdownCtx.Done() httpS.Shutdown(context.Background()) }() } diff --git a/weed/util/grace/signal_handling.go b/weed/util/grace/signal_handling.go index bb2b1f3db..75069fd5f 100644 --- a/weed/util/grace/signal_handling.go +++ b/weed/util/grace/signal_handling.go @@ -44,7 +44,10 @@ func init() { reloadHookLock.RUnlock() } else { interruptHookLock.RLock() - for _, hook := range interruptHooks { + // Execute hooks in reverse registration order (LIFO/defer-style) so + // later-started services shut down before the services they depend on. + for i := len(interruptHooks) - 1; i >= 0; i-- { + hook := interruptHooks[i] glog.V(4).Infof("exec interrupt hook func name:%s", GetFunctionName(hook)) hook() } diff --git a/weed/worker/client.go b/weed/worker/client.go index 1b08b053d..20fcd50aa 100644 --- a/weed/worker/client.go +++ b/weed/worker/client.go @@ -838,7 +838,11 @@ func (c *GrpcAdminClient) RequestTask(workerID string, capabilities []types.Task for { select { - case response := <-c.incoming: + case response, ok := <-c.incoming: + if !ok || response == nil { + // incoming was closed (e.g. during Disconnect). + return nil, fmt.Errorf("incoming channel closed") + } glog.V(4).Infof("RESPONSE RECEIVED: Worker %s received response from admin server: %T", workerID, response.Message) if taskAssign := response.GetTaskAssignment(); taskAssign != nil { // Validate TaskId is not empty before processing diff --git a/weed/worker/worker.go b/weed/worker/worker.go index ffcd00b9e..e361b30e0 100644 --- a/weed/worker/worker.go +++ b/weed/worker/worker.go @@ -465,16 +465,14 @@ func (w *Worker) handleStart(cmd workerCommand) { } func (w *Worker) Stop() error { - resp := make(chan error) - w.cmds <- workerCommand{ - action: ActionStop, - resp: resp, - } - if err := <-resp; err != nil { - return err - } + // Snapshot the admin client BEFORE ActionStop terminates the manager loop. + // Once the loop exits, any further w.cmds send (getTaskLoad, getAdmin, etc.) + // would deadlock because no one reads w.cmds anymore. + adminClient := w.getAdmin() - // Wait for tasks to finish + // Best-effort task drain: wait up to 30s for in-flight tasks to finish. + // Still done BEFORE ActionStop so getTaskLoad can round-trip through the + // manager loop. timeout := time.NewTimer(30 * time.Second) defer timeout.Stop() out: @@ -487,8 +485,18 @@ out: } } - // Disconnect from admin server - if adminClient := w.getAdmin(); adminClient != nil { + // Terminate the manager loop. + resp := make(chan error) + w.cmds <- workerCommand{ + action: ActionStop, + resp: resp, + } + if err := <-resp; err != nil { + return err + } + + // Disconnect from admin server. + if adminClient != nil { if err := adminClient.Disconnect(); err != nil { glog.Errorf("Error disconnecting from admin server: %v", err) }