[task scheduler] Start jobs in parallel

This should speed things up substantially. We still process jobs which
have the same RepoState (same issue+patchset+commit) sequentially, since
the cache will prevent parallelizing those anyway.

Bug: b/339625138
Change-Id: Ida88a4f0a96165656c9ed8fd817adba382e2289f
Reviewed-on: https://skia-review.googlesource.com/c/buildbot/+/875080
Auto-Submit: Eric Boren <borenet@google.com>
Reviewed-by: Ravi Mistry <rmistry@google.com>
Commit-Queue: Ravi Mistry <rmistry@google.com>

task_scheduler/go/tryjobs/BUILD.bazel

diff --git a/task_scheduler/go/tryjobs/BUILD.bazel b/task_scheduler/go/tryjobs/BUILD.bazel
index 70a5432..4bdae9d 100644
--- a/task_scheduler/go/tryjobs/BUILD.bazel
+++ b/task_scheduler/go/tryjobs/BUILD.bazel
@@ -11,7 +11,6 @@
         "//go/cleanup",
         "//go/gerrit",
         "//go/git/repograph",
-        "//go/human",
         "//go/metrics2",
         "//go/now",
         "//go/pubsub",

task_scheduler/go/tryjobs/tryjobs.go

diff --git a/task_scheduler/go/tryjobs/tryjobs.go b/task_scheduler/go/tryjobs/tryjobs.go
index 7d280fe..26f5bfd 100644
--- a/task_scheduler/go/tryjobs/tryjobs.go
+++ b/task_scheduler/go/tryjobs/tryjobs.go
@@ -18,7 +18,6 @@
 	"go.skia.org/infra/go/cleanup"
 	"go.skia.org/infra/go/gerrit"
 	"go.skia.org/infra/go/git/repograph"
-	"go.skia.org/infra/go/human"
 	"go.skia.org/infra/go/metrics2"
 	"go.skia.org/infra/go/now"
 	"go.skia.org/infra/go/pubsub"
@@ -361,35 +360,14 @@
 	for {
 		select {
 		case jobs := <-jobsCh:
-			sklog.Infof("Start processing jobs from modified jobs channel.")
-			for _, job := range jobs {
-				sklog.Infof("Found job %s (build %d) via modified jobs channel", job.Id, job.BuildbucketBuildId)
-			}
-			for _, job := range jobs {
-				if job.Status != types.JOB_STATUS_REQUESTED {
-					continue
-				}
-				if err := t.startJob(ctx, job); err != nil {
-					sklog.Errorf("failed to start job %s (build %d): %s", job.Id, job.BuildbucketBuildId, err)
-				}
-			}
-			sklog.Infof("Done processing jobs from modified jobs channel.")
+			t.startJobs(ctx, jobs, "modified jobs channel")
 		case <-tickCh:
-			sklog.Infof("Start processing jobs from periodic DB poll, cache updated %s ago.", human.Duration(time.Now().Sub(t.jCache.LastUpdated())))
 			jobs, err := t.jCache.RequestedJobs()
 			if err != nil {
 				sklog.Errorf("failed retrieving Jobs: %s", err)
 			} else {
-				for _, job := range jobs {
-					sklog.Infof("Found job %s (build %d) via periodic DB poll", job.Id, job.BuildbucketBuildId)
-				}
-				for _, job := range jobs {
-					if err := t.startJob(ctx, job); err != nil {
-						sklog.Errorf("failed to start job %s (build %d): %s", job.Id, job.BuildbucketBuildId, err)
-					}
-				}
+				t.startJobs(ctx, jobs, "periodic DB poll")
 			}
-			sklog.Infof("Done processing jobs from periodic DB poll.")
 		case <-doneCh:
 			ticker.Stop()
 			return
@@ -397,6 +375,32 @@
 	}
 }
 
+func (t *TryJobIntegrator) startJobs(ctx context.Context, jobs []*types.Job, foundVia string) {
+	sklog.Infof("Start processing jobs from %s.", foundVia)
+	// Organize jobs by RepoState, since jobs with the same RepoState must be
+	// processed serially.
+	byRepoState := map[types.RepoState][]*types.Job{}
+	for _, job := range jobs {
+		sklog.Infof("Found job %s (build %d) via %s", job.Id, job.BuildbucketBuildId, foundVia)
+		byRepoState[job.RepoState] = append(byRepoState[job.RepoState], job)
+	}
+	var wg sync.WaitGroup
+	for _, jobs := range byRepoState {
+		wg.Add(1)
+		jobs := jobs // https://golang.org/doc/faq#closures_and_goroutines
+		go func() {
+			defer wg.Done()
+			for _, job := range jobs {
+				if err := t.startJob(ctx, job); err != nil {
+					sklog.Errorf("failed to start job %s (build %d): %s", job.Id, job.BuildbucketBuildId, err)
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	sklog.Infof("Done processing jobs from %s.", foundVia)
+}
+
 func isBuildAlreadyStartedError(err error) bool {
 	return err != nil && strings.Contains(err.Error(), buildAlreadyStartedErr)
 }