Make worker start at head of queue again after executing a node

Author: marcphilipp

junit-platform-engine/src/main/java/org/junit/platform/engine/support/hierarchical/WorkerThreadPoolHierarchicalTestExecutorService.java

@@ -14,6 +14,7 @@
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.CompletableFuture.completedFuture;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static java.util.function.Predicate.isEqual;
 import static org.apiguardian.api.API.Status.EXPERIMENTAL;
 import static org.junit.platform.commons.util.ExceptionUtils.throwAsUncheckedException;
 import static org.junit.platform.engine.support.hierarchical.ExclusiveResource.GLOBAL_READ_WRITE;
@@ -45,6 +46,7 @@
 import java.util.function.BiFunction;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
+import java.util.function.Predicate;
 
 import org.apiguardian.api.API;
 import org.jspecify.annotations.Nullable;
@@ -292,7 +294,8 @@ void processQueueEntries(WorkerLease workerLease, BooleanSupplier doneCondition)
 		}
 
 		private void processQueueEntries() {
-			var queueEntriesByResult = tryToStealWorkWithoutBlocking(workQueue);
+			var queueEntriesByResult = tryToStealWorkWithoutBlocking(workQueue,
+				isEqual(WorkStealResult.EXECUTED_BY_THIS_WORKER));
 			var queueModified = queueEntriesByResult.containsKey(WorkStealResult.EXECUTED_BY_THIS_WORKER) //
 					|| queueEntriesByResult.containsKey(WorkStealResult.EXECUTED_BY_DIFFERENT_WORKER);
 			if (queueModified) {
@@ -336,7 +339,7 @@ void invokeAll(List<? extends TestTask> testTasks) {
 			List<TestTask> sameThreadTasks = new ArrayList<>(testTasks.size());
 			var queueEntries = forkConcurrentChildren(testTasks, isolatedTasks::add, sameThreadTasks);
 			executeAll(sameThreadTasks);
-			var queueEntriesByResult = tryToStealWorkWithoutBlocking(queueEntries);
+			var queueEntriesByResult = tryToStealWorkWithoutBlocking(queueEntries, __ -> false);
 			tryToStealWorkWithBlocking(queueEntriesByResult);
 			waitFor(queueEntriesByResult);
 			executeAll(isolatedTasks);
@@ -371,10 +374,10 @@ else if (child.getExecutionMode() == SAME_THREAD) {
 		}
 
 		private Map<WorkStealResult, List<WorkQueue.Entry>> tryToStealWorkWithoutBlocking(
-				Iterable<WorkQueue.Entry> queueEntries) {
+				Iterable<WorkQueue.Entry> queueEntries, Predicate<? super WorkStealResult> stopCondition) {
 
 			Map<WorkStealResult, List<WorkQueue.Entry>> queueEntriesByResult = new EnumMap<>(WorkStealResult.class);
-			tryToStealWork(queueEntries, BlockingMode.NON_BLOCKING, queueEntriesByResult);
+			tryToStealWork(queueEntries, BlockingMode.NON_BLOCKING, queueEntriesByResult, stopCondition);
 			return queueEntriesByResult;
 		}
 
@@ -383,14 +386,18 @@ private void tryToStealWorkWithBlocking(Map<WorkStealResult, List<WorkQueue.Entr
 			if (entriesRequiringResourceLocks == null) {
 				return;
 			}
-			tryToStealWork(entriesRequiringResourceLocks, BlockingMode.BLOCKING, queueEntriesByResult);
+			tryToStealWork(entriesRequiringResourceLocks, BlockingMode.BLOCKING, queueEntriesByResult, __ -> false);
 		}
 
 		private void tryToStealWork(Iterable<WorkQueue.Entry> entries, BlockingMode blocking,
-				Map<WorkStealResult, List<WorkQueue.Entry>> queueEntriesByResult) {
+				Map<WorkStealResult, List<WorkQueue.Entry>> queueEntriesByResult,
+				Predicate<? super WorkStealResult> stopCondition) {
 			for (var entry : entries) {
-				var state = tryToStealWork(entry, blocking);
-				queueEntriesByResult.computeIfAbsent(state, __ -> new ArrayList<>()).add(entry);
+				var result = tryToStealWork(entry, blocking);
+				queueEntriesByResult.computeIfAbsent(result, __ -> new ArrayList<>()).add(entry);
+				if (stopCondition.test(result)) {
+					break;
+				}
 			}
 		}
 

platform-tests/src/test/java/org/junit/platform/engine/support/hierarchical/WorkerThreadPoolHierarchicalTestExecutorServiceTests.java

@@ -42,7 +42,6 @@
 import org.jspecify.annotations.NullMarked;
 import org.jspecify.annotations.Nullable;
 import org.junit.jupiter.api.AutoClose;
-import org.junit.jupiter.api.RepeatedTest;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.api.function.Executable;
@@ -305,19 +304,28 @@ void invokeAllQueueEntriesSkipsOverUnavailableResources() throws Exception {
 		assertThat(child3.startTime).isAfterOrEqualTo(child4.startTime);
 	}
 
-	@RepeatedTest(value = 100, failureThreshold = 1)
+	@Test
 	void prioritizesChildrenOfStartedContainers() throws Exception {
-		service = new WorkerThreadPoolHierarchicalTestExecutorService(configuration(2));
+		service = new WorkerThreadPoolHierarchicalTestExecutorService(configuration(2, 2));
 
-		var leavesStarted = new CountDownLatch(2);
+		var leafSubmitted = new CountDownLatch(1);
+		var child2AndLeafStarted = new CountDownLatch(2);
+
+		var leaf = new TestTaskStub(ExecutionMode.CONCURRENT, child2AndLeafStarted::countDown) //
+				.withName("leaf").withLevel(3);
 
-		var child1 = new TestTaskStub(ExecutionMode.CONCURRENT, leavesStarted::await) //
+		Executable child3Behavior = () -> {
+			var future = requiredService().submit(leaf);
+			leafSubmitted.countDown();
+			child2AndLeafStarted.await();
+			future.get();
+		};
+
+		var child1 = new TestTaskStub(ExecutionMode.CONCURRENT, leafSubmitted::await) //
 				.withName("child1").withLevel(2);
-		var child2 = new TestTaskStub(ExecutionMode.CONCURRENT, leavesStarted::countDown) //
+		var child2 = new TestTaskStub(ExecutionMode.CONCURRENT, child2AndLeafStarted::countDown) //
 				.withName("child2").withLevel(2);
-		var leaf = new TestTaskStub(ExecutionMode.CONCURRENT, leavesStarted::countDown) //
-				.withName("leaf").withLevel(3);
-		var child3 = new TestTaskStub(ExecutionMode.CONCURRENT, () -> requiredService().submit(leaf).get()) //
+		var child3 = new TestTaskStub(ExecutionMode.CONCURRENT, child3Behavior) //
 				.withType(CONTAINER).withName("child3").withLevel(2);
 
 		var root = new TestTaskStub(ExecutionMode.SAME_THREAD,
@@ -327,11 +335,10 @@ void prioritizesChildrenOfStartedContainers() throws Exception {
 		service.submit(root).get();
 
 		root.assertExecutedSuccessfully();
-		assertThat(List.of(child1, child2, leaf, child3)).allSatisfy(TestTaskStub::assertExecutedSuccessfully);
-		leaf.assertExecutedSuccessfully();
+		assertThat(List.of(root, child1, child2, leaf, child3)).allSatisfy(TestTaskStub::assertExecutedSuccessfully);
 
 		assertThat(leaf.startTime).isBeforeOrEqualTo(child2.startTime);
-		assertThat(leaf.executionThread).isSameAs(child3.executionThread);
+		assertThat(leaf.executionThread).isSameAs(child2.executionThread).isNotSameAs(child3.executionThread);
 	}
 
 	@Test