Update PotentialDeadlockException to account for scheduling delay (#2964)

* Add tests to reproduce thread starvation

* Update docs for deadlock detection timeout

Author: dplyukhin

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);
 

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";

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);

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(

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.
-     *     <p>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.
+     *     <p>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
      */

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;