Adding poller autoscaler, to dynamically resize number of pollers. (#761)

Adding poller autoscaler, to dynamically resize number of pollers.

Author: mindaugasbarcauskas

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

@@ -0,0 +1,96 @@
+/*
+ *  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 java.time.Duration;
+import java.util.concurrent.Executors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PollerAutoScaler {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(PollerAutoScaler.class);
+
+  private final Duration coolDownTime;
+  private final PollerUsageEstimator pollerUsageEstimator;
+  private final Recommender recommender;
+  private final ResizableSemaphore semaphore;
+  private int semaphoreSize;
+  private boolean shuttingDown;
+
+  public PollerAutoScaler(
+      Duration coolDownTime, PollerUsageEstimator pollerUsageEstimator, Recommender recommender) {
+    this.coolDownTime = coolDownTime;
+    this.pollerUsageEstimator = pollerUsageEstimator;
+    this.recommender = recommender;
+    this.semaphore = new ResizableSemaphore(recommender.getUpperValue());
+    this.semaphoreSize = recommender.getUpperValue();
+  }
+
+  public void start() {
+    Executors.newSingleThreadExecutor()
+        .submit(
+            new Runnable() {
+              @Override
+              public void run() {
+                while (!shuttingDown) {
+                  try {
+                    Thread.sleep(coolDownTime.toMillis());
+                    if (!shuttingDown) {
+                      resizePollers();
+                    }
+                  } catch (InterruptedException e) {
+                    LOGGER.info("interrupted wait for next poller scaling");
+                  }
+                }
+              }
+            });
+  }
+
+  public void stop() {
+    LOGGER.info("shutting down poller autoscaler");
+    shuttingDown = true;
+  }
+
+  protected void resizePollers() {
+    PollerUsage pollerUsage = pollerUsageEstimator.estimate();
+    int pollerCount =
+        recommender.recommend(this.semaphoreSize, pollerUsage.getPollerUtilizationRate());
+
+    int diff = this.semaphoreSize - pollerCount;
+    if (diff < 0) {
+      semaphore.release(diff * -1);
+    } else {
+      semaphore.decreasePermits(diff);
+    }
+
+    LOGGER.info(String.format("resized pollers to: %d", pollerCount));
+    this.semaphoreSize = pollerCount;
+  }
+
+  public void acquire() throws InterruptedException {
+    semaphore.acquire();
+  }
+
+  public void release() {
+    semaphore.release();
+  }
+
+  // For testing
+  protected int getSemaphoreSize() {
+    return semaphoreSize;
+  }
+}

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

@@ -0,0 +1,29 @@
+/*
+ *  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;
+
+public class PollerUsage {
+
+  private final float pollerUtilizationRate;
+
+  public PollerUsage(float pollerUtilizationRate) {
+    this.pollerUtilizationRate = pollerUtilizationRate;
+  }
+
+  public float getPollerUtilizationRate() {
+    return pollerUtilizationRate;
+  }
+}

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

@@ -0,0 +1,48 @@
+/*
+ *  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 java.util.concurrent.atomic.AtomicInteger;
+
+public class PollerUsageEstimator {
+
+  private AtomicInteger noopTaskCount = new AtomicInteger();
+  private AtomicInteger actionableTaskCount = new AtomicInteger();
+
+  public void increaseNoopTaskCount() {
+    noopTaskCount.addAndGet(1);
+  }
+
+  public void increaseActionableTaskCount() {
+    actionableTaskCount.addAndGet(1);
+  }
+
+  public PollerUsage estimate() {
+    int actionableTasks = actionableTaskCount.get();
+    int noopTasks = noopTaskCount.get();
+    if (noopTasks + actionableTasks == 0) {
+      return new PollerUsage(0);
+    }
+    PollerUsage result = new PollerUsage((actionableTasks * 1f) / (noopTasks + actionableTasks));
+    reset();
+    return result;
+  }
+
+  private void reset() {
+    noopTaskCount.set(0);
+    actionableTaskCount.set(0);
+  }
+}

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

@@ -0,0 +1,46 @@
+/*
+ *  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;
+
+public class Recommender {
+
+  private final float targetPollerUtilRate;
+  private final int upperValue;
+  private final int lowerValue;
+
+  public Recommender(float targetPollerUtilRate, int upperValue, int lowerValue) {
+    this.targetPollerUtilRate = targetPollerUtilRate;
+    this.upperValue = upperValue;
+    this.lowerValue = lowerValue;
+  }
+
+  public int recommend(int currentPollers, float pollerUtilizationRate) {
+    if (pollerUtilizationRate == 1) {
+      return upperValue;
+    }
+
+    float r = currentPollers * pollerUtilizationRate / targetPollerUtilRate;
+    return Math.round(Math.min(upperValue, Math.max(lowerValue, r)));
+  }
+
+  public int getUpperValue() {
+    return upperValue;
+  }
+
+  public int getLowerValue() {
+    return lowerValue;
+  }
+}

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

@@ -0,0 +1,28 @@
+/*
+ *  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 java.util.concurrent.Semaphore;
+
+public class ResizableSemaphore extends Semaphore {
+  public ResizableSemaphore(int permits) {
+    super(permits);
+  }
+
+  public void decreasePermits(int reduction) {
+    reducePermits(reduction);
+  }
+}

src/test/java/com/uber/cadence/internal/worker/autoscaler/PollerAutoScalerTest.java

@@ -0,0 +1,48 @@
+/*
+ *  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 static org.junit.Assert.*;
+
+import java.time.Duration;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class PollerAutoScalerTest {
+
+  @Test
+  public void testAutoScalerScalesPollers() {
+    PollerUsageEstimator pollerUsageEstimator = new PollerUsageEstimator();
+    Recommender recommender = new Recommender(0.5f, 100, 10);
+    PollerAutoScaler pollerAutoScaler =
+        new PollerAutoScaler(Duration.ofSeconds(1), pollerUsageEstimator, recommender);
+
+    assertEquals(100, pollerAutoScaler.getSemaphoreSize());
+
+    pollerUsageEstimator.increaseActionableTaskCount();
+    pollerAutoScaler.resizePollers();
+
+    assertEquals(100, pollerAutoScaler.getSemaphoreSize());
+
+    pollerUsageEstimator.increaseNoopTaskCount();
+
+    pollerAutoScaler.resizePollers();
+
+    assertEquals(10, pollerAutoScaler.getSemaphoreSize());
+  }
+}

src/test/java/com/uber/cadence/internal/worker/autoscaler/PollerUsageEstimatorTest.java

@@ -0,0 +1,94 @@
+/*
+ *  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 static org.junit.Assert.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.runners.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class PollerUsageEstimatorTest {
+
+  @Test
+  public void testUsageIsZero() {
+    PollerUsageEstimator pollerUsageEstimator = new PollerUsageEstimator();
+    PollerUsage pollerUsage = pollerUsageEstimator.estimate();
+    assertEquals(0f, pollerUsage.getPollerUtilizationRate(), 0);
+  }
+
+  @Test
+  public void testUsagesIs100Percent() {
+    PollerUsageEstimator pollerUsageEstimator = new PollerUsageEstimator();
+    pollerUsageEstimator.increaseActionableTaskCount();
+    pollerUsageEstimator.increaseActionableTaskCount();
+    PollerUsage pollerUsage = pollerUsageEstimator.estimate();
+    assertEquals(1f, pollerUsage.getPollerUtilizationRate(), 0);
+  }
+
+  @Test
+  public void testUsageCalculatedCorrectly() {
+    PollerUsageEstimator pollerUsageEstimator = new PollerUsageEstimator();
+    pollerUsageEstimator.increaseNoopTaskCount();
+    pollerUsageEstimator.increaseActionableTaskCount();
+    pollerUsageEstimator.increaseNoopTaskCount();
+    pollerUsageEstimator.increaseNoopTaskCount();
+
+    PollerUsage pollerUsage = pollerUsageEstimator.estimate();
+    assertEquals(0.25f, pollerUsage.getPollerUtilizationRate(), 0);
+  }
+
+  @Test
+  public void estimationIsReset() {
+    PollerUsageEstimator pollerUsageEstimator = new PollerUsageEstimator();
+    pollerUsageEstimator.increaseActionableTaskCount();
+
+    PollerUsage pollerUsage = pollerUsageEstimator.estimate();
+    assertEquals(1f, pollerUsage.getPollerUtilizationRate(), 0);
+
+    pollerUsage = pollerUsageEstimator.estimate();
+    assertEquals(0f, pollerUsage.getPollerUtilizationRate(), 0);
+  }
+
+  @Test
+  public void testThreadSafety() throws Exception {
+    PollerUsageEstimator pollerUsageEstimator = new PollerUsageEstimator();
+    List<Thread> threadList = new ArrayList<>();
+    for (int i = 0; i < 100; i++) {
+      threadList.add(
+          new Thread(
+              new Runnable() {
+                @Override
+                public void run() {
+                  for (int i = 0; i < 100; i++) {
+                    pollerUsageEstimator.increaseActionableTaskCount();
+                  }
+                }
+              }));
+    }
+
+    threadList.forEach(Thread::start);
+    for (Thread thread : threadList) {
+      thread.join();
+    }
+
+    pollerUsageEstimator.increaseNoopTaskCount();
+    assertEquals(10000f / 10001, pollerUsageEstimator.estimate().getPollerUtilizationRate(), 0);
+  }
+}