From b080ca3d267b84d916e2eee346089bbace3cd97f Mon Sep 17 00:00:00 2001 From: Dan Plyukhin Date: Wed, 22 Jul 2026 12:52:19 -0400 Subject: [PATCH 1/2] Add tests to reproduce virtual thread starvation --- .github/workflows/ci.yml | 2 +- temporal-sdk/build.gradle | 20 +++++++- .../sync/PotentialDeadlockException.java | 17 +++++++ .../internal/sync/WorkflowThreadContext.java | 13 ++--- .../sync/DeterministicRunnerTest.java | 33 +++++++++++++ .../ThreadStarvationVirtualThreadTest.java | 49 +++++++++++++++++++ 6 files changed, 122 insertions(+), 12 deletions(-) create mode 100644 temporal-sdk/src/virtualThreadStarvationTests/java/io/temporal/internal/sync/ThreadStarvationVirtualThreadTest.java diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 598d8d02ef..5efc781402 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/temporal-sdk/build.gradle b/temporal-sdk/build.gradle index d7b090e5b5..9534e1e6ff 100644 --- a/temporal-sdk/build.gradle +++ b/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) -} \ No newline at end of file + dependsOn(testing.suites.virtualThreadStarvationTests) +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/PotentialDeadlockException.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/PotentialDeadlockException.java index 4f907783b0..e696b0fd6d 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/PotentialDeadlockException.java +++ b/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"; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThreadContext.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThreadContext.java index 49dd9f9468..0148297169 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThreadContext.java +++ b/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( diff --git a/temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java b/temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java index ba3a0eb332..02f306fd7d 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java +++ b/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; diff --git a/temporal-sdk/src/virtualThreadStarvationTests/java/io/temporal/internal/sync/ThreadStarvationVirtualThreadTest.java b/temporal-sdk/src/virtualThreadStarvationTests/java/io/temporal/internal/sync/ThreadStarvationVirtualThreadTest.java new file mode 100644 index 0000000000..a9baa4fec5 --- /dev/null +++ b/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(); + } + } +} From 78b7d3a373cc656e2a6974aa9be0615be50e070e Mon Sep 17 00:00:00 2001 From: Dan Plyukhin Date: Thu, 23 Jul 2026 17:38:06 -0400 Subject: [PATCH 2/2] Update docs for deadlock detection timeout --- .../java/io/temporal/internal/sync/DeterministicRunner.java | 4 ++-- .../main/java/io/temporal/internal/sync/WorkflowThread.java | 4 ++-- .../io/temporal/internal/sync/WorkflowThreadContext.java | 4 ++-- .../src/main/java/io/temporal/worker/WorkerOptions.java | 6 +++--- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunner.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunner.java index efafe230af..2147b518d4 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunner.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunner.java @@ -52,8 +52,8 @@ static DeterministicRunner newRunner( * completed or blocked. * * @throws Throwable if one of the threads didn't handle an exception. - * @param deadlockDetectionTimeout the maximum time in milliseconds a thread can run without - * calling yield. + * @param deadlockDetectionTimeout the maximum time in milliseconds after a thread is + * scheduled until it yields or completes. */ void runUntilAllBlocked(long deadlockDetectionTimeout); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThread.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThread.java index fecd98e74f..688abfce7b 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThread.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThread.java @@ -65,8 +65,8 @@ static WorkflowThread newThread(Runnable runnable, boolean detached, String name SyncWorkflowContext getWorkflowContext(); /** - * @param deadlockDetectionTimeoutMs maximum time in milliseconds the thread can run before - * calling yield. + * @param deadlockDetectionTimeoutMs the maximum time in milliseconds after a thread is + * scheduled until it yields or completes. * @return true if coroutine made some progress. */ boolean runUntilBlocked(long deadlockDetectionTimeoutMs); diff --git a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThreadContext.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThreadContext.java index 0148297169..446db368c9 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThreadContext.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThreadContext.java @@ -210,8 +210,8 @@ public String getYieldReason() { } /** - * @param deadlockDetectionTimeoutMs maximum time in milliseconds the thread can run before - * calling yield. Discarded if {@code TEMPORAL_DEBUG} env variable is set. + * @param deadlockDetectionTimeoutMs the maximum time in milliseconds after a thread is scheduled + * until it yields or completes. Discarded if {@code TEMPORAL_DEBUG} env variable is set. * @return true if thread made some progress. Which is await was unblocked and some code after it * * was executed. */ diff --git a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java index 4e351cb74d..a3bb649345 100644 --- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java +++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java @@ -302,9 +302,9 @@ public Builder setLocalActivityWorkerOnly(boolean localActivityWorkerOnly) { /** * @param defaultDeadlockDetectionTimeoutMs time period in ms that will be used to detect * workflows deadlock. Default is 1000ms, which is chosen if set to zero. - *

Specifies an amount of time in milliseconds that workflow tasks are allowed to execute - * without interruption. If workflow task runs longer than specified interval without - * yielding (like calling an Activity), it will fail automatically. + *

Specifies a time interval in milliseconds within which a workflow task must + * yield (like calling an Activity) or complete. If a workflow task runs longer than + * the specified interval or takes too long to begin running, it will fail automatically. * @return {@code this} * @see io.temporal.internal.sync.PotentialDeadlockException */