3. Thread based eviction (#203)

Author: halakaraki

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

@@ -17,6 +17,7 @@
 
 package com.uber.cadence.internal.sync;
 
+import com.uber.cadence.internal.replay.DeciderCache;
 import com.uber.cadence.workflow.CancellationScope;
 import com.uber.cadence.workflow.Workflow;
 import java.util.concurrent.ExecutorService;
@@ -37,6 +38,25 @@ static DeterministicRunner newRunner(Supplier<Long> clock, Runnable root) {
     return new DeterministicRunnerImpl(clock, root);
   }
 
+  /**
+   * Create new instance of DeterministicRunner
+   *
+   * @param decisionContext decision context to use
+   * @param clock Supplier that returns current time that sync should use
+   * @param root function that root thread of the runner executes.
+   * @param cache DeciderCache used cache inflight workflows. New workflow threads will evict this
+   *     cache when the thread pool runs out
+   * @return instance of the DeterministicRunner.
+   */
+  static DeterministicRunner newRunner(
+      ExecutorService threadPool,
+      SyncDecisionContext decisionContext,
+      Supplier<Long> clock,
+      Runnable root,
+      DeciderCache cache) {
+    return new DeterministicRunnerImpl(threadPool, decisionContext, clock, root, cache);
+  }
+
   /**
    * Create new instance of DeterministicRunner
    *
@@ -50,7 +70,7 @@ static DeterministicRunner newRunner(
       SyncDecisionContext decisionContext,
       Supplier<Long> clock,
       Runnable root) {
-    return new DeterministicRunnerImpl(threadPool, decisionContext, clock, root);
+    return new DeterministicRunnerImpl(threadPool, decisionContext, clock, root, null);
   }
 
   /**

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

@@ -23,6 +23,7 @@
 import com.uber.cadence.converter.JsonDataConverter;
 import com.uber.cadence.internal.common.CheckedExceptionWrapper;
 import com.uber.cadence.internal.replay.ContinueAsNewWorkflowExecutionParameters;
+import com.uber.cadence.internal.replay.DeciderCache;
 import com.uber.cadence.internal.replay.DecisionContext;
 import com.uber.cadence.internal.replay.ExecuteActivityParameters;
 import com.uber.cadence.internal.replay.SignalExternalWorkflowParameters;
@@ -81,6 +82,7 @@ private NamedRunnable(String name, Runnable runnable) {
   private final List<WorkflowThread> threadsToAdd = Collections.synchronizedList(new ArrayList<>());
   private final List<NamedRunnable> toExecuteInWorkflowThread = new ArrayList<>();
   private final Supplier<Long> clock;
+  private DeciderCache cache;
   private boolean inRunUntilAllBlocked;
   private boolean closeRequested;
   private boolean closed;
@@ -118,7 +120,7 @@ static void setCurrentThreadInternal(WorkflowThread coroutine) {
   }
 
   DeterministicRunnerImpl(Supplier<Long> clock, Runnable root) {
-    this(getDefaultThreadPool(), newDummySyncDecisionContext(), clock, root);
+    this(getDefaultThreadPool(), newDummySyncDecisionContext(), clock, root, null);
   }
 
   private static ThreadPoolExecutor getDefaultThreadPool() {
@@ -139,10 +141,20 @@ public Thread newThread(Runnable r) {
       SyncDecisionContext decisionContext,
       Supplier<Long> clock,
       Runnable root) {
+    this(threadPool, decisionContext, clock, root, null);
+  }
+
+  DeterministicRunnerImpl(
+      ExecutorService threadPool,
+      SyncDecisionContext decisionContext,
+      Supplier<Long> clock,
+      Runnable root,
+      DeciderCache cache) {
     this.threadPool = threadPool;
     this.decisionContext =
         decisionContext != null ? decisionContext : newDummySyncDecisionContext();
     this.clock = clock;
+    this.cache = cache;
     runnerCancellationScope = new CancellationScopeImpl(true, null, null);
     // TODO: workflow instance specific thread name
     rootWorkflowThread =
@@ -153,7 +165,8 @@ public Thread newThread(Runnable r) {
             WORKFLOW_ROOT_THREAD_NAME,
             false,
             runnerCancellationScope,
-            root);
+            root,
+            cache);
     threads.addLast(rootWorkflowThread);
     rootWorkflowThread.start();
   }
@@ -183,7 +196,14 @@ public void runUntilAllBlocked() throws Throwable {
         for (NamedRunnable nr : toExecuteInWorkflowThread) {
           WorkflowThread thread =
               new WorkflowThreadImpl(
-                  false, threadPool, this, nr.name, false, runnerCancellationScope, nr.runnable);
+                  false,
+                  threadPool,
+                  this,
+                  nr.name,
+                  false,
+                  runnerCancellationScope,
+                  nr.runnable,
+                  cache);
           // 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
@@ -363,7 +383,14 @@ WorkflowThread newThread(Runnable runnable, boolean detached, String name) {
     checkClosed();
     WorkflowThread result =
         new WorkflowThreadImpl(
-            false, threadPool, this, name, detached, CancellationScopeImpl.current(), runnable);
+            false,
+            threadPool,
+            this,
+            name,
+            detached,
+            CancellationScopeImpl.current(),
+            runnable,
+            cache);
     threadsToAdd.add(result); // This is synchronized collection.
     return result;
   }

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

@@ -24,6 +24,7 @@
 import com.uber.cadence.internal.common.CheckedExceptionWrapper;
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.metrics.MetricsType;
+import com.uber.cadence.internal.replay.DeciderCache;
 import com.uber.cadence.internal.replay.ReplayWorkflow;
 import com.uber.cadence.internal.replay.ReplayWorkflowFactory;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
@@ -67,16 +68,19 @@ final class POJOWorkflowImplementationFactory implements ReplayWorkflowFactory {
 
   private final ExecutorService threadPool;
   private final Scope metricsScope;
+  private DeciderCache cache;
 
   POJOWorkflowImplementationFactory(
       DataConverter dataConverter,
       ExecutorService threadPool,
       Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
-      Scope metricsScope) {
+      Scope metricsScope,
+      DeciderCache cache) {
     this.dataConverter = Objects.requireNonNull(dataConverter);
     this.threadPool = Objects.requireNonNull(threadPool);
     this.interceptorFactory = Objects.requireNonNull(interceptorFactory);
     this.metricsScope = metricsScope;
+    this.cache = cache;
   }
 
   void setWorkflowImplementationTypes(Class<?>[] workflowImplementationTypes) {
@@ -183,7 +187,7 @@ public void setDataConverter(DataConverter dataConverter) {
   @Override
   public ReplayWorkflow getWorkflow(WorkflowType workflowType) {
     SyncWorkflowDefinition workflow = getWorkflowDefinition(workflowType);
-    return new SyncWorkflow(workflow, dataConverter, threadPool, interceptorFactory);
+    return new SyncWorkflow(workflow, dataConverter, threadPool, interceptorFactory, cache);
   }
 
   @Override

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

@@ -23,6 +23,7 @@
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.client.WorkflowClient;
 import com.uber.cadence.converter.DataConverter;
+import com.uber.cadence.internal.replay.DeciderCache;
 import com.uber.cadence.internal.replay.DecisionContext;
 import com.uber.cadence.internal.replay.ReplayWorkflow;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
@@ -41,18 +42,21 @@ class SyncWorkflow implements ReplayWorkflow {
   private final ExecutorService threadPool;
   private final SyncWorkflowDefinition workflow;
   private final Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory;
+  private DeciderCache cache;
   private WorkflowRunnable workflowProc;
   private DeterministicRunner runner;
 
   public SyncWorkflow(
       SyncWorkflowDefinition workflow,
       DataConverter dataConverter,
       ExecutorService threadPool,
-      Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory) {
+      Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
+      DeciderCache cache) {
     this.workflow = Objects.requireNonNull(workflow);
     this.dataConverter = Objects.requireNonNull(dataConverter);
     this.threadPool = Objects.requireNonNull(threadPool);
     this.interceptorFactory = Objects.requireNonNull(interceptorFactory);
+    this.cache = cache;
   }
 
   @Override
@@ -74,7 +78,7 @@ public void start(HistoryEvent event, DecisionContext context) {
             syncContext, workflow, event.getWorkflowExecutionStartedEventAttributes());
     runner =
         DeterministicRunner.newRunner(
-            threadPool, syncContext, context::currentTimeMillis, workflowProc);
+            threadPool, syncContext, context::currentTimeMillis, workflowProc, cache);
     syncContext.setRunner(runner);
   }
 

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

@@ -60,7 +60,9 @@ public SyncWorkflowWorker(
             options.getDataConverter(),
             workflowThreadPool,
             interceptorFactory,
-            options.getMetricsScope());
+            options.getMetricsScope(),
+            cache);
+
     DecisionTaskHandler taskHandler =
         new ReplayDecisionTaskHandler(
             domain,

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

@@ -18,6 +18,7 @@
 package com.uber.cadence.internal.sync;
 
 import com.uber.cadence.internal.logging.LoggerTag;
+import com.uber.cadence.internal.replay.DeciderCache;
 import com.uber.cadence.internal.replay.DecisionContext;
 import com.uber.cadence.workflow.Promise;
 import java.io.PrintWriter;
@@ -148,6 +149,7 @@ public void setName(String name) {
   private final boolean root;
   private final ExecutorService threadPool;
   private final WorkflowThreadContext context;
+  private DeciderCache cache;
   private final DeterministicRunnerImpl runner;
   private final RunnableWrapper task;
   private Thread thread;
@@ -169,12 +171,14 @@ public void setName(String name) {
       String name,
       boolean detached,
       CancellationScopeImpl parentCancellationScope,
-      Runnable runnable) {
+      Runnable runnable,
+      DeciderCache cache) {
     this.root = root;
     this.threadPool = threadPool;
     this.runner = runner;
     this.context = new WorkflowThreadContext(runner.getLock());
-    // TODO: Use thread pool instead of creating new threads.
+    this.cache = cache;
+
     if (name == null) {
       name = "workflow-" + super.hashCode();
     }
@@ -224,6 +228,14 @@ public void start() {
       throw new IllegalThreadStateException("already started");
     }
     context.setStatus(Status.RUNNING);
+
+    try {
+      taskFuture = threadPool.submit(task);
+      return;
+    } catch (RejectedExecutionException e) {
+      cache.evictNext();
+    }
+
     try {
       taskFuture = threadPool.submit(task);
     } catch (RejectedExecutionException e) {

src/main/java/com/uber/cadence/worker/Worker.java

@@ -342,7 +342,6 @@ public static final class Factory {
     private final String domain;
     private final UUID id =
         UUID.randomUUID(); // Guarantee uniqueness for stickyTaskListName when multiple factories
-    // are created.
     private final ThreadPoolExecutor workflowThreadPool;
     private final AtomicInteger workflowThreadCounter = new AtomicInteger();
     private final FactoryOptions factoryOptions;