Skip over unavailable resources

Author: mpkorstanje

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

@@ -226,13 +226,26 @@ void processQueueEntries(WorkerLease workerLease, BooleanSupplier doneCondition)
 					LOGGER.trace(() -> "yielding resource lock");
 					break;
 				}
-				var entry = workQueue.poll();
-				if (entry == null) {
-					LOGGER.trace(() -> "no queue entry available");
+				var queueEntries = workQueue.peekAll();
+				if (queueEntries.isEmpty()) {
+					LOGGER.trace(() -> "no queue entries available");
 					break;
 				}
-				LOGGER.trace(() -> "processing: " + entry.task);
-				execute(entry);
+				var queueEntriesByResult = tryToStealWorkWithoutBlocking(queueEntries);
+				maybeTryToStealWorkWithBlocking(queueEntriesByResult);
+			}
+		}
+
+		private void maybeTryToStealWorkWithBlocking(Map<WorkStealResult, List<WorkQueue.Entry>> queueEntriesByResult) {
+			if (queueEntriesByResult.containsKey(WorkStealResult.EXECUTED_BY_THIS_WORKER) || // 
+					queueEntriesByResult.containsKey(WorkStealResult.EXECUTED_BY_DIFFERENT_WORKER)) {
+				// Queue changed. Try to see if there is work that does not need locking
+				return;
+			}
+			// All resources locked, start blocking
+			var entriesRequiringResourceLocks = queueEntriesByResult.remove(WorkStealResult.RESOURCE_LOCK_UNAVAILABLE);
+			if (entriesRequiringResourceLocks != null) {
+				tryToStealWork(entriesRequiringResourceLocks.get(0), BlockingMode.BLOCKING);
 			}
 		}
 
@@ -557,9 +570,13 @@ private Entry doAdd(Entry entry) {
 			return entry;
 		}
 
-		@Nullable
-		Entry poll() {
-			return queue.poll();
+		private List<WorkQueue.Entry> peekAll() {
+			List<Entry> entries = new ArrayList<>(queue);
+			// Iteration order isn't the same as queue order.
+			// TODO: This makes the queue kinda pointless
+			// TODO: This also makes retries pointless
+			entries.sort(naturalOrder());
+			return entries;
 		}
 
 		boolean remove(Entry entry) {

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

@@ -227,6 +227,87 @@ void runsTasksWithoutConflictingLocksConcurrently() throws Exception {
 		assertThat(leaves).allSatisfy(TestTaskStub::assertExecutedSuccessfully);
 	}
 
+
+	@Test
+	void processingQueueEntriesSkipsOverUnavailableResources() throws Exception {
+		service = new ConcurrentHierarchicalTestExecutorService(configuration(2));
+
+		var resourceLock = new SingleLock(exclusiveResource(LockMode.READ_WRITE), new ReentrantLock());
+		
+		var lockFreeChildrenStarted = new CountDownLatch(2);
+		var child1Started = new CountDownLatch(1);
+
+		Executable child1Behaviour = () -> {
+			child1Started.countDown();
+			lockFreeChildrenStarted.await();
+		};
+		Executable child4Behaviour = () -> {
+			lockFreeChildrenStarted.countDown();
+			child1Started.await();
+		};
+		
+		var child1 = new TestTaskStub(ExecutionMode.CONCURRENT, child1Behaviour)  //
+				.withResourceLock(resourceLock) //
+				.withName("child1");
+		var child2 = new TestTaskStub(ExecutionMode.CONCURRENT, lockFreeChildrenStarted::countDown)
+				.withName("child2"); //
+		var child3 = new TestTaskStub(ExecutionMode.CONCURRENT)
+				.withResourceLock(resourceLock) //
+				.withName("child3");
+		var child4 = new TestTaskStub(ExecutionMode.CONCURRENT, child4Behaviour)
+				.withName("child4");
+		var children = List.of(child1, child2, child3, child4);
+		var root = new TestTaskStub(ExecutionMode.CONCURRENT, () -> requiredService().invokeAll(children)) //
+				.withName("root");
+
+		service.submit(root).get();
+
+		root.assertExecutedSuccessfully();
+		assertThat(children).allSatisfy(TestTaskStub::assertExecutedSuccessfully);
+		assertThat(child4.executionThread).isEqualTo(child2.executionThread);
+		assertThat(child3.startTime).isAfterOrEqualTo(child2.startTime);
+	}
+
+	@Test
+	void invokeAllQueueEntriesSkipsOverUnavailableResources() throws Exception {
+		service = new ConcurrentHierarchicalTestExecutorService(configuration(2));
+
+		var resourceLock = new SingleLock(exclusiveResource(LockMode.READ_WRITE), new ReentrantLock());
+		
+		var lockFreeChildrenStarted = new CountDownLatch(2);
+		var child4Started = new CountDownLatch(1);
+
+		Executable child1Behaviour = () -> {
+			lockFreeChildrenStarted.countDown();
+			child4Started.await();
+		};
+		Executable child4Behaviour = () -> {
+			child4Started.countDown();
+			lockFreeChildrenStarted.await();
+		};
+		
+		var child1 = new TestTaskStub(ExecutionMode.CONCURRENT, child1Behaviour)  //
+				.withName("child1");
+		var child2 = new TestTaskStub(ExecutionMode.CONCURRENT)
+				.withResourceLock(resourceLock) //
+				.withName("child2"); //
+		var child3 = new TestTaskStub(ExecutionMode.CONCURRENT, lockFreeChildrenStarted::countDown)
+				.withName("child3");
+		var child4 = new TestTaskStub(ExecutionMode.CONCURRENT, child4Behaviour)
+				.withResourceLock(resourceLock) //
+				.withName("child4");
+		var children = List.of(child1, child2, child3, child4);
+		var root = new TestTaskStub(ExecutionMode.CONCURRENT, () -> requiredService().invokeAll(children)) //
+				.withName("root");
+
+		service.submit(root).get();
+
+		root.assertExecutedSuccessfully();
+		assertThat(children).allSatisfy(TestTaskStub::assertExecutedSuccessfully);
+		assertThat(child1.executionThread).isEqualTo(child3.executionThread);
+		assertThat(child2.startTime).isAfterOrEqualTo(child3.startTime);
+	}
+
 	@Test
 	void prioritizesChildrenOfStartedContainers() throws Exception {
 		service = new ConcurrentHierarchicalTestExecutorService(configuration(2));