feature(worker): when SIGINT/SIGTERM received, stop all the jobs and update their status before quit. Close #19.

worker/worker.go

@@ -20,6 +20,7 @@ type Worker struct {
 
 	managerChan chan jobMessage
 	semaphore   chan empty
+	exit        chan empty
 
 	schedule   *scheduleQueue
 	httpEngine *gin.Engine
@@ -38,6 +39,7 @@ func GetTUNASyncWorker(cfg *Config) *Worker {
 
 		managerChan: make(chan jobMessage, 32),
 		semaphore:   make(chan empty, cfg.Global.Concurrent),
+		exit:        make(chan empty),
 
 		schedule: newScheduleQueue(),
 	}
@@ -222,6 +224,21 @@ func (w *Worker) runHTTPServer() {
 	}
 }
 
+// Halt stops all jobs
+func (w *Worker) Halt() {
+	w.L.Lock()
+	logger.Notice("Stopping all the jobs")
+	for _, job := range w.jobs {
+		if job.State() != stateDisabled {
+			job.ctrlChan <- jobHalt
+		}
+	}
+	jobsDone.Wait()
+	logger.Notice("All the jobs are stopped")
+	w.L.Unlock()
+	close(w.exit)
+}
+
 // Run runs worker forever
 func (w *Worker) Run() {
 	w.registorWorker()
@@ -284,7 +301,7 @@ func (w *Worker) runSchedule() {
 				continue
 			}
 
-			if job.State() != stateReady {
+			if (job.State() != stateReady) && (job.State() != stateHalting) {
 				logger.Infof("Job %s state is not ready, skip adding new schedule", jobMsg.name)
 				continue
 			}
@@ -312,6 +329,25 @@ func (w *Worker) runSchedule() {
 			if job := w.schedule.Pop(); job != nil {
 				job.ctrlChan <- jobStart
 			}
+		case <-w.exit:
+			// flush status update messages
+			w.L.Lock()
+			defer w.L.Unlock()
+			for {
+				select {
+				case jobMsg := <-w.managerChan:
+					logger.Debugf("status update from %s", jobMsg.name)
+					job, ok := w.jobs[jobMsg.name]
+					if !ok {
+						continue
+					}
+					if jobMsg.status == Failed || jobMsg.status == Success {
+						w.updateStatus(job, jobMsg)
+					}
+				default:
+					return
+				}
+			}
 		}
 	}
 }