Add tests to reproduce virtual thread starvation

Author: dplyukhin

.github/workflows/ci.yml

@@ -147,7 +147,7 @@ jobs:
           USER: unittest
           TEMPORAL_SERVICE_ADDRESS: localhost:7233
           USE_EXTERNAL_SERVICE: true
-        run: ./gradlew --no-daemon :temporal-sdk:virtualThreadTests -x spotlessCheck -x spotlessApply -x spotlessJava -PtestJavaVersion=21
+        run: ./gradlew --no-daemon :temporal-sdk:virtualThreadTests :temporal-sdk:virtualThreadStarvationTests -x spotlessCheck -x spotlessApply -x spotlessJava -PtestJavaVersion=21
 
       - name: Publish Test Report
         uses: mikepenz/action-junit-report@bccf2e31636835cf0874589931c4116687171386 # v6

temporal-sdk/build.gradle

@@ -256,6 +256,23 @@ testing {
             }
         }
 
+        virtualThreadStarvationTests(JvmTestSuite) {
+            targets {
+                all {
+                    testTask.configure {
+                        jvmArgs '-Djdk.virtualThreadScheduler.parallelism=1',
+                                '-Djdk.virtualThreadScheduler.maxPoolSize=1'
+                        if (project.hasProperty("testJavaVersion")) {
+                            javaLauncher = javaToolchains.launcherFor {
+                                languageVersion = JavaLanguageVersion.of(project.property("testJavaVersion") as int)
+                            }
+                        }
+                        shouldRunAfter(virtualThreadTests)
+                    }
+                }
+            }
+        }
+
         // Run the same test as the normal test task with virtual threads
         testsWithVirtualThreads(JvmTestSuite) {
             // Use the same source and resources as the main test set
@@ -287,4 +304,5 @@ testing {
 tasks.named('check') {
     dependsOn(testing.suites.jackson3Tests)
     dependsOn(testing.suites.virtualThreadTests)
-}
+    dependsOn(testing.suites.virtualThreadStarvationTests)
+}

temporal-sdk/src/main/java/io/temporal/internal/sync/PotentialDeadlockException.java

@@ -43,6 +43,23 @@ public class PotentialDeadlockException extends RuntimeException {
     this.detectionTimestamp = detectionTimestamp;
   }
 
+  /**
+   * @param workflowThreadContext context of the thread that is in a potential deadlock state
+   * @param detectionTimestamp a timestamp the deadlock was detected
+   * @param timeoutMillis configured deadlock detection timeout in milliseconds
+   */
+  PotentialDeadlockException(
+      WorkflowThreadContext workflowThreadContext,
+      long detectionTimestamp,
+      long timeoutMillis) {
+    super(
+        "[TMPRL1101] Potential deadlock detected. Workflow thread could not start executing within "
+            + formatTimeout(timeoutMillis)
+            + ".");
+    this.workflowThreadContext = workflowThreadContext;
+    this.detectionTimestamp = detectionTimestamp;
+  }
+
   private static String formatTimeout(long timeoutMillis) {
     if (timeoutMillis % 1000 == 0) {
       return timeoutMillis / 1000 + "s";

temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThreadContext.java

@@ -10,12 +10,8 @@
 import java.util.concurrent.locks.Lock;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 class WorkflowThreadContext {
-  private static final Logger log = LoggerFactory.getLogger(WorkflowThreadContext.class);
-
   // Shared runner lock
   private final Lock runnerLock;
   private final WorkflowThreadScheduler scheduler;
@@ -241,13 +237,10 @@ public boolean runUntilBlocked(long deadlockDetectionTimeoutMs) {
           throw new PotentialDeadlockException(
               currentThread.getName(), this, detectionTimestamp, deadlockDetectionTimeoutMs);
         } else {
-          // This should never happen.
-          // We clear currentThread only after setting the status to DONE.
-          // And we check for it by the status condition check after waking up on the condition
-          // and acquiring the lock back
-          log.warn("Illegal State: WorkflowThreadContext has no currentThread in {} state", status);
+          // We clear currentThread only after setting the status to DONE, so this case should
+          // only happen if the WorkflowThread is starving.
           throw new PotentialDeadlockException(
-              "UnknownThread", this, detectionTimestamp, deadlockDetectionTimeoutMs);
+              this, detectionTimestamp, deadlockDetectionTimeoutMs);
         }
       }
       Preconditions.checkState(

temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java

@@ -34,7 +34,9 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -820,6 +822,37 @@ public void testRejectedExecutionError() {
     }
   }
 
+  @Test
+  public void testThreadStarvationBeforeWorkflowThreadStarts() throws InterruptedException {
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    CountDownLatch executorThreadOccupied = new CountDownLatch(1);
+    CountDownLatch releaseExecutorThread = new CountDownLatch(1);
+    executor.submit(
+        () -> {
+          executorThreadOccupied.countDown();
+          releaseExecutorThread.await();
+          return null;
+        });
+    executorThreadOccupied.await();
+
+    DeterministicRunner runner =
+        new DeterministicRunnerImpl(
+            executor::submit,
+            DummySyncWorkflowContext.newDummySyncWorkflowContext(),
+            () -> fail("workflow code should not start while the executor thread is occupied"));
+
+    try {
+      PotentialDeadlockException e =
+          Assert.assertThrows(
+              PotentialDeadlockException.class, () -> runner.runUntilAllBlocked(10));
+      assertTrue(e.getMessage().contains("could not start executing within 10ms"));
+    } finally {
+      executor.shutdownNow();
+      releaseExecutorThread.countDown();
+      assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+    }
+  }
+
   @Test
   public void testCloseBlockedUntilDone() throws InterruptedException {
     final int THREAD_SLEEP_MS = 2000;

temporal-sdk/src/virtualThreadStarvationTests/java/io/temporal/internal/sync/ThreadStarvationVirtualThreadTest.java

@@ -0,0 +1,49 @@
+package io.temporal.internal.sync;
+
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.Test;
+
+public class ThreadStarvationVirtualThreadTest {
+
+  @Test(timeout = 10_000)
+  public void workflowThreadDoesNotStartWhenAllCarriersAreBusy() throws Exception {
+    CountDownLatch carrierOccupied = new CountDownLatch(1);
+    AtomicBoolean releaseCarrier = new AtomicBoolean();
+    Thread carrierBlocker =
+        Thread.ofVirtual()
+            .name("carrier-blocker")
+            .start(
+                () -> {
+                  carrierOccupied.countDown();
+                  while (!releaseCarrier.get()) {
+                    Thread.onSpinWait();
+                  }
+                });
+    carrierOccupied.await();
+
+    ExecutorService workflowThreadExecutor = Executors.newVirtualThreadPerTaskExecutor();
+    DeterministicRunner runner =
+        DeterministicRunner.newRunner(
+            workflowThreadExecutor::submit,
+            DummySyncWorkflowContext.newDummySyncWorkflowContext(),
+            () -> {
+              throw new AssertionError("workflow code should not start while the carrier is busy");
+            });
+
+    try {
+      PotentialDeadlockException e =
+          assertThrows(PotentialDeadlockException.class, () -> runner.runUntilAllBlocked(100));
+      assertTrue(e.getMessage().contains("could not start executing within 100ms"));
+    } finally {
+      workflowThreadExecutor.shutdownNow();
+      releaseCarrier.set(true);
+      carrierBlocker.join();
+    }
+  }
+}