From 29b78f84d7fd25765e8e0e8e4886165507436ed6 Mon Sep 17 00:00:00 2001 From: Agniva De Sarker Date: Mon, 21 Mar 2022 09:04:06 +0530 Subject: [PATCH] MM-40272: Fix job server graceful stop (#19760) This fixes a classic concurrency bug where the stop channel was being listened in 2 places. In the Run method and DoJob method. The problem with that is that one of them will consume the signal, and if the DoJob method does, then the parent for-select loop never exits, because the return statement only runs when the stop signal is consumed. This would lead to server hanging on shutdown. To fix this, we close the channel rather than sending a message. This exits all listeners listening on the channel and lets the job exit gracefully. This is a sample stack dump of a stuck goroutine ``` goroutine 1 [chan receive]: runtime.gopark(0x235be40?, 0xc000faa530?, 0xa8?, 0xf3?, 0x27a1170?) /home/agniva/gosource/go/src/runtime/proc.go:366 +0xd6 fp=0xc002e77510 sp=0xc002e774f0 pc=0x43c856 runtime.chanrecv(0xc000faa7e0, 0x0, 0x1) /home/agniva/gosource/go/src/runtime/chan.go:577 +0x56c fp=0xc002e775a0 sp=0xc002e77510 pc=0x40868c runtime.chanrecv1(0x27cab4c?, 0xf?) /home/agniva/gosource/go/src/runtime/chan.go:440 +0x18 fp=0xc002e775c8 sp=0xc002e775a0 pc=0x4080b8 github.com/mattermost/mattermost-server/v6/jobs/migrations.(*Worker).Stop(0xc001954280) /home/agniva/mattermost/mattermost-server/jobs/migrations/worker.go:66 +0x171 fp=0xc002e77698 sp=0xc002e775c8 pc=0x187f8d1 github.com/mattermost/mattermost-server/v6/jobs.(*Workers).Stop(0xc0019df2c0) /home/agniva/mattermost/mattermost-server/jobs/workers.go:84 +0x12c fp=0xc002e77748 sp=0xc002e77698 pc=0x17b04ec github.com/mattermost/mattermost-server/v6/jobs.(*JobServer).StopWorkers(0xc00021f180?) /home/agniva/mattermost/mattermost-server/jobs/server.go:104 +0x9c fp=0xc002e777a0 sp=0xc002e77748 pc=0x17af8bc github.com/mattermost/mattermost-server/v6/app.(*Server).Shutdown(0xc00021f180) /home/agniva/mattermost/mattermost-server/app/server.go:1019 +0x965 fp=0xc002e77c48 sp=0xc002e777a0 pc=0x1dd6865 github.com/mattermost/mattermost-server/v6/cmd/mattermost/commands.runServer.func3() /home/agniva/mattermost/mattermost-server/cmd/mattermost/commands/server.go:79 +0x26 fp=0xc002e77c60 sp=0xc002e77c48 pc=0x214bbe6 github.com/mattermost/mattermost-server/v6/cmd/mattermost/commands.runServer(0xc000e1e000, 0xb?) /home/agniva/mattermost/mattermost-server/cmd/mattermost/commands/server.go:122 +0x239 fp=0xc002e77d50 sp=0xc002e77c60 pc=0x214ba39 github.com/mattermost/mattermost-server/v6/cmd/mattermost/commands.serverCmdF(0x4207ac0?, {0x279d9d6?, 0x0?, 0x0?}) /home/agniva/mattermost/mattermost-server/cmd/mattermost/commands/server.go:58 +0x11d fp=0xc002e77dd0 sp=0xc002e77d50 pc=0x214b37d github.com/spf13/cobra.(*Command).execute(0x4207ac0, {0xc00003c200, 0x0, 0x0}) /home/agniva/mattermost/mattermost-server/vendor/github.com/spf13/cobra/command.go:856 +0x67c fp=0xc002e77ea8 sp=0xc002e77dd0 pc=0x211a21c github.com/spf13/cobra.(*Command).ExecuteC(0x4207ac0) /home/agniva/mattermost/mattermost-server/vendor/github.com/spf13/cobra/command.go:974 +0x3b4 fp=0xc002e77f60 sp=0xc002e77ea8 pc=0x211a894 github.com/spf13/cobra.(*Command).Execute(...) /home/agniva/mattermost/mattermost-server/vendor/github.com/spf13/cobra/command.go:902 github.com/mattermost/mattermost-server/v6/cmd/mattermost/commands.Run(...) /home/agniva/mattermost/mattermost-server/cmd/mattermost/commands/root.go:14 main.main() /home/agniva/mattermost/mattermost-server/cmd/mattermost/main.go:31 +0x86 fp=0xc002e77f80 sp=0xc002e77f60 pc=0x21f9ee6 runtime.main() /home/agniva/gosource/go/src/runtime/proc.go:255 +0x227 fp=0xc002e77fe0 sp=0xc002e77f80 pc=0x43c487 runtime.goexit() /home/agniva/gosource/go/src/runtime/asm_amd64.s:1571 +0x1 fp=0xc002e77fe8 sp=0xc002e77fe0 pc=0x46d361 ``` This is not strictly related to the JIRA ticket but a general refactor https://mattermost.atlassian.net/browse/MM-40272 ```release-note NONE ``` --- jobs/jobs.go | 15 +++++++++------ jobs/migrations/worker.go | 18 ++++++++++++++---- .../bleveengine/indexer/indexing_job.go | 18 ++++++++++++++---- 3 files changed, 37 insertions(+), 14 deletions(-) diff --git a/jobs/jobs.go b/jobs/jobs.go index 5e739b3d9d..897fe87c25 100644 --- a/jobs/jobs.go +++ b/jobs/jobs.go @@ -194,7 +194,7 @@ func (srv *JobServer) RequestCancellation(jobId string) *model.AppError { return model.NewAppError("RequestCancellation", "jobs.request_cancellation.status.error", nil, "id="+jobId, http.StatusInternalServerError) } -func (srv *JobServer) CancellationWatcher(ctx context.Context, jobId string, cancelChan chan interface{}) { +func (srv *JobServer) CancellationWatcher(ctx context.Context, jobId string, cancelChan chan struct{}) { for { select { case <-ctx.Done(): @@ -202,11 +202,14 @@ func (srv *JobServer) CancellationWatcher(ctx context.Context, jobId string, can return case <-time.After(CancelWatcherPollingInterval * time.Millisecond): mlog.Debug("CancellationWatcher for Job started polling.", mlog.String("job_id", jobId)) - if jobStatus, err := srv.Store.Job().Get(jobId); err == nil { - if jobStatus.Status == model.JobStatusCancelRequested { - close(cancelChan) - return - } + jobStatus, err := srv.Store.Job().Get(jobId) + if err != nil { + mlog.Warn("Error getting job", mlog.String("job_id", jobId), mlog.Err(err)) + continue + } + if jobStatus.Status == model.JobStatusCancelRequested { + close(cancelChan) + return } } } diff --git a/jobs/migrations/worker.go b/jobs/migrations/worker.go index 1872b68811..001c04a493 100644 --- a/jobs/migrations/worker.go +++ b/jobs/migrations/worker.go @@ -6,6 +6,7 @@ package migrations import ( "context" "net/http" + "sync/atomic" "time" "github.com/mattermost/mattermost-server/v6/jobs" @@ -20,17 +21,18 @@ const ( type Worker struct { name string - stop chan bool + stop chan struct{} stopped chan bool jobs chan model.Job jobServer *jobs.JobServer store store.Store + closed int32 } func MakeWorker(jobServer *jobs.JobServer, store store.Store) model.Worker { worker := Worker{ name: "Migrations", - stop: make(chan bool, 1), + stop: make(chan struct{}), stopped: make(chan bool, 1), jobs: make(chan model.Job), jobServer: jobServer, @@ -41,6 +43,10 @@ func MakeWorker(jobServer *jobs.JobServer, store store.Store) model.Worker { } func (worker *Worker) Run() { + // Set to open if closed before. We are not bothered about multiple opens. + if atomic.CompareAndSwapInt32(&worker.closed, 1, 0) { + worker.stop = make(chan struct{}) + } mlog.Debug("Worker started", mlog.String("worker", worker.name)) defer func() { @@ -61,8 +67,12 @@ func (worker *Worker) Run() { } func (worker *Worker) Stop() { + // Set to close, and if already closed before, then return. + if !atomic.CompareAndSwapInt32(&worker.closed, 0, 1) { + return + } mlog.Debug("Worker stopping", mlog.String("worker", worker.name)) - worker.stop <- true + close(worker.stop) <-worker.stopped } @@ -86,7 +96,7 @@ func (worker *Worker) DoJob(job *model.Job) { } cancelCtx, cancelCancelWatcher := context.WithCancel(context.Background()) - cancelWatcherChan := make(chan interface{}, 1) + cancelWatcherChan := make(chan struct{}, 1) go worker.jobServer.CancellationWatcher(cancelCtx, job.Id, cancelWatcherChan) defer cancelCancelWatcher() diff --git a/services/searchengine/bleveengine/indexer/indexing_job.go b/services/searchengine/bleveengine/indexer/indexing_job.go index 510be5ed57..db52e7eaeb 100644 --- a/services/searchengine/bleveengine/indexer/indexing_job.go +++ b/services/searchengine/bleveengine/indexer/indexing_job.go @@ -7,6 +7,7 @@ import ( "context" "net/http" "strconv" + "sync/atomic" "time" "github.com/mattermost/mattermost-server/v6/jobs" @@ -26,11 +27,12 @@ const ( type BleveIndexerWorker struct { name string - stop chan bool + stop chan struct{} stopped chan bool jobs chan model.Job jobServer *jobs.JobServer engine *bleveengine.BleveEngine + closed int32 } func MakeWorker(jobServer *jobs.JobServer, engine *bleveengine.BleveEngine) model.Worker { @@ -39,7 +41,7 @@ func MakeWorker(jobServer *jobs.JobServer, engine *bleveengine.BleveEngine) mode } return &BleveIndexerWorker{ name: "BleveIndexer", - stop: make(chan bool, 1), + stop: make(chan struct{}), stopped: make(chan bool, 1), jobs: make(chan model.Job), jobServer: jobServer, @@ -83,6 +85,10 @@ func (worker *BleveIndexerWorker) IsEnabled(cfg *model.Config) bool { } func (worker *BleveIndexerWorker) Run() { + // Set to open if closed before. We are not bothered about multiple opens. + if atomic.CompareAndSwapInt32(&worker.closed, 1, 0) { + worker.stop = make(chan struct{}) + } mlog.Debug("Worker Started", mlog.String("workername", worker.name)) defer func() { @@ -103,8 +109,12 @@ func (worker *BleveIndexerWorker) Run() { } func (worker *BleveIndexerWorker) Stop() { + // Set to close, and if already closed before, then return. + if !atomic.CompareAndSwapInt32(&worker.closed, 0, 1) { + return + } mlog.Debug("Worker Stopping", mlog.String("workername", worker.name)) - worker.stop <- true + close(worker.stop) <-worker.stopped } @@ -215,7 +225,7 @@ func (worker *BleveIndexerWorker) DoJob(job *model.Job) { } cancelCtx, cancelCancelWatcher := context.WithCancel(context.Background()) - cancelWatcherChan := make(chan interface{}, 1) + cancelWatcherChan := make(chan struct{}, 1) go worker.jobServer.CancellationWatcher(cancelCtx, job.Id, cancelWatcherChan) defer cancelCancelWatcher()