Introduce experimental flag and add logic for capacity based polling (#664)

Author: spmistry

build.gradle

@@ -108,11 +108,11 @@ compileTestJava {
 // Generation version.properties for value to be included into the request header
 task createProperties(dependsOn: processResources) {
     doLast {
-        def subdir = new File('$buildDir/resources/main/com/uber/cadence/')
+        def subdir = new File("$buildDir/resources/main/com/uber/cadence/")
         if( !subdir.exists() ) {
             subdir.mkdirs()
         }
-        new File('$buildDir/resources/main/com/uber/cadence/version.properties').withWriter { w ->
+        new File("$buildDir/resources/main/com/uber/cadence/version.properties").withWriter { w ->
             Properties p = new Properties()
             p['cadence-client-version'] = project.version.toString()
             p.store w, null

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

@@ -82,6 +82,11 @@ public void process(PollForDecisionTaskResponse t) {
     }
   }
 
+  @Override
+  public boolean hasCapacity() {
+    return true;
+  }
+
   public void subscribe(String taskList, Consumer<PollForDecisionTaskResponse> consumer) {
     subscribers.put(taskList, consumer);
   }

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

@@ -81,6 +81,11 @@ public void process(T task) {
         });
   }
 
+  @Override
+  public boolean hasCapacity() {
+    return taskExecutor.getActiveCount() < taskExecutor.getPoolSize();
+  }
+
   @Override
   public boolean isShutdown() {
     return taskExecutor.isShutdown();

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

@@ -256,6 +256,8 @@ 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() {
@@ -272,7 +274,31 @@ public void run() throws Exception {
         }
         taskExecutor.process(task);
       } finally {
+        releasePollSemaphore();
+      }
+    }
+
+    private void releasePollSemaphore() {
+      if (!pollerOptions.getPollOnlyIfExecutorHasCapacity()) {
         pollSemaphore.release();
+      } else {
+        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
+            try {
+              Thread.sleep(EXECUTOR_CAPACITY_CHECK_INTERVAL_MS);
+            } catch (InterruptedException ignored) {
+            }
+          }
+        }
       }
     }
   }

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

@@ -60,6 +60,8 @@ public static final class Builder {
 
     private String pollThreadNamePrefix;
 
+    private Boolean pollOnlyIfExecutorHasCapacity = Boolean.FALSE;
+
     private Thread.UncaughtExceptionHandler uncaughtExceptionHandler;
 
     private Builder() {}
@@ -75,6 +77,7 @@ private Builder(PollerOptions o) {
       this.pollBackoffMaximumInterval = o.getPollBackoffMaximumInterval();
       this.pollThreadCount = o.getPollThreadCount();
       this.pollThreadNamePrefix = o.getPollThreadNamePrefix();
+      this.pollOnlyIfExecutorHasCapacity = o.getPollOnlyIfExecutorHasCapacity();
       this.uncaughtExceptionHandler = o.getUncaughtExceptionHandler();
     }
 
@@ -133,6 +136,15 @@ public Builder setPollThreadNamePrefix(String pollThreadNamePrefix) {
       return this;
     }
 
+    /**
+     * The poller will check task executor's remaining capacity before polling tasks.
+     * This is to prevent task to get started but not being able to execute in time.
+     */
+    public Builder setPollOnlyIfExecutorHasCapacity(boolean pollOnlyIfExecutorHasCapacity) {
+      this.pollOnlyIfExecutorHasCapacity = pollOnlyIfExecutorHasCapacity;
+      return this;
+    }
+
     public PollerOptions build() {
       if (uncaughtExceptionHandler == null) {
         uncaughtExceptionHandler = (t, e) -> log.error("uncaught exception", e);
@@ -145,7 +157,8 @@ public PollerOptions build() {
           pollBackoffMaximumInterval,
           pollThreadCount,
           uncaughtExceptionHandler,
-          pollThreadNamePrefix);
+          pollThreadNamePrefix,
+          pollOnlyIfExecutorHasCapacity);
     }
   }
 
@@ -165,6 +178,8 @@ public PollerOptions build() {
 
   private final String pollThreadNamePrefix;
 
+  private final Boolean pollOnlyIfExecutorHasCapacity;
+
   private PollerOptions(
       int maximumPollRateIntervalMilliseconds,
       double maximumPollRatePerSecond,
@@ -173,7 +188,8 @@ private PollerOptions(
       Duration pollBackoffMaximumInterval,
       int pollThreadCount,
       Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
-      String pollThreadNamePrefix) {
+      String pollThreadNamePrefix,
+      boolean pollOnlyIfExecutorHasCapacity) {
     this.maximumPollRateIntervalMilliseconds = maximumPollRateIntervalMilliseconds;
     this.maximumPollRatePerSecond = maximumPollRatePerSecond;
     this.pollBackoffCoefficient = pollBackoffCoefficient;
@@ -182,6 +198,7 @@ private PollerOptions(
     this.pollThreadCount = pollThreadCount;
     this.uncaughtExceptionHandler = uncaughtExceptionHandler;
     this.pollThreadNamePrefix = pollThreadNamePrefix;
+    this.pollOnlyIfExecutorHasCapacity = pollOnlyIfExecutorHasCapacity;
   }
 
   public int getMaximumPollRateIntervalMilliseconds() {
@@ -216,6 +233,10 @@ public String getPollThreadNamePrefix() {
     return pollThreadNamePrefix;
   }
 
+  public Boolean getPollOnlyIfExecutorHasCapacity() {
+    return pollOnlyIfExecutorHasCapacity;
+  }
+
   @Override
   public String toString() {
     return "PollerOptions{"
@@ -233,6 +254,8 @@ public String toString() {
         + pollThreadCount
         + ", pollThreadNamePrefix='"
         + pollThreadNamePrefix
+        + ", pollOnlyIfExecutorHasCapacity='"
+        + pollOnlyIfExecutorHasCapacity
         + '\''
         + '}';
   }

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

@@ -19,4 +19,5 @@
 
 interface TaskExecutor<T> {
   void process(T task);
+  boolean hasCapacity();
 }