DeterministicRunnerImpl::close: wait for tasks outside lock (#213)

* DeterministicRunnerImpl: wait for tasks outside lock

* return completable future on null

Author: halakaraki

src/main/java/com/uber/cadence/internal/sync/DeterministicRunnerImpl.java

@@ -287,6 +287,7 @@ public void cancel(String reason) {
 
   @Override
   public void close() {
+    List<Future<?>> threadFutures = new ArrayList<>();
     lock.lock();
     if (closed) {
       return;
@@ -299,7 +300,7 @@ public void close() {
     }
     try {
       for (WorkflowThread c : threads) {
-        c.stop();
+        threadFutures.add(c.stopNow());
       }
       threads.clear();
 
@@ -326,6 +327,18 @@ public void close() {
       closed = true;
       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) {
+      try {
+        future.get();
+      } catch (InterruptedException e) {
+        throw new Error("Unexpected interrupt", e);
+      } catch (ExecutionException e) {
+        throw new Error("Unexpected failure stopping coroutine", e);
+      }
+    }
   }
 
   @Override

src/main/java/com/uber/cadence/internal/sync/WorkflowThread.java

@@ -22,6 +22,7 @@
 import com.uber.cadence.workflow.CancellationScope;
 import java.util.Optional;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.Future;
 import java.util.function.Supplier;
 
 /** Thread that is scheduled deterministically by {@link DeterministicRunner}. */
@@ -89,7 +90,7 @@ static WorkflowThread newThread(Runnable runnable, boolean detached, String name
 
   boolean isDone();
 
-  void stop();
+  Future<?> stopNow();
 
   void addStackTrace(StringBuilder result);
 

src/main/java/com/uber/cadence/internal/sync/WorkflowThreadImpl.java

@@ -27,17 +27,15 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.CancellationException;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.*;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
+import static javafx.scene.input.KeyCode.T;
+
 class WorkflowThreadImpl implements WorkflowThread {
 
   /**
@@ -331,10 +329,10 @@ public void evaluateInCoroutineContext(Consumer<String> function) {
 
   /**
    * Interrupt coroutine by throwing DestroyWorkflowThreadError from a await method it is blocked on
-   * and wait for coroutine thread to finish execution.
+   * and return underlying Future to be waited on.
    */
   @Override
-  public void stop() {
+  public Future<?> stopNow() {
     // Cannot call destroy() on itself
     if (thread == Thread.currentThread()) {
       throw new Error("Cannot call destroy on itself: " + thread.getName());
@@ -344,16 +342,16 @@ public void stop() {
       throw new RuntimeException(
           "Couldn't destroy the thread. " + "The blocked thread stack trace: " + getStackTrace());
     }
-    try {
-      // Check if thread was started
-      if (taskFuture != null) {
-        taskFuture.get();
-      }
-    } catch (InterruptedException e) {
-      throw new Error("Unexpected interrupt", e);
-    } catch (ExecutionException e) {
-      throw new Error("Unexpected failure stopping coroutine", e);
+    if(taskFuture == null){
+     return getCompletedFuture();
     }
+    return taskFuture;
+  }
+
+  private Future<?> getCompletedFuture(){
+    CompletableFuture<String> f = new CompletableFuture<>();
+    f.complete("done");
+    return f;
   }
 
   @Override