Fix evict next (#214)

Author: halakaraki

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

@@ -27,16 +27,19 @@
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.m3.tally.Scope;
 import java.util.Objects;
+import java.util.Random;
 import java.util.UUID;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 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 static final Logger log = LoggerFactory.getLogger(DeciderCache.class);
+  private Lock evictionLock = new ReentrantLock();
+  Random rand = new Random();
 
   public DeciderCache(int maxCacheSize, Scope scope) {
     Preconditions.checkArgument(maxCacheSize > 0, "Max cache size must be greater than 0");
@@ -45,6 +48,7 @@ public DeciderCache(int maxCacheSize, Scope scope) {
     this.cache =
         CacheBuilder.newBuilder()
             .maximumWeight(maxCacheSize)
+            .concurrencyLevel(1)
             .weigher(
                 (Weigher<String, WeightedCacheEntry<Decider>>) (key, value) -> value.getWeight())
             .removalListener(
@@ -70,7 +74,7 @@ public Decider getOrCreate(
     String runId = decisionTask.getWorkflowExecution().getRunId();
     metricsScope.gauge(MetricsType.STICKY_CACHE_SIZE).update(size());
     if (isFullHistory(decisionTask)) {
-      cache.invalidate(runId);
+      invalidate(decisionTask);
       return cache.get(
               runId, () -> new WeightedCacheEntry<>(createReplayDecider.apply(decisionTask), 1))
           .entry;
@@ -89,22 +93,43 @@ public Decider getUnchecked(String runId) throws Exception {
     }
   }
 
-  public void evictNext() {
-    metricsScope.gauge(MetricsType.STICKY_CACHE_SIZE).update(size());
-    int remainingSpace = (int) (maxCacheSize - cache.size());
-    // Force eviction to happen
-    cache.put(evictionEntryId, new WeightedCacheEntry<>(null, remainingSpace + 1));
-    invalidate(evictionEntryId);
+  public void evictNext() 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);
+      metricsScope.counter(MetricsType.STICKY_CACHE_THREAD_FORCED_EVICTION).inc(1);
+    } finally {
+      evictionLock.unlock();
+    }
   }
 
-  public void invalidate(PollForDecisionTaskResponse decisionTask) {
+  public void invalidate(PollForDecisionTaskResponse decisionTask) throws InterruptedException {
     String runId = decisionTask.getWorkflowExecution().getRunId();
     invalidate(runId);
   }
 
-  public void invalidate(String runId) {
-    metricsScope.counter(MetricsType.STICKY_CACHE_TOTAL_FORCED_EVICTION).inc(1);
-    cache.invalidate(runId);
+  private void invalidate(String runId) throws InterruptedException {
+    if (!evictionLock.tryLock(rand.nextInt(4), TimeUnit.SECONDS)) {
+      return;
+    }
+    try {
+      cache.invalidate(runId);
+      metricsScope.counter(MetricsType.STICKY_CACHE_TOTAL_FORCED_EVICTION).inc(1);
+    } finally {
+      evictionLock.unlock();
+    }
   }
 
   public long size() {

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

@@ -236,7 +236,13 @@ public void start() {
           .getMetricsScope()
           .counter(MetricsType.STICKY_CACHE_THREAD_FORCED_EVICTION)
           .inc(1);
-      cache.evictNext();
+      try {
+        if (cache != null) {
+          cache.evictNext();
+        }
+      } catch (InterruptedException e1) {
+        log.warn("Unable to evict cache", e1);
+      }
     }
 
     try {

src/test/java/com/uber/cadence/internal/replay/ReplayDeciderCacheTests.java

@@ -20,9 +20,9 @@
 import static junit.framework.TestCase.assertEquals;
 import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.fail;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
+import static org.mockito.Matchers.anyInt;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.*;
 
 import com.uber.cadence.HistoryEvent;
 import com.uber.cadence.PollForDecisionTaskResponse;
@@ -160,10 +160,10 @@ public void evictNextWillInvalidateTheNextEntryInLineToBeEvicted() throws Except
             .build();
     StatsReporter reporter = mock(StatsReporter.class);
     Scope scope =
-        new RootScopeBuilder().reporter(reporter).reportEvery(Duration.ofMillis(10)).tagged(tags);
+        new RootScopeBuilder().reporter(reporter).reportEvery(Duration.ofMillis(100)).tagged(tags);
 
     // Arrange
-    DeciderCache replayDeciderCache = new DeciderCache(10, scope);
+    DeciderCache replayDeciderCache = new DeciderCache(50, scope);
     PollForDecisionTaskResponse decisionTask1 =
         HistoryUtils.generateDecisionTaskWithInitialHistory();
     PollForDecisionTaskResponse decisionTask2 =
@@ -187,8 +187,9 @@ public void evictNextWillInvalidateTheNextEntryInLineToBeEvicted() throws Except
 
     // Wait for reporter
     Thread.sleep(600);
-    verify(reporter, times(1))
-        .reportCounter(MetricsType.STICKY_CACHE_TOTAL_FORCED_EVICTION, tags, 1);
+    verify(reporter, atLeastOnce())
+        .reportCounter(
+                eq(MetricsType.STICKY_CACHE_TOTAL_FORCED_EVICTION), eq(tags), anyInt());
   }
 
   private void assertCacheIsEmpty(DeciderCache cache, String runId) throws Exception {

src/test/java/com/uber/cadence/internal/sync/DeterministicRunnerTest.java

@@ -22,10 +22,7 @@
 import static junit.framework.TestCase.assertTrue;
 import static junit.framework.TestCase.fail;
 import static org.junit.Assert.assertFalse;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
 
 import com.uber.cadence.Decision;
 import com.uber.cadence.PollForDecisionTaskResponse;
@@ -718,8 +715,8 @@ public void workflowThreadsWillEvictCacheWhenMaxThreadCountIsHit() throws Throwa
     assertEquals(0, cache.size()); // cache was evicted
     // Wait for reporter
     Thread.sleep(600);
-    verify(reporter, times(1))
-        .reportCounter(MetricsType.STICKY_CACHE_THREAD_FORCED_EVICTION, tags, 1);
+    verify(reporter, atLeastOnce())
+        .reportCounter(eq(MetricsType.STICKY_CACHE_THREAD_FORCED_EVICTION), eq(tags), anyInt());
   }
 
   @Test