Handle Self Eviction - enable sticky by default (#219)

Author: halakaraki

build.gradle

@@ -37,7 +37,7 @@ googleJavaFormat {
 }
 
 group = 'com.uber.cadence'
-version = '2.0.0'
+version = '2.1.0'
 
 description = """Uber Cadence Java Client"""
 

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

@@ -91,7 +91,7 @@ public AsyncBackoffThrottler(
 
   private long calculateSleepTime() {
     double sleepMillis =
-        (Math.pow(backoffCoefficient, failureCount.get() - 1)) * initialSleep.toMillis();
+        Math.pow(backoffCoefficient, failureCount.get() - 1) * initialSleep.toMillis();
     if (maxSleep != null) {
       return Math.min((long) sleepMillis, maxSleep.toMillis());
     }

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

@@ -78,7 +78,7 @@ public BackoffThrottler(Duration initialSleep, Duration maxSleep, double backoff
 
   private long calculateSleepTime() {
     double sleepMillis =
-        (Math.pow(backoffCoefficient, failureCount.get() - 1)) * initialSleep.toMillis();
+        Math.pow(backoffCoefficient, failureCount.get() - 1) * initialSleep.toMillis();
     if (maxSleep != null) {
       return Math.min((long) sleepMillis, maxSleep.toMillis());
     }

src/main/java/com/uber/cadence/internal/metrics/MetricsType.java

@@ -140,4 +140,6 @@ public class MetricsType {
       CADENCE_METRICS_PREFIX + "sticky-cache-thread-forced-eviction";
   public static final String STICKY_CACHE_STALL = CADENCE_METRICS_PREFIX + "sticky-cache-stall";
   public static final String STICKY_CACHE_SIZE = CADENCE_METRICS_PREFIX + "sticky-cache-size";
+  public static final String WORKFLOW_ACTIVE_THREAD_COUNT =
+      CADENCE_METRICS_PREFIX + "workflow_active_thread_count";
 }

src/main/java/com/uber/cadence/internal/replay/Decider.java

@@ -29,4 +29,4 @@ public interface Decider {
   byte[] query(PollForDecisionTaskResponse decisionTask, WorkflowQuery query) throws Throwable;
 
   void close();
-}
+}

src/main/java/com/uber/cadence/internal/replay/DeciderCache.java

@@ -21,47 +21,45 @@
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
-import com.google.common.cache.Weigher;
 import com.uber.cadence.PollForDecisionTaskResponse;
 import com.uber.cadence.internal.common.ThrowableFunc1;
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.m3.tally.Scope;
+import java.util.Iterator;
 import java.util.Objects;
 import java.util.Random;
-import java.util.UUID;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public final class DeciderCache {
-  private final String evictionEntryId = UUID.randomUUID().toString();
-  private final int maxCacheSize;
   private final Scope metricsScope;
-  private LoadingCache<String, WeightedCacheEntry<Decider>> cache;
+  private LoadingCache<String, Decider> cache;
   private Lock evictionLock = new ReentrantLock();
   Random rand = new Random();
 
+  private static final Logger log = LoggerFactory.getLogger(DeciderCache.class);
+
   public DeciderCache(int maxCacheSize, Scope scope) {
     Preconditions.checkArgument(maxCacheSize > 0, "Max cache size must be greater than 0");
-    this.maxCacheSize = maxCacheSize;
     this.metricsScope = Objects.requireNonNull(scope);
     this.cache =
         CacheBuilder.newBuilder()
-            .maximumWeight(maxCacheSize)
-            .concurrencyLevel(1)
-            .weigher(
-                (Weigher<String, WeightedCacheEntry<Decider>>) (key, value) -> value.getWeight())
+            .maximumSize(maxCacheSize)
             .removalListener(
                 e -> {
-                  Decider entry = e.getValue().entry;
+                  Decider entry = (Decider) e.getValue();
                   if (entry != null) {
                     entry.close();
                   }
                 })
             .build(
-                new CacheLoader<String, WeightedCacheEntry<Decider>>() {
+                new CacheLoader<String, Decider>() {
                   @Override
-                  public WeightedCacheEntry<Decider> load(String key) {
+                  public Decider load(String key) {
                     return null;
                   }
                 });
@@ -75,16 +73,14 @@ public Decider getOrCreate(
     metricsScope.gauge(MetricsType.STICKY_CACHE_SIZE).update(size());
     if (isFullHistory(decisionTask)) {
       invalidate(decisionTask);
-      return cache.get(
-              runId, () -> new WeightedCacheEntry<>(createReplayDecider.apply(decisionTask), 1))
-          .entry;
+      return cache.get(runId, () -> createReplayDecider.apply(decisionTask));
     }
     return getUnchecked(runId);
   }
 
   public Decider getUnchecked(String runId) throws Exception {
     try {
-      Decider cachedDecider = cache.getUnchecked(runId).entry;
+      Decider cachedDecider = cache.getUnchecked(runId);
       metricsScope.counter(MetricsType.STICKY_CACHE_HIT).inc(1);
       return cachedDecider;
     } catch (CacheLoader.InvalidCacheLoadException e) {
@@ -93,22 +89,32 @@ public Decider getUnchecked(String runId) throws Exception {
     }
   }
 
-  public void evictNext() throws InterruptedException {
+  public void evictAny(String runId) throws InterruptedException {
     // Timeout is to guard against workflows trying to evict each other.
     if (!evictionLock.tryLock(rand.nextInt(4), TimeUnit.SECONDS)) {
       return;
     }
     try {
       metricsScope.gauge(MetricsType.STICKY_CACHE_SIZE).update(size());
-      int remainingSpace = (int) (maxCacheSize - cache.size());
-      // Force eviction to happen. This assumes a concurrency level of 1 which implies a single
-      // underlying segment and lock. If higher concurrency levels are assumed this may not work
-      // since
-      // the weight could be greater than the segment size and put will simply noop.
-      // ConcurrenyLevel limits cache modification but reads and cache loading computations still
-      // have concurrently.
-      cache.put(evictionEntryId, new WeightedCacheEntry<>(null, remainingSpace + 1));
-      invalidate(evictionEntryId);
+      Set<String> set = cache.asMap().keySet();
+      if (set.isEmpty()) {
+        return;
+      }
+      Iterator<String> iter = cache.asMap().keySet().iterator();
+      String key = "";
+      while (iter.hasNext()) {
+        key = iter.next();
+        if (!key.equals(runId)) {
+          break;
+        }
+      }
+
+      if (key.equals(runId)) {
+        log.warn(String.format("%s attempted to self evict. Ignoring eviction", runId));
+        return;
+      }
+      cache.invalidate(key);
+      metricsScope.gauge(MetricsType.STICKY_CACHE_SIZE).update(size());
       metricsScope.counter(MetricsType.STICKY_CACHE_THREAD_FORCED_EVICTION).inc(1);
     } finally {
       evictionLock.unlock();
@@ -146,25 +152,6 @@ public void invalidateAll() {
     cache.invalidateAll();
   }
 
-  // Used for eviction
-  private static class WeightedCacheEntry<T> {
-    private T entry;
-    private int weight;
-
-    private WeightedCacheEntry(T entry, int weight) {
-      this.entry = entry;
-      this.weight = weight;
-    }
-
-    public T getEntry() {
-      return entry;
-    }
-
-    public int getWeight() {
-      return weight;
-    }
-  }
-
   public static class EvictedException extends Exception {
 
     public EvictedException(String runId) {

src/main/java/com/uber/cadence/internal/replay/DecisionId.java

@@ -43,7 +43,7 @@ public boolean equals(Object o) {
     if (this == o) {
       return true;
     }
-    if (o == null || getClass() != o.getClass()) {
+    if (o == null || !(o instanceof DecisionId)) {
       return false;
     }
 

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

@@ -22,6 +22,7 @@
 import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.converter.JsonDataConverter;
 import com.uber.cadence.internal.common.CheckedExceptionWrapper;
+import com.uber.cadence.internal.metrics.NoopScope;
 import com.uber.cadence.internal.replay.ContinueAsNewWorkflowExecutionParameters;
 import com.uber.cadence.internal.replay.DeciderCache;
 import com.uber.cadence.internal.replay.DecisionContext;
@@ -290,12 +291,14 @@ public void close() {
     List<Future<?>> threadFutures = new ArrayList<>();
     lock.lock();
     if (closed) {
+      lock.unlock();
       return;
     }
     // Do not close while runUntilAllBlocked executes.
     // closeRequested tells it to call close() at the end.
     closeRequested = true;
     if (inRunUntilAllBlocked) {
+      lock.unlock();
       return;
     }
     try {
@@ -343,10 +346,10 @@ public void close() {
 
   @Override
   public String stackTrace() {
-    lock.lock();
-    checkClosed();
     StringBuilder result = new StringBuilder();
+    lock.lock();
     try {
+      checkClosed();
       for (WorkflowThread coroutine : threads) {
         if (result.length() > 0) {
           result.append("\n");
@@ -415,8 +418,8 @@ WorkflowThread newThread(Runnable runnable, boolean detached, String name) {
   @Override
   public void executeInWorkflowThread(String name, Runnable runnable) {
     lock.lock();
-    checkClosed();
     try {
+      checkClosed();
       toExecuteInWorkflowThread.add(new NamedRunnable(name, runnable));
     } finally {
       lock.unlock();
@@ -594,7 +597,7 @@ public Random newRandom() {
 
     @Override
     public Scope getMetricsScope() {
-      throw new UnsupportedOperationException("not implemented");
+      return NoopScope.getInstance();
     }
 
     @Override

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

@@ -164,7 +164,7 @@ private static long calculateSleepTime(long attempt, RetryOptions options) {
             ? DEFAULT_COEFFICIENT
             : options.getBackoffCoefficient();
     double sleepMillis =
-        (Math.pow(backoffCoefficient, attempt - 1)) * options.getInitialInterval().toMillis();
+        Math.pow(backoffCoefficient, attempt - 1) * options.getInitialInterval().toMillis();
     Duration maximumInterval = options.getMaximumInterval();
     if (maximumInterval == null) {
       return (long)

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

@@ -34,8 +34,6 @@
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
-import static javafx.scene.input.KeyCode.T;
-
 class WorkflowThreadImpl implements WorkflowThread {
 
   /**
@@ -228,6 +226,11 @@ public void start() {
     }
     context.setStatus(Status.RUNNING);
 
+    getDecisionContext()
+        .getMetricsScope()
+        .gauge(MetricsType.STICKY_CACHE_SIZE)
+        .update(((ThreadPoolExecutor) threadPool).getActiveCount());
+
     try {
       taskFuture = threadPool.submit(task);
       return;
@@ -238,7 +241,7 @@ public void start() {
           .inc(1);
       try {
         if (cache != null) {
-          cache.evictNext();
+          cache.evictAny(this.runner.getDecisionContext().getContext().getRunId());
         }
       } catch (InterruptedException e1) {
         log.warn("Unable to evict cache", e1);
@@ -348,13 +351,13 @@ public Future<?> stopNow() {
       throw new RuntimeException(
           "Couldn't destroy the thread. " + "The blocked thread stack trace: " + getStackTrace());
     }
-    if(taskFuture == null){
-     return getCompletedFuture();
+    if (taskFuture == null) {
+      return getCompletedFuture();
     }
     return taskFuture;
   }
 
-  private Future<?> getCompletedFuture(){
+  private Future<?> getCompletedFuture() {
     CompletableFuture<String> f = new CompletableFuture<>();
     f.complete("done");
     return f;

src/main/java/com/uber/cadence/internal/testservice/StateMachine.java

@@ -69,7 +69,7 @@ public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+      if (o == null || !(o instanceof Transition)) {
         return false;
       }
 

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowMutableStateImpl.java

@@ -697,7 +697,7 @@ private void processStartTimer(
       throws BadRequestError, InternalServiceError {
     String timerId = a.getTimerId();
     if (timerId == null) {
-      throw new BadRequestError(("A valid TimerId is not set on StartTimerDecision"));
+      throw new BadRequestError("A valid TimerId is not set on StartTimerDecision");
     }
     StateMachine<TimerData> timer = timers.get(timerId);
     if (timer != null) {

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowStore.java

@@ -61,7 +61,8 @@ public boolean equals(Object o) {
       if (this == o) {
         return true;
       }
-      if (o == null || getClass() != o.getClass()) {
+
+      if (o == null || !(o instanceof TaskListId)) {
         return false;
       }
 

src/main/java/com/uber/cadence/internal/testservice/WorkflowId.java

@@ -42,7 +42,7 @@ public boolean equals(Object o) {
     if (this == o) {
       return true;
     }
-    if (o == null || getClass() != o.getClass()) {
+    if (o == null || !(o instanceof WorkflowId)) {
       return false;
     }
 

src/main/java/com/uber/cadence/internal/worker/ExecutorThreadFactory.java

@@ -35,7 +35,7 @@ class ExecutorThreadFactory implements ThreadFactory {
   @Override
   public Thread newThread(Runnable r) {
     Thread result = new Thread(r);
-    result.setName(threadPrefix + ": " + (threadIndex.incrementAndGet()));
+    result.setName(threadPrefix + ": " + threadIndex.incrementAndGet());
     result.setUncaughtExceptionHandler(uncaughtExceptionHandler);
     return result;
   }

src/main/java/com/uber/cadence/internal/worker/PollDecisionTaskDispatcherFactory.java

@@ -24,7 +24,7 @@
 
 public final class PollDecisionTaskDispatcherFactory
     implements DispatcherFactory<String, PollForDecisionTaskResponse> {
-  private IWorkflowService service = new WorkflowServiceTChannel();
+  private IWorkflowService service;
   private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
 
   public PollDecisionTaskDispatcherFactory(
@@ -37,8 +37,6 @@ public PollDecisionTaskDispatcherFactory(IWorkflowService service) {
     this.service = Objects.requireNonNull(service);
   }
 
-  public PollDecisionTaskDispatcherFactory() {}
-
   @Override
   public Dispatcher<String, PollForDecisionTaskResponse> create() {
     return new PollDecisionTaskDispatcher(service, uncaughtExceptionHandler);

src/main/java/com/uber/cadence/testing/TestEnvironmentOptions.java

@@ -92,6 +92,10 @@ public TestEnvironmentOptions build() {
         metricsScope = NoopScope.getInstance();
       }
 
+      if (factoryOptions == null) {
+        factoryOptions = new Worker.FactoryOptions.Builder().Build();
+      }
+
       return new TestEnvironmentOptions(
           dataConverter,
           domain,