Configure eager reservations per workflow task

Author: Sushisource

temporal-sdk/src/main/java/io/temporal/internal/worker/ActivityWorker.java

@@ -218,9 +218,8 @@ public WorkerLifecycleState getLifecycleState() {
     return poller.getLifecycleState();
   }
 
-  public EagerActivityDispatcher getEagerActivityDispatcher(
-      int maxConcurrentEagerActivityExecutionSize) {
-    return new EagerActivityDispatcherImpl(maxConcurrentEagerActivityExecutionSize);
+  public EagerActivityDispatcher getEagerActivityDispatcher() {
+    return new EagerActivityDispatcherImpl();
   }
 
   private PollerOptions getPollerOptions(SingleWorkerOptions options) {
@@ -486,13 +485,6 @@ private void logExceptionDuringResultReporting(
   }
 
   private final class EagerActivityDispatcherImpl implements EagerActivityDispatcher {
-    private final EagerActivitySlotLimiter eagerActivitySlotLimiter;
-
-    private EagerActivityDispatcherImpl(int maxConcurrentEagerActivityExecutionSize) {
-      this.eagerActivitySlotLimiter =
-          new EagerActivitySlotLimiter(maxConcurrentEagerActivityExecutionSize);
-    }
-
     @Override
     public Optional<SlotPermit> tryReserveActivitySlot(
         ScheduleActivityTaskCommandAttributesOrBuilder commandAttributes) {
@@ -501,31 +493,15 @@ public Optional<SlotPermit> tryReserveActivitySlot(
               commandAttributes.getTaskQueue().getName(), ActivityWorker.this.taskQueue)) {
         return Optional.empty();
       }
-      if (!eagerActivitySlotLimiter.tryReserve()) {
-        return Optional.empty();
-      }
-      Optional<SlotPermit> permit = Optional.empty();
-      try {
-        permit =
-            ActivityWorker.this.slotSupplier.tryReserveSlot(
-                new SlotReservationData(
-                    ActivityWorker.this.taskQueue, options.getIdentity(), options.getBuildId()));
-        return permit;
-      } finally {
-        if (!permit.isPresent()) {
-          eagerActivitySlotLimiter.release();
-        }
-      }
+      return ActivityWorker.this.slotSupplier.tryReserveSlot(
+          new SlotReservationData(
+              ActivityWorker.this.taskQueue, options.getIdentity(), options.getBuildId()));
     }
 
     @Override
     public void releaseActivitySlotReservations(Iterable<SlotPermit> permits) {
       for (SlotPermit permit : permits) {
-        try {
-          ActivityWorker.this.slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);
-        } finally {
-          eagerActivitySlotLimiter.release();
-        }
+        ActivityWorker.this.slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);
       }
     }
 
@@ -535,39 +511,9 @@ public void dispatchActivity(PollActivityTaskQueueResponse activity, SlotPermit
           new ActivityTask(
               activity,
               permit,
-              () -> {
-                try {
+              () ->
                   ActivityWorker.this.slotSupplier.releaseSlot(
-                      SlotReleaseReason.taskComplete(), permit);
-                } finally {
-                  eagerActivitySlotLimiter.release();
-                }
-              }));
-    }
-  }
-
-  static final class EagerActivitySlotLimiter {
-    private final int maxConcurrent;
-    private int heldSlotCount;
-
-    EagerActivitySlotLimiter(int maxConcurrent) {
-      this.maxConcurrent = maxConcurrent;
-    }
-
-    synchronized boolean tryReserve() {
-      if (maxConcurrent > 0 && heldSlotCount >= maxConcurrent) {
-        return false;
-      }
-      heldSlotCount++;
-      return true;
-    }
-
-    synchronized void release() {
-      if (heldSlotCount <= 0) {
-        throw new IllegalStateException(
-            "Trying to release an unreserved eager activity slot. This is an SDK bug.");
-      }
-      heldSlotCount--;
+                      SlotReleaseReason.taskComplete(), permit)));
     }
   }
 }

temporal-sdk/src/main/java/io/temporal/internal/worker/EagerActivitySlotsReservation.java

@@ -7,7 +7,6 @@
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse;
 import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedRequest;
 import io.temporal.api.workflowservice.v1.RespondWorkflowTaskCompletedResponse;
-import io.temporal.internal.Config;
 import io.temporal.worker.tuning.SlotPermit;
 import java.io.Closeable;
 import java.util.ArrayList;
@@ -19,10 +18,13 @@
 @NotThreadSafe
 class EagerActivitySlotsReservation implements Closeable {
   private final EagerActivityDispatcher eagerActivityDispatcher;
+  private final int maxReservations;
   private final List<SlotPermit> reservedSlots = new ArrayList<>();
 
-  EagerActivitySlotsReservation(EagerActivityDispatcher eagerActivityDispatcher) {
+  EagerActivitySlotsReservation(
+      EagerActivityDispatcher eagerActivityDispatcher, int maxReservations) {
     this.eagerActivityDispatcher = eagerActivityDispatcher;
+    this.maxReservations = maxReservations;
   }
 
   public void applyToRequest(RespondWorkflowTaskCompletedRequest.Builder mutableRequest) {
@@ -33,7 +35,7 @@ public void applyToRequest(RespondWorkflowTaskCompletedRequest.Builder mutableRe
       ScheduleActivityTaskCommandAttributes commandAttributes =
           command.getScheduleActivityTaskCommandAttributes();
       if (!commandAttributes.getRequestEagerExecution()) continue;
-      boolean atLimit = this.reservedSlots.size() >= Config.EAGER_ACTIVITIES_LIMIT;
+      boolean atLimit = this.reservedSlots.size() >= this.maxReservations;
       Optional<SlotPermit> permit = Optional.empty();
       if (!atLimit) {
         permit = this.eagerActivityDispatcher.tryReserveActivitySlot(commandAttributes);

temporal-sdk/src/main/java/io/temporal/internal/worker/SyncActivityWorker.java

@@ -163,9 +163,8 @@ public WorkerLifecycleState getLifecycleState() {
     }
   }
 
-  public EagerActivityDispatcher getEagerActivityDispatcher(
-      int maxConcurrentEagerActivityExecutionSize) {
-    return this.worker.getEagerActivityDispatcher(maxConcurrentEagerActivityExecutionSize);
+  public EagerActivityDispatcher getEagerActivityDispatcher() {
+    return this.worker.getEagerActivityDispatcher();
   }
 
   public boolean isAnyTypeSupported() {

temporal-sdk/src/main/java/io/temporal/internal/worker/SyncWorkflowWorker.java

@@ -67,6 +67,7 @@ public SyncWorkflowWorker(
       String stickyTaskQueueName,
       @Nonnull WorkflowThreadExecutor workflowThreadExecutor,
       @Nonnull EagerActivityDispatcher eagerActivityDispatcher,
+      int maxEagerActivityReservationsPerWorkflowTask,
       @Nonnull SlotSupplier<WorkflowSlotInfo> slotSupplier,
       @Nonnull SlotSupplier<LocalActivitySlotInfo> laSlotSupplier,
       @Nonnull NamespaceCapabilities namespaceCapabilities) {
@@ -123,6 +124,7 @@ public SyncWorkflowWorker(
             cache,
             taskHandler,
             eagerActivityDispatcher,
+            maxEagerActivityReservationsPerWorkflowTask,
             slotSupplier,
             namespaceCapabilities);
 

temporal-sdk/src/main/java/io/temporal/internal/worker/WorkflowWorker.java

@@ -52,6 +52,7 @@ final class WorkflowWorker implements SuspendableWorker {
   private final Scope workerMetricsScope;
   private final GrpcRetryer grpcRetryer;
   private final EagerActivityDispatcher eagerActivityDispatcher;
+  private final int maxEagerActivityReservationsPerWorkflowTask;
   private final TrackingSlotSupplier<WorkflowSlotInfo> slotSupplier;
 
   private final TaskCounter taskCounter = new TaskCounter();
@@ -77,6 +78,7 @@ public WorkflowWorker(
       @Nonnull WorkflowExecutorCache cache,
       @Nonnull WorkflowTaskHandler handler,
       @Nonnull EagerActivityDispatcher eagerActivityDispatcher,
+      int maxEagerActivityReservationsPerWorkflowTask,
       @Nonnull SlotSupplier<WorkflowSlotInfo> slotSupplier,
       @Nonnull NamespaceCapabilities namespaceCapabilities) {
     this.service = Objects.requireNonNull(service);
@@ -92,6 +94,7 @@ public WorkflowWorker(
     this.handler = Objects.requireNonNull(handler);
     this.grpcRetryer = new GrpcRetryer(service.getServerCapabilities());
     this.eagerActivityDispatcher = eagerActivityDispatcher;
+    this.maxEagerActivityReservationsPerWorkflowTask = maxEagerActivityReservationsPerWorkflowTask;
     this.slotSupplier = new TrackingSlotSupplier<>(slotSupplier, this.workerMetricsScope);
     this.namespaceCapabilities = namespaceCapabilities;
   }
@@ -478,7 +481,8 @@ public void handle(WorkflowTask task) throws Exception {
                     RespondWorkflowTaskCompletedRequest.Builder requestBuilder =
                         taskCompleted.toBuilder();
                     try (EagerActivitySlotsReservation activitySlotsReservation =
-                        new EagerActivitySlotsReservation(eagerActivityDispatcher)) {
+                        new EagerActivitySlotsReservation(
+                            eagerActivityDispatcher, maxEagerActivityReservationsPerWorkflowTask)) {
                       activitySlotsReservation.applyToRequest(requestBuilder);
                       RespondWorkflowTaskCompletedResponse response =
                           sendTaskCompleted(

temporal-sdk/src/main/java/io/temporal/worker/Worker.java

@@ -173,8 +173,7 @@ private static final class TaskSnapshot {
 
     EagerActivityDispatcher eagerActivityDispatcher =
         (activityWorker != null && !this.options.isEagerExecutionDisabled())
-            ? activityWorker.getEagerActivityDispatcher(
-                this.options.getMaxConcurrentEagerActivityExecutionSize())
+            ? activityWorker.getEagerActivityDispatcher()
             : new EagerActivityDispatcher.NoopEagerActivityDispatcher();
 
     SingleWorkerOptions nexusOptions =
@@ -245,6 +244,7 @@ private static final class TaskSnapshot {
             stickyTaskQueueName,
             workflowThreadExecutor,
             eagerActivityDispatcher,
+            this.options.getMaxEagerActivityReservationsPerWorkflowTask(),
             workflowSlotSupplier,
             localActivitySlotSupplier,
             namespaceCapabilities);

temporal-sdk/src/main/java/io/temporal/worker/WorkerOptions.java

@@ -4,6 +4,7 @@
 
 import com.google.common.base.Preconditions;
 import io.temporal.common.Experimental;
+import io.temporal.internal.Config;
 import io.temporal.serviceclient.WorkflowServiceStubsOptions;
 import io.temporal.worker.tuning.*;
 import java.time.Duration;
@@ -64,7 +65,7 @@ public static final class Builder {
     private Duration defaultHeartbeatThrottleInterval;
     private Duration stickyQueueScheduleToStartTimeout;
     private boolean disableEagerExecution;
-    private int maxConcurrentEagerActivityExecutionSize;
+    private int maxEagerActivityReservationsPerWorkflowTask = Config.EAGER_ACTIVITIES_LIMIT;
     private String buildId;
     private boolean useBuildIdForVersioning;
     private Duration stickyTaskQueueDrainTimeout;
@@ -110,7 +111,8 @@ private Builder(WorkerOptions o) {
       this.defaultHeartbeatThrottleInterval = o.defaultHeartbeatThrottleInterval;
       this.stickyQueueScheduleToStartTimeout = o.stickyQueueScheduleToStartTimeout;
       this.disableEagerExecution = o.disableEagerExecution;
-      this.maxConcurrentEagerActivityExecutionSize = o.maxConcurrentEagerActivityExecutionSize;
+      this.maxEagerActivityReservationsPerWorkflowTask =
+          o.maxEagerActivityReservationsPerWorkflowTask;
       this.useBuildIdForVersioning = o.useBuildIdForVersioning;
       this.buildId = o.buildId;
       this.stickyTaskQueueDrainTimeout = o.stickyTaskQueueDrainTimeout;
@@ -403,15 +405,15 @@ public Builder setDisableEagerExecution(boolean disableEagerExecution) {
     }
 
     /**
-     * Sets the maximum number of eager activities that can be running concurrently.
+     * Sets the maximum number of activity slots that may be reserved for eager execution when
+     * completing a workflow task.
      *
-     * <p>When nonzero, eager activity execution will not be requested if it would cause the number
-     * of running eager activities to exceed this value. The default of zero means unlimited and
-     * therefore only bound by the activity slot supplier.
+     * <p>The default is 3. Setting this to zero disables eager activity execution.
      */
-    public Builder setMaxConcurrentEagerActivityExecutionSize(
-        int maxConcurrentEagerActivityExecutionSize) {
-      this.maxConcurrentEagerActivityExecutionSize = maxConcurrentEagerActivityExecutionSize;
+    public Builder setMaxEagerActivityReservationsPerWorkflowTask(
+        int maxEagerActivityReservationsPerWorkflowTask) {
+      this.maxEagerActivityReservationsPerWorkflowTask =
+          maxEagerActivityReservationsPerWorkflowTask;
       return this;
     }
 
@@ -638,7 +640,7 @@ public WorkerOptions build() {
           defaultHeartbeatThrottleInterval,
           stickyQueueScheduleToStartTimeout,
           disableEagerExecution,
-          maxConcurrentEagerActivityExecutionSize,
+          maxEagerActivityReservationsPerWorkflowTask,
           useBuildIdForVersioning,
           buildId,
           stickyTaskQueueDrainTimeout,
@@ -664,8 +666,8 @@ public WorkerOptions validateAndBuildWithDefaults() {
       Preconditions.checkState(
           maxConcurrentActivityExecutionSize >= 0, "negative maxConcurrentActivityExecutionSize");
       Preconditions.checkState(
-          maxConcurrentEagerActivityExecutionSize >= 0,
-          "negative maxConcurrentEagerActivityExecutionSize");
+          maxEagerActivityReservationsPerWorkflowTask >= 0,
+          "negative maxEagerActivityReservationsPerWorkflowTask");
       Preconditions.checkState(
           maxConcurrentWorkflowTaskExecutionSize >= 0,
           "negative maxConcurrentWorkflowTaskExecutionSize");
@@ -777,7 +779,7 @@ public WorkerOptions validateAndBuildWithDefaults() {
               ? DEFAULT_STICKY_SCHEDULE_TO_START_TIMEOUT
               : stickyQueueScheduleToStartTimeout,
           disableEagerExecution,
-          maxConcurrentEagerActivityExecutionSize,
+          maxEagerActivityReservationsPerWorkflowTask,
           useBuildIdForVersioning,
           buildId,
           stickyTaskQueueDrainTimeout == null
@@ -816,7 +818,7 @@ public WorkerOptions validateAndBuildWithDefaults() {
   private final Duration defaultHeartbeatThrottleInterval;
   private final @Nonnull Duration stickyQueueScheduleToStartTimeout;
   private final boolean disableEagerExecution;
-  private final int maxConcurrentEagerActivityExecutionSize;
+  private final int maxEagerActivityReservationsPerWorkflowTask;
   private final boolean useBuildIdForVersioning;
   private final String buildId;
   private final Duration stickyTaskQueueDrainTimeout;
@@ -852,7 +854,7 @@ private WorkerOptions(
       Duration defaultHeartbeatThrottleInterval,
       @Nonnull Duration stickyQueueScheduleToStartTimeout,
       boolean disableEagerExecution,
-      int maxConcurrentEagerActivityExecutionSize,
+      int maxEagerActivityReservationsPerWorkflowTask,
       boolean useBuildIdForVersioning,
       String buildId,
       Duration stickyTaskQueueDrainTimeout,
@@ -886,7 +888,7 @@ private WorkerOptions(
     this.defaultHeartbeatThrottleInterval = defaultHeartbeatThrottleInterval;
     this.stickyQueueScheduleToStartTimeout = stickyQueueScheduleToStartTimeout;
     this.disableEagerExecution = maxTaskQueueActivitiesPerSecond > 0 ? true : disableEagerExecution;
-    this.maxConcurrentEagerActivityExecutionSize = maxConcurrentEagerActivityExecutionSize;
+    this.maxEagerActivityReservationsPerWorkflowTask = maxEagerActivityReservationsPerWorkflowTask;
     this.useBuildIdForVersioning = useBuildIdForVersioning;
     this.buildId = buildId;
     this.stickyTaskQueueDrainTimeout = stickyTaskQueueDrainTimeout;
@@ -1012,8 +1014,8 @@ public boolean isEagerExecutionDisabled() {
     return disableEagerExecution;
   }
 
-  public int getMaxConcurrentEagerActivityExecutionSize() {
-    return maxConcurrentEagerActivityExecutionSize;
+  public int getMaxEagerActivityReservationsPerWorkflowTask() {
+    return maxEagerActivityReservationsPerWorkflowTask;
   }
 
   public boolean isUsingBuildIdForVersioning() {
@@ -1097,7 +1099,8 @@ && compare(maxTaskQueueActivitiesPerSecond, that.maxTaskQueueActivitiesPerSecond
         && localActivityWorkerOnly == that.localActivityWorkerOnly
         && defaultDeadlockDetectionTimeout == that.defaultDeadlockDetectionTimeout
         && disableEagerExecution == that.disableEagerExecution
-        && maxConcurrentEagerActivityExecutionSize == that.maxConcurrentEagerActivityExecutionSize
+        && maxEagerActivityReservationsPerWorkflowTask
+            == that.maxEagerActivityReservationsPerWorkflowTask
         && useBuildIdForVersioning == that.useBuildIdForVersioning
         && Objects.equals(workerTuner, that.workerTuner)
         && Objects.equals(maxHeartbeatThrottleInterval, that.maxHeartbeatThrottleInterval)
@@ -1137,7 +1140,7 @@ public int hashCode() {
         defaultHeartbeatThrottleInterval,
         stickyQueueScheduleToStartTimeout,
         disableEagerExecution,
-        maxConcurrentEagerActivityExecutionSize,
+        maxEagerActivityReservationsPerWorkflowTask,
         useBuildIdForVersioning,
         buildId,
         stickyTaskQueueDrainTimeout,
@@ -1189,8 +1192,8 @@ public String toString() {
         + stickyQueueScheduleToStartTimeout
         + ", disableEagerExecution="
         + disableEagerExecution
-        + ", maxConcurrentEagerActivityExecutionSize="
-        + maxConcurrentEagerActivityExecutionSize
+        + ", maxEagerActivityReservationsPerWorkflowTask="
+        + maxEagerActivityReservationsPerWorkflowTask
         + ", useBuildIdForVersioning="
         + useBuildIdForVersioning
         + ", buildId='"