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 efafe230a..cebe4babd 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/PotentialDeadlockException.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/PotentialDeadlockException.java
index 4f907783b..2e952d504 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,21 @@ 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/WorkflowThread.java b/temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowThread.java
index fecd98e74..c79318d55 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 49dd9f946..446db368c 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;
@@ -214,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.
*/
@@ -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/main/java/io/temporal/worker/WorkerOptions.java b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java
index 3da57ef81..aa61c6fa4 100644
--- a/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java
+++ b/temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java
@@ -327,9 +327,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
*/
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 ba3a0eb33..a2d9cadc6 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,39 @@ 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 {
+ // Try starting the root workflow thread, but it won't run because the executor is tied up.
+ // We should detect this and throw a PotentialDeadlockException.
+ 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;