syz-ci: explicitly stop all running jobs on update

Otherwise we might end up in a situation when we have stopped all
fuzzing, but wait for the job processor to report Done to the waitgroup
object.

2 files changed, 27 insertions, 19 deletions

diff --git a/syz-ci/jobs.go b/syz-ci/jobs.go
index fcdca9ada..04c241ab2 100644
--- a/syz-ci/jobs.go
+++ b/syz-ci/jobs.go
@@ -58,6 +58,23 @@ func newJobManager(cfg *Config, managers []*Manager, shutdownPending chan struct
 	}, nil
 }
 
+// startLoop starts a job loop in parallel and returns a blocking function
+// to gracefully stop job processing.
+func (jm *JobManager) startLoop(wg *sync.WaitGroup) func() {
+	stop := make(chan struct{})
+	done := make(chan struct{}, 1)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		jm.loop(stop)
+		done <- struct{}{}
+	}()
+	return func() {
+		stop <- struct{}{}
+		<-done
+	}
+}
+
 func (jm *JobManager) loop(stop chan struct{}) {
 	if err := jm.resetJobs(); err != nil {
 		if jm.dash != nil {
diff --git a/syz-ci/syz-ci.go b/syz-ci/syz-ci.go
index 41564538c..e9b51d267 100644
--- a/syz-ci/syz-ci.go
+++ b/syz-ci/syz-ci.go
@@ -227,20 +227,7 @@ func main() {
 		}()
 	}
 
-	var wg sync.WaitGroup
-	wg.Add(1)
 	stop := make(chan struct{})
-	go func() {
-		select {
-		case <-shutdownPending:
-		case <-updatePending:
-		}
-		buildSem.WaitAll() // wait for all current builds
-		testSem.WaitAll()
-		close(stop)
-		wg.Done()
-	}()
-
 	var managers []*Manager
 	for _, mgrcfg := range cfg.Managers {
 		mgr, err := createManager(cfg, mgrcfg, stop, *flagDebug)
@@ -253,6 +240,7 @@ func main() {
 	if len(managers) == 0 {
 		log.Fatalf("failed to create all managers")
 	}
+	var wg sync.WaitGroup
 	if *flagManagers {
 		for _, mgr := range managers {
 			mgr := mgr
@@ -267,11 +255,7 @@ func main() {
 	if err != nil {
 		log.Fatalf("failed to create dashapi connection %v", err)
 	}
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		jp.loop(stop)
-	}()
+	stopJobs := jp.startLoop(&wg)
 
 	// For testing. Racy. Use with care.
 	http.HandleFunc("/upload_cover", func(w http.ResponseWriter, r *http.Request) {
@@ -286,11 +270,18 @@ func main() {
 
 	wg.Add(1)
 	go deprecateAssets(cfg, stop, &wg)
-	wg.Wait()
 
 	select {
 	case <-shutdownPending:
 	case <-updatePending:
+	}
+	stopJobs() // Gracefully wait for the running jobs to finish.
+	close(stop)
+	wg.Wait()
+
+	select {
+	case <-shutdownPending:
+	default:
 		updater.UpdateAndRestart()
 	}
 }