add dispatcher and replay decider cache (#197)

* add dispatcher and replay decider cache

Author: halakaraki

build.gradle

@@ -49,7 +49,7 @@ dependencies {
     compile group: 'org.slf4j', name: 'slf4j-api', version: '1.7.25'
     compile group: 'org.apache.thrift', name: 'libthrift', version: '0.9.3'
     compile group: 'com.google.code.gson', name: 'gson', version: '2.8.2'
-    compile group: 'com.uber.m3', name: 'tally-core', version: '0.2.1'
+    compile group: 'com.uber.m3', name: 'tally-core', version: '0.2.3'
     compile group: 'com.google.guava', name: 'guava', version: '25.1-jre'
     testCompile group: 'junit', name: 'junit', version: '4.12'
     testCompile group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'

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

@@ -0,0 +1,22 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.internal.common;
+
+public interface ThrowableFunc1<T, R, E extends Throwable> {
+  R apply(T t) throws E;
+}

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

@@ -0,0 +1,32 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.internal.replay;
+
+import com.uber.cadence.WorkflowQuery;
+import com.uber.cadence.internal.worker.DecisionTaskWithHistoryIterator;
+
+public interface Decider {
+
+  // TODO: refactor in future CR. Merge methods and decide should return a list of decisions.
+  void decide(DecisionTaskWithHistoryIterator iterator) throws Throwable;
+
+  byte[] query(DecisionTaskWithHistoryIterator decisionTaskIterator, WorkflowQuery query)
+      throws Throwable;
+
+  void close();
+}

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

@@ -0,0 +1,133 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.internal.replay;
+
+import com.google.common.base.Preconditions;
+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 java.util.UUID;
+
+public final class DeciderCache {
+  private final String evictionEntryId = UUID.randomUUID().toString();
+  private final int maxCacheSize;
+  private LoadingCache<String, WeightedCacheEntry<Decider>> cache;
+
+  public DeciderCache(int maxCacheSize) {
+    Preconditions.checkArgument(maxCacheSize > 0, "Max cache size must be greater than 0");
+    this.maxCacheSize = maxCacheSize;
+    this.cache =
+        CacheBuilder.newBuilder()
+            .maximumWeight(maxCacheSize)
+            .weigher(
+                (Weigher<String, WeightedCacheEntry<Decider>>) (key, value) -> value.getWeight())
+            .removalListener(
+                e -> {
+                  Decider entry = e.getValue().entry;
+                  if (entry != null) {
+                    entry.close();
+                  }
+                })
+            .build(
+                new CacheLoader<String, WeightedCacheEntry<Decider>>() {
+                  @Override
+                  public WeightedCacheEntry<Decider> load(String key) {
+                    return null;
+                  }
+                });
+  }
+
+  public Decider getOrCreate(
+      PollForDecisionTaskResponse decisionTask,
+      ThrowableFunc1<PollForDecisionTaskResponse, Decider, Exception> createReplayDecider)
+      throws Exception {
+    String runId = decisionTask.getWorkflowExecution().getRunId();
+    if (isFullHistory(decisionTask)) {
+      cache.invalidate(runId);
+      return cache.get(
+              runId, () -> new WeightedCacheEntry<>(createReplayDecider.apply(decisionTask), 1))
+          .entry;
+    }
+    return getUnchecked(runId);
+  }
+
+  public Decider getUnchecked(String runId) throws Exception {
+    try {
+      return cache.getUnchecked(runId).entry;
+    } catch (CacheLoader.InvalidCacheLoadException e) {
+      throw new EvictedException(runId);
+    }
+  }
+
+  public void evictNext() {
+    int remainingSpace = (int) (maxCacheSize - cache.size());
+    // Force eviction to happen
+    cache.put(evictionEntryId, new WeightedCacheEntry<>(null, remainingSpace + 1));
+    cache.invalidate(evictionEntryId);
+  }
+
+  public void invalidate(PollForDecisionTaskResponse decisionTask) {
+    String runId = decisionTask.getWorkflowExecution().getRunId();
+    invalidate(runId);
+  }
+
+  public void invalidate(String runId) {
+    cache.invalidate(runId);
+  }
+
+  public long size() {
+    return cache.size();
+  }
+
+  private boolean isFullHistory(PollForDecisionTaskResponse decisionTask) {
+    return decisionTask.history.events.get(0).getEventId() == 1;
+  }
+
+  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) {
+      super(String.format("cache was evicted for the decisionTask. RunId: %s", runId));
+    }
+  }
+}

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

@@ -28,6 +28,7 @@
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.internal.replay.HistoryHelper.DecisionEvents;
 import com.uber.cadence.internal.replay.HistoryHelper.DecisionEventsIterator;
+import com.uber.cadence.internal.worker.DecisionTaskWithHistoryIterator;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
 import com.uber.cadence.workflow.Functions;
 import com.uber.m3.tally.Scope;
@@ -43,7 +44,7 @@
  * Implements decider that relies on replay of a worklfow code. An instance of this class is created
  * per decision.
  */
-class ReplayDecider {
+class ReplayDecider implements Decider {
 
   private static final Logger log = LoggerFactory.getLogger(ReplayDecider.class);
 
@@ -92,10 +93,6 @@ class ReplayDecider {
     context.setMetricsScope(metricsScope);
   }
 
-  public boolean isCancelRequested() {
-    return cancelRequested;
-  }
-
   private void handleWorkflowExecutionStarted(HistoryEvent event) throws Exception {
     workflow.start(event, context);
   }
@@ -351,15 +348,33 @@ private void handleDecisionTaskCompleted(HistoryEvent event) {
     decisionsHelper.handleDecisionCompletion(event.getDecisionTaskCompletedEventAttributes());
   }
 
-  void decide(HistoryHelper historyHelper) throws Throwable {
-    decideImpl(historyHelper, null);
+  @Override
+  public void decide(DecisionTaskWithHistoryIterator decisionTaskWithHistoryIterator)
+      throws Throwable {
+    decideImpl(decisionTaskWithHistoryIterator, null);
   }
 
-  private void decideImpl(HistoryHelper historyHelper, Functions.Proc query) throws Throwable {
+  private void decideImpl(
+      DecisionTaskWithHistoryIterator decisionTaskWithHistoryIterator, Functions.Proc query)
+      throws Throwable {
     try {
+      HistoryHelper historyHelper = new HistoryHelper(decisionTaskWithHistoryIterator);
       DecisionEventsIterator iterator = historyHelper.getIterator();
+      if ((decisionsHelper.getNextDecisionEventId()
+              != historyHelper.getPreviousStartedEventId()
+                  + 2) // getNextDecisionEventId() skips over completed.
+          && (decisionsHelper.getNextDecisionEventId() != 0
+              && historyHelper.getPreviousStartedEventId() != 0)) {
+        throw new IllegalStateException(
+            String.format(
+                "ReplayDecider expects next event id at %d. History's previous started event id is %d",
+                decisionsHelper.getNextDecisionEventId(),
+                historyHelper.getPreviousStartedEventId()));
+      }
+
       while (iterator.hasNext()) {
         DecisionEvents decision = iterator.next();
+
         context.setReplaying(decision.isReplay());
         context.setReplayCurrentTimeMilliseconds(decision.getReplayCurrentTimeMilliseconds());
 
@@ -388,17 +403,26 @@ private void decideImpl(HistoryHelper historyHelper, Functions.Proc query) throw
       if (query != null) {
         query.apply();
       }
-      workflow.close();
+      if (completed) {
+        close();
+      }
     }
   }
 
+  @Override
+  public void close() {
+    workflow.close();
+  }
+
   DecisionsHelper getDecisionsHelper() {
     return decisionsHelper;
   }
 
-  public byte[] query(HistoryHelper historyHelper, WorkflowQuery query) throws Throwable {
+  @Override
+  public byte[] query(DecisionTaskWithHistoryIterator decisionTaskIterator, WorkflowQuery query)
+      throws Throwable {
     AtomicReference<byte[]> result = new AtomicReference<>();
-    decideImpl(historyHelper, () -> result.set(workflow.query(query)));
+    decideImpl(decisionTaskIterator, () -> result.set(workflow.query(query)));
     return result.get();
   }
 }