Add log warning if workflow eviction / closure is taking too long time to close

Add guards making sure all threads are drained.

Author: Spikhalskiy

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

@@ -298,15 +298,14 @@ private static <R> Promise<R> execute(boolean temporalStub, Functions.Func<R> fu
     } else {
       CompletablePromise<R> result = Workflow.newPromise();
       WorkflowThread.newThread(
-              () -> {
-                try {
-                  result.complete(func.apply());
-                } catch (Exception e) {
-                  result.completeExceptionally(Workflow.wrap(e));
-                }
-              },
-              false)
-          .start();
+          () -> {
+            try {
+              result.complete(func.apply());
+            } catch (Exception e) {
+              result.completeExceptionally(Workflow.wrap(e));
+            }
+          },
+          false);
       return result;
     }
   }

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

@@ -37,9 +37,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
+import java.util.concurrent.*;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Supplier;
@@ -192,18 +190,19 @@ public void runUntilAllBlocked(long deadlockDetectionTimeout) {
       threads.add(rootWorkflowThread);
       rootWorkflowThread.start();
     }
-    if (!rootWorkflowThread.isStarted()) {
-      throw new IllegalStateException("start not called");
-    }
     lock.lock();
     try {
       checkNotClosed();
-
+      checkNotCloseRequestedLocked();
       inRunUntilAllBlocked = true;
       // Keep repeating until at least one of the threads makes progress.
       boolean progress;
       outerLoop:
       do {
+        if (exitRequested) {
+          closeRequested = true;
+          break;
+        }
         if (!toExecuteInWorkflowThread.isEmpty()) {
           for (NamedRunnable nr : toExecuteInWorkflowThread) {
             Object callbackThread =
@@ -222,15 +221,7 @@ public void runUntilAllBlocked(long deadlockDetectionTimeout) {
                     + callbackThread);
           }
 
-          // It is important to prepend threads as there are callbacks
-          // like signals that have to run before any other threads.
-          // Otherwise signal might be never processed if it was received
-          // after workflow decided to close.
-          // Adding the callbacks in the same order as they appear in history.
-          for (int i = callbackThreadsToAdd.size() - 1; i >= 0; i--) {
-            threads.add(callbackThreadsToAdd.get(i));
-          }
-          callbackThreadsToAdd.clear();
+          appendCallbackThreadsLocked();
         }
         toExecuteInWorkflowThread.clear();
         progress = false;
@@ -251,8 +242,7 @@ public void runUntilAllBlocked(long deadlockDetectionTimeout) {
             }
           }
         }
-        threads.addAll(workflowThreadsToAdd);
-        workflowThreadsToAdd.clear();
+        appendWorkflowThreadsLocked();
       } while (progress && !threads.isEmpty());
     } catch (PotentialDeadlockException e) {
       String triggerThreadStackTrace = "";
@@ -284,7 +274,9 @@ public void runUntilAllBlocked(long deadlockDetectionTimeout) {
   public boolean isDone() {
     lock.lock();
     try {
-      return closeFuture.isDone() || threads.isEmpty();
+      return closeFuture.isDone()
+          // if close is requested, we should wait for the closeFuture to be filled
+          || !closeRequested && !areThreadsToBeExecuted();
     } finally {
       lock.unlock();
     }
@@ -302,7 +294,6 @@ public void cancel(String reason) {
    */
   @Override
   public void close() {
-    List<Future<?>> threadFutures = new ArrayList<>();
     lock.lock();
     if (closeFuture.isDone()) {
       lock.unlock();
@@ -324,52 +315,80 @@ public void close() {
       closeFuture.join();
       return;
     }
+
     closeStarted = true;
+    // lock is taken here
     try {
-      try {
-        threads.addAll(workflowThreadsToAdd);
-        workflowThreadsToAdd.clear();
-
-        for (WorkflowThread c : threads) {
-          threadFutures.add(c.stopNow());
-        }
-        threads.clear();
-
-        // We cannot use an iterator to unregister failed Promises since f.get()
-        // will remove the promise directly from failedPromises. This causes an
-        // ConcurrentModificationException
-        // For this reason we will loop over a copy of failedPromises.
-        Set<Promise<?>> failedPromisesLoop = new HashSet<>(failedPromises);
-        for (Promise<?> f : failedPromisesLoop) {
-          try {
-            f.get();
-            throw new Error("unreachable");
-          } catch (RuntimeException e) {
-            log.warn(
-                "Promise completed with exception and was never accessed. The ignored exception:",
-                CheckedExceptionWrapper.unwrap(e));
+      // in some circumstances when a workflow broke Deadline Detector,
+      // runUntilAllBlocked may return while workflow threads are still running.
+      // If this happens, these threads may potentially start new additional threads that will be
+      // in workflowThreadsToAdd and callbackThreadsToAdd.
+      // That's why we need to make sure that all the spawned threads were shut down in a cycle.
+      while (areThreadsToBeExecuted()) {
+        List<WorkflowThreadStopFuture> threadFutures = new ArrayList<>();
+        try {
+          toExecuteInWorkflowThread.clear();
+          appendWorkflowThreadsLocked();
+          appendCallbackThreadsLocked();
+          for (WorkflowThread workflowThread : threads) {
+            threadFutures.add(
+                new WorkflowThreadStopFuture(workflowThread, workflowThread.stopNow()));
           }
+          threads.clear();
+
+          // We cannot use an iterator to unregister failed Promises since f.get()
+          // will remove the promise directly from failedPromises. This causes an
+          // ConcurrentModificationException
+          // For this reason we will loop over a copy of failedPromises.
+          Set<Promise<?>> failedPromisesLoop = new HashSet<>(failedPromises);
+          for (Promise<?> f : failedPromisesLoop) {
+            try {
+              f.get();
+              throw new Error("unreachable");
+            } catch (RuntimeException e) {
+              log.warn(
+                  "Promise completed with exception and was never accessed. The ignored exception:",
+                  CheckedExceptionWrapper.unwrap(e));
+            }
+          }
+        } finally {
+          // we need to unlock for the further code because threads will not be able to proceed with
+          // destruction otherwise.
+          lock.unlock();
         }
-      } finally {
-        lock.unlock();
-      }
 
-      // Context is destroyed in c.StopNow(). Wait on all tasks outside the lock since
-      // these tasks use the same lock to execute.
-      for (Future<?> future : threadFutures) {
+        // Wait on all tasks outside the lock since these tasks use the same lock to execute.
         try {
-          future.get();
+          for (WorkflowThreadStopFuture threadFuture : threadFutures) {
+            try {
+              threadFuture.stopFuture.get(10, TimeUnit.SECONDS);
+            } catch (TimeoutException e) {
+              WorkflowThread workflowThread = threadFuture.workflowThread;
+              log.error(
+                  "[BUG] Workflow thread '{}' of workflow '{}' can't be destroyed in time. "
+                      + "This will lead to a workflow cache leak. "
+                      + "This problem is usually caused by a workflow implementation swallowing java.lang.Error instead of rethrowing it. "
+                      + " Thread dump of the stuck thread:\n{}",
+                  workflowThread.getName(),
+                  workflowContext.getContext().getWorkflowId(),
+                  workflowThread.getStackTrace());
+            }
+          }
         } catch (InterruptedException e) {
           Thread.currentThread().interrupt();
           // Worker is likely stopped with shutdownNow()
           // TODO consider propagating as an original interrupted exception to the top level
           throw new Error("Worker executor thread interrupted during stopping of a coroutine", e);
         } catch (ExecutionException e) {
           throw new Error("[BUG] Unexpected failure while stopping a coroutine", e);
+        } finally {
+          // acquire the lock back as it should be taken for the loop condition check.
+          lock.lock();
         }
       }
     } finally {
       closeFuture.complete(null);
+      lock.unlock();
     }
   }
 
@@ -393,6 +412,26 @@ public String stackTrace() {
     return result.toString();
   }
 
+  private void appendWorkflowThreadsLocked() {
+    threads.addAll(workflowThreadsToAdd);
+    workflowThreadsToAdd.clear();
+  }
+
+  private void appendCallbackThreadsLocked() {
+    // TODO I'm not sure this comment makes sense, because threads list has comparator and we use
+    // thread priorities anyway.
+
+    // It is important to prepend threads as there are callbacks
+    // like signals that have to run before any other threads.
+    // Otherwise signal might be never processed if it was received
+    // after workflow decided to close.
+    // Adding the callbacks in the same order as they appear in history.
+    for (int i = callbackThreadsToAdd.size() - 1; i >= 0; i--) {
+      threads.add(callbackThreadsToAdd.get(i));
+    }
+    callbackThreadsToAdd.clear();
+  }
+
   /** Creates a new instance of a root workflow thread. */
   private WorkflowThread newRootThread(Runnable runnable) {
     String name = WORKFLOW_ROOT_THREAD_NAME;
@@ -519,12 +558,28 @@ private void checkWorkflowThreadOnly() {
     }
   }
 
+  private void checkNotCloseRequestedLocked() {
+    if (closeRequested) {
+      throw new Error("close requested");
+    }
+  }
+
   private void checkNotClosed() {
     if (closeFuture.isDone()) {
       throw new Error("closed");
     }
   }
 
+  /**
+   * @return true if there are no threads left to be processed for this workflow.
+   */
+  private boolean areThreadsToBeExecuted() {
+    return !threads.isEmpty()
+        || !workflowThreadsToAdd.isEmpty()
+        || !callbackThreadsToAdd.isEmpty()
+        || !toExecuteInWorkflowThread.isEmpty();
+  }
+
   @SuppressWarnings("unchecked")
   <T> Optional<T> getRunnerLocal(RunnerLocalInternal<T> key) {
     if (!runnerLocalMap.containsKey(key)) {
@@ -576,4 +631,14 @@ private NamedRunnable(String name, Runnable runnable) {
       this.runnable = runnable;
     }
   }
+
+  private static class WorkflowThreadStopFuture {
+    private final WorkflowThread workflowThread;
+    private final Future<?> stopFuture;
+
+    public WorkflowThreadStopFuture(WorkflowThread workflowThread, Future<?> stopFuture) {
+      this.workflowThread = workflowThread;
+      this.stopFuture = stopFuture;
+    }
+  }
 }

temporal-test-server/build.gradle

@@ -1,7 +1,14 @@
+buildscript {
+    ext {
+        // 0.11.0 and later are build on JDK 11 bytecode version
+        graalVersion = "${JavaVersion.current().isCompatibleWith(JavaVersion.VERSION_11) ? '0.12.0' : '0.10.0'}"
+    }
+}
+
 plugins {
     id 'application'
-    id 'com.palantir.graal' version '0.10.0'
-    id 'com.palantir.docker' version '0.33.0'
+    id 'com.palantir.graal' version "${graalVersion}"
+    id 'com.palantir.docker' version '0.34.0'
     id 'com.google.protobuf' version '0.8.18'
 }