Fixed thread exhaustion with large number of parallel async activities (#234)

* Moved thread start to runUntilBlocked to delay grabbing thread from the pool.

* Added testLargeHistory unit test

* Fixed Retryer when maxAttempts or expiration is not specified

Author: mfateev

src/main/java/com/uber/cadence/internal/common/Retryer.java

@@ -20,6 +20,7 @@
 import static com.uber.cadence.internal.common.CheckedExceptionWrapper.unwrap;
 
 import com.uber.cadence.common.RetryOptions;
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.function.BiFunction;
@@ -105,8 +106,10 @@ public static <R, T extends Throwable> R retryWithResult(
           }
         }
         long elapsed = System.currentTimeMillis() - startTime;
-        if (attempt >= options.getMaximumAttempts()
-            || elapsed >= options.getExpiration().toMillis()) {
+        int maxAttempts = options.getMaximumAttempts();
+        Duration expiration = options.getExpiration();
+        if ((maxAttempts > 0 && attempt >= maxAttempts)
+            || (expiration != null && elapsed >= expiration.toMillis())) {
           rethrow(e);
         }
         log.warn("Retrying after failure", e);

src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java

@@ -94,7 +94,7 @@ public class WorkflowExecutionUtils {
       new RetryOptions.Builder()
           .setBackoffCoefficient(2)
           .setInitialInterval(Duration.ofMillis(500))
-          .setMaximumInterval(Duration.ofSeconds(30))
+          .setMaximumInterval(Duration.ofSeconds(10))
           .setDoNotRetry(BadRequestError.class, EntityNotExistsError.class)
           .build();
 

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

@@ -211,7 +211,6 @@ public void runUntilAllBlocked() throws Throwable {
           // Otherwise signal might be never processed if it was received
           // after workflow decided to close.
           threads.addFirst(thread);
-          thread.start();
         }
         toExecuteInWorkflowThread.clear();
         progress = false;

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

@@ -49,8 +49,12 @@ class WorkflowThreadContext {
   }
 
   public void initialYield() {
-    if (getStatus() != Status.RUNNING) {
-      throw new IllegalStateException("not in RUNNING but in " + getStatus() + " state");
+    Status status = getStatus();
+    if (status == Status.DONE) {
+      throw new DestroyWorkflowThreadError("done in initialYield");
+    }
+    if (status != Status.RUNNING) {
+      throw new IllegalStateException("not in RUNNING but in " + status + " state");
     }
     yield("created", () -> true);
   }

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

@@ -300,8 +300,7 @@ private void setBlockedUntil(long blockedUntil) {
   @Override
   public boolean runUntilBlocked() {
     if (taskFuture == null) {
-      // Thread is not yet started
-      return false;
+      start();
     }
     return context.runUntilBlocked();
   }

src/test/java/com/uber/cadence/workflow/WorkflowTest.java

@@ -45,6 +45,7 @@
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.converter.JsonDataConverter;
 import com.uber.cadence.internal.sync.DeterministicRunnerTest;
+import com.uber.cadence.internal.worker.PollerOptions;
 import com.uber.cadence.serviceclient.WorkflowServiceTChannel;
 import com.uber.cadence.testing.TestEnvironmentOptions;
 import com.uber.cadence.testing.TestWorkflowEnvironment;
@@ -78,6 +79,7 @@
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TestName;
@@ -184,21 +186,21 @@ private static WorkflowOptions.Builder newWorkflowOptionsBuilder(String taskList
 
   private static ActivityOptions newActivityOptions1(String taskList) {
     if (DEBUGGER_TIMEOUTS) {
-      return new ActivityOptions.Builder()
-          .setTaskList(taskList)
-          .setScheduleToCloseTimeout(Duration.ofSeconds(5))
-          .setHeartbeatTimeout(Duration.ofSeconds(5))
-          .setScheduleToStartTimeout(Duration.ofSeconds(5))
-          .setStartToCloseTimeout(Duration.ofSeconds(10))
-          .build();
-    } else {
       return new ActivityOptions.Builder()
           .setTaskList(taskList)
           .setScheduleToCloseTimeout(Duration.ofSeconds(1000))
           .setHeartbeatTimeout(Duration.ofSeconds(1000))
           .setScheduleToStartTimeout(Duration.ofSeconds(1000))
           .setStartToCloseTimeout(Duration.ofSeconds(10000))
           .build();
+    } else {
+      return new ActivityOptions.Builder()
+          .setTaskList(taskList)
+          .setScheduleToCloseTimeout(Duration.ofSeconds(5))
+          .setHeartbeatTimeout(Duration.ofSeconds(5))
+          .setScheduleToStartTimeout(Duration.ofSeconds(5))
+          .setStartToCloseTimeout(Duration.ofSeconds(10))
+          .build();
     }
   }
 
@@ -223,7 +225,11 @@ public void setUp() {
               .build();
       workerFactory = new Worker.Factory(service, DOMAIN, factoryOptions);
       WorkerOptions workerOptions =
-          new WorkerOptions.Builder().setInterceptorFactory(tracer).build();
+          new WorkerOptions.Builder()
+              .setActivityPollerOptions(new PollerOptions.Builder().setPollThreadCount(5).build())
+              .setMaxConcurrentActivityExecutionSize(1000)
+              .setInterceptorFactory(tracer)
+              .build();
       worker = workerFactory.newWorker(taskList, workerOptions);
       workflowClient = WorkflowClient.newInstance(service, DOMAIN);
       WorkflowClientOptions clientOptions =
@@ -3645,6 +3651,56 @@ public void testGenericParametersWorkflow() throws ExecutionException, Interrupt
     assertEquals(expectedResult, result);
   }
 
+  public interface TestLargeWorkflow {
+
+    @WorkflowMethod
+    String execute(int activityCount, String taskList);
+  }
+
+  public interface TestLargeWorkflowActivity {
+    String activity();
+  }
+
+  public static class TestLargeWorkflowActivityImpl implements TestLargeWorkflowActivity {
+
+    @Override
+    public String activity() {
+      return "done";
+    }
+  }
+
+  public static class TestLargeHistory implements TestLargeWorkflow {
+
+    @Override
+    public String execute(int activityCount, String taskList) {
+      TestLargeWorkflowActivity activities =
+          Workflow.newActivityStub(TestLargeWorkflowActivity.class, newActivityOptions1(taskList));
+      List<Promise<String>> results = new ArrayList<>();
+      for (int i = 0; i < activityCount; i++) {
+        Promise<String> result = Async.function(activities::activity);
+        results.add(result);
+      }
+      Promise.allOf(results).get();
+      return "done";
+    }
+  }
+
+  @Test
+  @Ignore // Requires DEBUG_TIMEOUTS=true
+  public void testLargeHistory() {
+    final int activityCount = 1000;
+    worker.registerActivitiesImplementations(new TestLargeWorkflowActivityImpl());
+    startWorkerFor(TestLargeHistory.class);
+    TestLargeWorkflow workflowStub =
+        workflowClient.newWorkflowStub(
+            TestLargeWorkflow.class, newWorkflowOptionsBuilder(taskList).build());
+    long start = System.currentTimeMillis();
+    String result = workflowStub.execute(activityCount, taskList);
+    long duration = System.currentTimeMillis() - start;
+    log.info(testName.toString() + " duration is " + duration);
+    assertEquals("done", result);
+  }
+
   private static class FilteredTrace {
 
     private final List<String> impl = Collections.synchronizedList(new ArrayList<>());