Making dynamic poller configurable. (#762)

Using dynamic poller and making it configurable.

Author: mindaugasbarcauskas

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

@@ -20,11 +20,12 @@
 import com.uber.cadence.internal.common.BackoffThrottler;
 import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.internal.metrics.MetricsType;
+import com.uber.cadence.internal.worker.autoscaler.AutoScaler;
+import com.uber.cadence.internal.worker.autoscaler.AutoScalerFactory;
 import com.uber.m3.tally.Scope;
 import java.util.Objects;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -69,6 +70,8 @@ interface ThrowingRunnable {
         log.error("Failure in thread " + t.getName(), e);
       };
 
+  private final AutoScaler pollerAutoScaler;
+
   public Poller(
       String identity,
       PollTask<T> pollTask,
@@ -86,6 +89,7 @@ public Poller(
     this.taskExecutor = taskExecutor;
     this.pollerOptions = pollerOptions;
     this.metricsScope = metricsScope;
+    this.pollerAutoScaler = AutoScalerFactory.getInstance().createAutoScaler(pollerOptions);
   }
 
   @Override
@@ -124,6 +128,8 @@ public void start() {
       pollExecutor.execute(new PollLoopTask(new PollExecutionTask()));
       metricsScope.counter(MetricsType.POLLER_START_COUNTER).inc(1);
     }
+
+    pollerAutoScaler.start();
   }
 
   @Override
@@ -155,6 +161,7 @@ public void shutdown() {
     } catch (InterruptedException e) {
     }
     taskExecutor.shutdown();
+    pollerAutoScaler.stop();
   }
 
   @Override
@@ -258,38 +265,39 @@ public void run() {
   private class PollExecutionTask implements Poller.ThrowingRunnable {
     private static final int EXECUTOR_CAPACITY_CHECK_INTERVAL_MS = 100;
     private static final int EXECUTOR_CAPACITY_CHECK_OFFSET_MS = 10;
-    private Semaphore pollSemaphore;
 
-    PollExecutionTask() {
-      this.pollSemaphore = new Semaphore(pollerOptions.getPollThreadCount());
-    }
+    PollExecutionTask() {}
 
     @Override
     public void run() throws Exception {
       try {
-        pollSemaphore.acquire();
-        T task = pollTask.poll();
-        if (task == null) {
-          return;
+        pollerAutoScaler.acquire();
+        try {
+          T task = pollTask.poll();
+          if (task == null) {
+            pollerAutoScaler.increaseNoopPollCount();
+            return;
+          }
+
+          pollerAutoScaler.increaseActionablePollCount();
+          taskExecutor.process(task);
+        } finally {
+          checkIfTaskHasExecutorHasCapacity();
         }
-        taskExecutor.process(task);
       } finally {
-        releasePollSemaphore();
+        pollerAutoScaler.release();
       }
     }
 
-    private void releasePollSemaphore() {
-      if (!pollerOptions.getPollOnlyIfExecutorHasCapacity()) {
-        pollSemaphore.release();
-      } else {
+    private void checkIfTaskHasExecutorHasCapacity() {
+      if (pollerOptions.getPollOnlyIfExecutorHasCapacity()) {
         while (true) {
           // sleep to avoid racing condition
           try {
             Thread.sleep(EXECUTOR_CAPACITY_CHECK_OFFSET_MS);
           } catch (InterruptedException ignored) {
           }
           if (taskExecutor.hasCapacity()) {
-            pollSemaphore.release();
             break;
           } else {
             // sleep to avoid busy loop

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

@@ -0,0 +1,103 @@
+/*
+ *  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.worker;
+
+import java.time.Duration;
+import java.util.Objects;
+
+public class PollerAutoScalerOptions {
+
+  private Duration pollerScalingInterval;
+  private int minConcurrentPollers;
+  private float targetPollerUtilisation;
+
+  private PollerAutoScalerOptions() {}
+
+  public static class Builder {
+
+    private Duration pollerScalingInterval = Duration.ofMinutes(1);
+    private int minConcurrentPollers = 1;
+    private float targetPollerUtilisation = 0.6f;
+
+    private Builder() {}
+
+    public static Builder newBuilder() {
+      return new Builder();
+    }
+
+    public Builder setPollerScalingInterval(Duration duration) {
+      this.pollerScalingInterval = duration;
+      return this;
+    }
+
+    public Builder setMinConcurrentPollers(int minConcurrentPollers) {
+      this.minConcurrentPollers = minConcurrentPollers;
+      return this;
+    }
+
+    public Builder setTargetPollerUtilisation(float targetPollerUtilisation) {
+      this.targetPollerUtilisation = targetPollerUtilisation;
+      return this;
+    }
+
+    public PollerAutoScalerOptions build() {
+      PollerAutoScalerOptions pollerAutoScalerOptions = new PollerAutoScalerOptions();
+      pollerAutoScalerOptions.pollerScalingInterval = this.pollerScalingInterval;
+      pollerAutoScalerOptions.minConcurrentPollers = this.minConcurrentPollers;
+      pollerAutoScalerOptions.targetPollerUtilisation = this.targetPollerUtilisation;
+      return pollerAutoScalerOptions;
+    }
+  }
+
+  public Duration getPollerScalingInterval() {
+    return pollerScalingInterval;
+  }
+
+  public int getMinConcurrentPollers() {
+    return minConcurrentPollers;
+  }
+
+  public float getTargetPollerUtilisation() {
+    return targetPollerUtilisation;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    PollerAutoScalerOptions that = (PollerAutoScalerOptions) o;
+    return minConcurrentPollers == that.minConcurrentPollers
+        && Float.compare(that.targetPollerUtilisation, targetPollerUtilisation) == 0
+        && Objects.equals(pollerScalingInterval, that.pollerScalingInterval);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(pollerScalingInterval, minConcurrentPollers, targetPollerUtilisation);
+  }
+
+  @Override
+  public String toString() {
+    return "PollerAutoScalerOptions{"
+        + "pollerScalingInterval="
+        + pollerScalingInterval
+        + ", minConcurrentPollers="
+        + minConcurrentPollers
+        + ", targetPollerUtilisation="
+        + targetPollerUtilisation
+        + '}';
+  }
+}

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

@@ -64,6 +64,8 @@ public static final class Builder {
 
     private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
 
+    private PollerAutoScalerOptions pollerAutoScalerOptions;
+
     private Builder() {}
 
     private Builder(PollerOptions o) {
@@ -79,6 +81,7 @@ private Builder(PollerOptions o) {
       this.pollThreadNamePrefix = o.getPollThreadNamePrefix();
       this.pollOnlyIfExecutorHasCapacity = o.getPollOnlyIfExecutorHasCapacity();
       this.uncaughtExceptionHandler = o.getUncaughtExceptionHandler();
+      this.pollerAutoScalerOptions = o.getPollerAutoScalerOptions();
     }
 
     /** Defines interval for measuring poll rate. Larger the interval more spiky can be the load. */
@@ -145,6 +148,11 @@ public Builder setPollOnlyIfExecutorHasCapacity(boolean pollOnlyIfExecutorHasCap
       return this;
     }
 
+    public Builder setPollerAutoScalerOptions(PollerAutoScalerOptions pollerAutoScalerOptions) {
+      this.pollerAutoScalerOptions = pollerAutoScalerOptions;
+      return this;
+    }
+
     public PollerOptions build() {
       if (uncaughtExceptionHandler == null) {
         uncaughtExceptionHandler = (t, e) -> log.error("uncaught exception", e);
@@ -158,7 +166,8 @@ public PollerOptions build() {
           pollThreadCount,
           uncaughtExceptionHandler,
           pollThreadNamePrefix,
-          pollOnlyIfExecutorHasCapacity);
+          pollOnlyIfExecutorHasCapacity,
+          pollerAutoScalerOptions);
     }
   }
 
@@ -180,6 +189,8 @@ public PollerOptions build() {
 
   private final Boolean pollOnlyIfExecutorHasCapacity;
 
+  private final PollerAutoScalerOptions pollerAutoScalerOptions;
+
   private PollerOptions(
       int maximumPollRateIntervalMilliseconds,
       double maximumPollRatePerSecond,
@@ -189,7 +200,8 @@ private PollerOptions(
       int pollThreadCount,
       Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
       String pollThreadNamePrefix,
-      boolean pollOnlyIfExecutorHasCapacity) {
+      boolean pollOnlyIfExecutorHasCapacity,
+      PollerAutoScalerOptions pollerAutoScalerOptions) {
     this.maximumPollRateIntervalMilliseconds = maximumPollRateIntervalMilliseconds;
     this.maximumPollRatePerSecond = maximumPollRatePerSecond;
     this.pollBackoffCoefficient = pollBackoffCoefficient;
@@ -199,6 +211,7 @@ private PollerOptions(
     this.uncaughtExceptionHandler = uncaughtExceptionHandler;
     this.pollThreadNamePrefix = pollThreadNamePrefix;
     this.pollOnlyIfExecutorHasCapacity = pollOnlyIfExecutorHasCapacity;
+    this.pollerAutoScalerOptions = pollerAutoScalerOptions;
   }
 
   public int getMaximumPollRateIntervalMilliseconds() {
@@ -237,6 +250,10 @@ public Boolean getPollOnlyIfExecutorHasCapacity() {
     return pollOnlyIfExecutorHasCapacity;
   }
 
+  public PollerAutoScalerOptions getPollerAutoScalerOptions() {
+    return pollerAutoScalerOptions;
+  }
+
   @Override
   public String toString() {
     return "PollerOptions{"
@@ -256,6 +273,8 @@ public String toString() {
         + pollThreadNamePrefix
         + ", pollOnlyIfExecutorHasCapacity='"
         + pollOnlyIfExecutorHasCapacity
+        + ", pollerAutoScalerOptions='"
+        + pollerAutoScalerOptions
         + '\''
         + '}';
   }

src/main/java/com/uber/cadence/internal/worker/autoscaler/AutoScaler.java

@@ -0,0 +1,42 @@
+/*
+ *  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.worker.autoscaler;
+
+/**
+ * Poller autoscaler interface for acquiring and releasing locks. In order to control number of
+ * concurrent operations.
+ */
+public interface AutoScaler {
+
+  void start();
+
+  void stop();
+
+  /**
+   * Reduce the number of available locks. Intended to be blocking operation until lock is acquired.
+   */
+  void acquire() throws InterruptedException;
+
+  /**
+   * Releases lock into the autoscaler pool. Release should be always called in same process,
+   * failing to do so is considered a usage error.
+   */
+  void release();
+
+  void increaseNoopPollCount();
+
+  void increaseActionablePollCount();
+}

src/main/java/com/uber/cadence/internal/worker/autoscaler/AutoScalerFactory.java

@@ -0,0 +1,45 @@
+/*
+ *  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.worker.autoscaler;
+
+import com.uber.cadence.internal.worker.PollerAutoScalerOptions;
+import com.uber.cadence.internal.worker.PollerOptions;
+
+public class AutoScalerFactory {
+
+  private static final AutoScalerFactory INSTANCE = new AutoScalerFactory();
+
+  private AutoScalerFactory() {}
+
+  public AutoScaler createAutoScaler(PollerOptions pollerOptions) {
+    if (pollerOptions == null || pollerOptions.getPollerAutoScalerOptions() == null) {
+      return new NoopAutoScaler();
+    }
+
+    PollerAutoScalerOptions autoScalerOptions = pollerOptions.getPollerAutoScalerOptions();
+    return new PollerAutoScaler(
+        autoScalerOptions.getPollerScalingInterval(),
+        new PollerUsageEstimator(),
+        new Recommender(
+            autoScalerOptions.getTargetPollerUtilisation(),
+            pollerOptions.getPollThreadCount(),
+            autoScalerOptions.getMinConcurrentPollers()));
+  }
+
+  public static AutoScalerFactory getInstance() {
+    return INSTANCE;
+  }
+}