Use executor service for triggering

3 files changed, 59 insertions, 13 deletions

diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/ApplicationController.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/ApplicationController.java
index 1eae7317b7b..8d231e150e8 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/ApplicationController.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/ApplicationController.java
@@ -40,7 +40,6 @@ import com.yahoo.vespa.hosted.controller.application.ApplicationVersion;
 import com.yahoo.vespa.hosted.controller.tenant.AthenzTenant;
 import com.yahoo.vespa.hosted.controller.application.Deployment;
 import com.yahoo.vespa.hosted.controller.application.DeploymentJobs;
-import com.yahoo.vespa.hosted.controller.application.DeploymentJobs.JobReport;
 import com.yahoo.vespa.hosted.controller.tenant.Tenant;
 import com.yahoo.vespa.hosted.controller.deployment.DeploymentTrigger;
 import com.yahoo.vespa.hosted.controller.maintenance.DeploymentExpirer;
diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentTrigger.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentTrigger.java
index a421f7a11c5..09f49391d14 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentTrigger.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/deployment/DeploymentTrigger.java
@@ -23,23 +23,34 @@ import com.yahoo.vespa.hosted.controller.application.JobList;
 import com.yahoo.vespa.hosted.controller.application.JobStatus;
 import com.yahoo.vespa.hosted.controller.persistence.CuratorDb;
 
+import java.io.UncheckedIOException;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Logger;
+import java.util.stream.Stream;
 
 import static java.util.Comparator.comparing;
 import static java.util.Comparator.naturalOrder;
 import static java.util.stream.Collectors.groupingBy;
 import static java.util.stream.Collectors.joining;
+import static java.util.stream.Collectors.partitioningBy;
+import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toSet;
 
 /**
@@ -65,6 +76,7 @@ public class DeploymentTrigger {
     private final Clock clock;
     private final DeploymentOrder order;
     private final BuildService buildService;
+    private final ExecutorService executor;
 
     public DeploymentTrigger(Controller controller, CuratorDb curator, BuildService buildService, Clock clock) {
         Objects.requireNonNull(controller, "controller cannot be null");
@@ -75,6 +87,7 @@ public class DeploymentTrigger {
         this.order = new DeploymentOrder(controller::system);
         this.buildService = buildService;
         this.jobTimeout = controller.system().equals(SystemName.main) ? Duration.ofHours(12) : Duration.ofHours(1);
+        this.executor = Executors.newFixedThreadPool(20);
     }
 
     /**
@@ -124,10 +137,46 @@ public class DeploymentTrigger {
      * Only one job is triggered each run for test jobs, since those environments have limited capacity.
      */
     public void triggerReadyJobs() {
-        computeReadyJobs().forEach((jobType, jobs) -> {
-            for (Job job : jobs)
-                if (canTrigger(job) && trigger(job) && jobType.isTest()) break;
-        });
+        List<Callable<Void>> tasks = new ArrayList<>();
+
+        computeReadyJobs().collect(partitioningBy(job -> job.jobType().isTest()))
+                          .forEach((isConstrained, jobList) -> {
+                              if (isConstrained) {
+                                  jobList.stream()
+                                         .sorted(comparing(Job::isRetry)
+                                                         .thenComparing(Job::applicationUpgrade)
+                                                         .reversed()
+                                                         .thenComparing(Job::availableSince))
+                                         .collect(groupingBy(Job::jobType))
+                                         .values().forEach(jobs -> tasks.add(
+                                          () -> {
+                                              for (Job job : jobs)
+                                                  if (canTrigger(job) && trigger(job))
+                                                      break;
+                                              return null;
+                                          }));
+                              }
+                              else {
+                                  jobList.stream()
+                                         .collect(groupingBy(Job::id))
+                                         .values().forEach(jobs -> tasks.add(
+                                          () -> {
+                                              for (Job job : jobs)
+                                                  applications().lockIfPresent(job.id, ignored -> {
+                                                      if (canTrigger(job))
+                                                          trigger(job);
+                                                  });
+                                              return null;
+                                          }));
+                              }
+                          });
+
+        try {
+            executor.invokeAll(tasks);
+        }
+        catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
     }
 
     /**
@@ -229,19 +278,14 @@ public class DeploymentTrigger {
     /**
      * Returns the set of all jobs which have changes to propagate from the upstream steps, sorted by job.
      */
-    public Map<JobType, List<Job>> computeReadyJobs() {
+    public Stream<Job> computeReadyJobs() {
         return ApplicationList.from(applications().asList())
                               .notPullRequest()
                               .withProjectId()
                               .deploying()
                               .idList().stream()
                               .map(this::computeReadyJobs)
-                              .flatMap(List::stream)
-                              .sorted(comparing(Job::isRetry)
-                                              .thenComparing(Job::applicationUpgrade)
-                                              .reversed()
-                                              .thenComparing(Job::availableSince))
-                              .collect(groupingBy(Job::jobType));
+                              .flatMap(List::stream);
     }
 
     private Optional<Instant> completedAt(Application application, JobType jobType) {
diff --git a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/screwdriver/ScrewdriverApiHandler.java b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/screwdriver/ScrewdriverApiHandler.java
index 86e655dc9b0..3db15806f26 100644
--- a/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/screwdriver/ScrewdriverApiHandler.java
+++ b/controller-server/src/main/java/com/yahoo/vespa/hosted/controller/restapi/screwdriver/ScrewdriverApiHandler.java
@@ -26,6 +26,9 @@ import java.util.Optional;
 import java.util.Scanner;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.stream.Collectors;
+
+import static java.util.stream.Collectors.groupingBy;
 
 /**
  * This API lists deployment jobs that are queued for execution on Screwdriver.
@@ -68,7 +71,7 @@ public class ScrewdriverApiHandler extends LoggingRequestHandler {
             return vespaVersion();
         }
         if (path.matches("/screwdriver/v1/jobsToRun"))
-            return buildJobs(controller.applications().deploymentTrigger().computeReadyJobs());
+            return buildJobs(controller.applications().deploymentTrigger().computeReadyJobs().collect(groupingBy(job -> job.jobType())));
         return notFound(request);
     }