Synchronize aggregate test runner threads

1 files changed, 47 insertions, 38 deletions

diff --git a/vespa-osgi-testrunner/src/main/java/com/yahoo/vespa/testrunner/AggregateTestRunner.java b/vespa-osgi-testrunner/src/main/java/com/yahoo/vespa/testrunner/AggregateTestRunner.java
index 6e3393b2761..5c54349335b 100644
--- a/vespa-osgi-testrunner/src/main/java/com/yahoo/vespa/testrunner/AggregateTestRunner.java
+++ b/vespa-osgi-testrunner/src/main/java/com/yahoo/vespa/testrunner/AggregateTestRunner.java
@@ -4,9 +4,7 @@ package com.yahoo.vespa.testrunner;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.LogRecord;
 
 /**
@@ -15,7 +13,8 @@ import java.util.logging.LogRecord;
 public class AggregateTestRunner implements TestRunner {
 
     private final List<TestRunner> wrapped;
-    private final AtomicInteger current = new AtomicInteger(-1);
+    private volatile int current = -1;
+    private final Object monitor = new Object();
 
     private AggregateTestRunner(List<TestRunner> testRunners) {
         this.wrapped = testRunners;
@@ -28,7 +27,7 @@ public class AggregateTestRunner implements TestRunner {
     @Override
     public Collection<LogRecord> getLog(long after) {
         ArrayList<LogRecord> records = new ArrayList<>();
-        for (int i = 0; i <= current.get() && i < wrapped.size(); i++)
+        for (int i = 0; i <= current && i < wrapped.size(); i++)
             records.addAll(wrapped.get(i).getLog(after));
 
         return records;
@@ -36,57 +35,67 @@ public class AggregateTestRunner implements TestRunner {
 
     @Override
     public Status getStatus() {
-        if (current.get() == -1)
-            return Status.NOT_STARTED;
-
-        Status status = Status.NO_TESTS;
-        for (int i = 0; i <= current.get(); i++) {
-            if (i == wrapped.size())
-                return status;
-
-            Status next = wrapped.get(i).getStatus();
-            status = status.ordinal() < next.ordinal() ? status : next;
+        synchronized (monitor) {
+            if (current == -1)
+                return Status.NOT_STARTED;
+
+            Status status = Status.NO_TESTS;
+            for (int i = 0; i <= current; i++) {
+                if (i == wrapped.size())
+                    return status;
+
+                Status next = wrapped.get(i).getStatus();
+                status = status.ordinal() < next.ordinal() ? status : next;
+            }
+            return Status.RUNNING;
         }
-        return Status.RUNNING;
     }
 
     @Override
     public CompletableFuture<?> test(Suite suite, byte[] config) {
-        if (0 <= current.get() && current.get() < wrapped.size())
-            throw new IllegalStateException("Tests already running, should not attempt to start now");
-
-        current.set(-1);
-        CompletableFuture<?> aggregate = new CompletableFuture<>();
-        CompletableFuture<?> vessel = CompletableFuture.completedFuture(null);
-        runNext(suite, config, vessel, aggregate);
-        return aggregate;
+        synchronized (monitor) {
+            if (0 <= current && current < wrapped.size())
+                throw new IllegalStateException("Tests already running, should not attempt to start now");
+
+            current = -1;
+            CompletableFuture<?> aggregate = new CompletableFuture<>();
+            CompletableFuture<?> vessel = CompletableFuture.completedFuture(null);
+            runNext(suite, config, vessel, aggregate);
+            return aggregate;
+        }
     }
 
     private void runNext(Suite suite, byte[] config, CompletableFuture<?> vessel, CompletableFuture<?> aggregate) {
         vessel.whenComplete((__, ___) -> {
-            int next = current.incrementAndGet();
-            if (next == wrapped.size())
-                aggregate.complete(null);
-            else
-                runNext(suite, config, wrapped.get(next).test(suite, config), aggregate);
+            synchronized (monitor) {
+                if (++current < wrapped.size())
+                    runNext(suite, config, wrapped.get(current).test(suite, config), aggregate);
+                else
+                    aggregate.complete(null);
+            }
         });
     }
 
     @Override
     public TestReport getReport() {
-        return wrapped.stream().map(TestRunner::getReport).filter(Objects::nonNull)
-                      .reduce(AggregateTestRunner::merge).orElse(null);
+        TestReport report = null;
+        for (int i = 0; i < current && i < wrapped.size(); i++)
+            report = merge(report, wrapped.get(i).getReport());
+
+        return report;
     }
 
     static TestReport merge(TestReport first, TestReport second) {
-        return TestReport.builder()
-                         .withAbortedCount(first.abortedCount + second.abortedCount)
-                         .withFailedCount(first.failedCount + second.failedCount)
-                         .withIgnoredCount(first.ignoredCount + second.ignoredCount)
-                         .withSuccessCount(first.successCount + second.successCount)
-                         .withFailures(merged(first.failures, second.failures))
-                         .withLogs(merged(first.logLines, second.logLines))
-                         .build();
+        return first == null ? second
+                             : second == null ? first
+                                              : TestReport.builder()
+                                                          .withAbortedCount(first.abortedCount + second.abortedCount)
+                                                          .withFailedCount(first.failedCount + second.failedCount)
+                                                          .withIgnoredCount(first.ignoredCount + second.ignoredCount)
+                                                          .withSuccessCount(first.successCount + second.successCount)
+                                                          .withFailures(merged(first.failures, second.failures))
+                                                          .withLogs(merged(first.logLines, second.logLines))
+                                                          .build();
     }
 
     static <T> List<T> merged(List<T> first, List<T> second) {