Revert configurable slot provider (#2134)

* Revert "Resource based tuner (#2110)"

This reverts commit 8a2d5cd.

* Revert "Slot supplier interface & fixed-size implementation (#2014)"

This reverts commit d2a06fc.

* Fix merge conflict

* Keep Publish Test Report step

* Add tests for worker slots

* Fix white space

* One other whitespace change

Author: Quinn-With-Two-Ns

.github/workflows/ci.yml

@@ -33,12 +33,6 @@ jobs:
         USE_DOCKER_SERVICE: false
       run: ./gradlew --no-daemon test -x checkLicenseMain -x checkLicenses -x spotlessCheck -x spotlessApply -x spotlessJava -P edgeDepsTest
 
-    - name: Run independent resource tuner test
-      env:
-        USER: unittest
-        USE_DOCKER_SERVICE: false
-      run: ./gradlew --no-daemon temporal-sdk:testResourceIndependent -x checkLicenseMain -x checkLicenses -x spotlessCheck -x spotlessApply -x spotlessJava -P edgeDepsTest
-
     - name: Publish Test Report
       uses: mikepenz/action-junit-report@v4
       if: success() || failure() # always run even if the previous step fails

temporal-sdk/build.gradle

@@ -36,16 +36,3 @@ task registerNamespace(type: JavaExec) {
 }
 
 test.dependsOn 'registerNamespace'
-
-test {
-    useJUnit {
-        excludeCategories 'io.temporal.worker.IndependentResourceBasedTests'
-    }
-}
-
-task testResourceIndependent(type: Test) {
-    useJUnit {
-        includeCategories 'io.temporal.worker.IndependentResourceBasedTests'
-        maxParallelForks = 1
-    }
-}

temporal-sdk/src/main/java/io/temporal/internal/activity/ActivityPollResponseToInfo.java

@@ -30,12 +30,11 @@
 import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueRequest;
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse;
-import io.temporal.internal.activity.ActivityPollResponseToInfo;
 import io.temporal.internal.common.ProtobufTimeUtils;
 import io.temporal.serviceclient.WorkflowServiceStubs;
 import io.temporal.worker.MetricsType;
-import io.temporal.worker.tuning.*;
 import java.util.Objects;
+import java.util.concurrent.Semaphore;
 import java.util.function.Supplier;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -46,7 +45,7 @@ final class ActivityPollTask implements Poller.PollTask<ActivityTask> {
   private static final Logger log = LoggerFactory.getLogger(ActivityPollTask.class);
 
   private final WorkflowServiceStubs service;
-  private final TrackingSlotSupplier<ActivitySlotInfo> slotSupplier;
+  private final Semaphore pollSemaphore;
   private final Scope metricsScope;
   private final PollActivityTaskQueueRequest pollRequest;
 
@@ -58,11 +57,11 @@ public ActivityPollTask(
       @Nullable String buildId,
       boolean useBuildIdForVersioning,
       double activitiesPerSecond,
-      @Nonnull TrackingSlotSupplier<ActivitySlotInfo> slotSupplier,
+      Semaphore pollSemaphore,
       @Nonnull Scope metricsScope,
       @Nonnull Supplier<GetSystemInfoResponse.Capabilities> serverCapabilities) {
     this.service = Objects.requireNonNull(service);
-    this.slotSupplier = slotSupplier;
+    this.pollSemaphore = pollSemaphore;
     this.metricsScope = Objects.requireNonNull(metricsScope);
 
     PollActivityTaskQueueRequest.Builder pollRequest =
@@ -93,22 +92,13 @@ public ActivityTask poll() {
       log.trace("poll request begin: " + pollRequest);
     }
     PollActivityTaskQueueResponse response;
-    SlotPermit permit;
     boolean isSuccessful = false;
 
     try {
-      permit =
-          slotSupplier.reserveSlot(
-              new SlotReservationData(
-                  pollRequest.getTaskQueue().getName(),
-                  pollRequest.getIdentity(),
-                  pollRequest.getWorkerVersionCapabilities().getBuildId()));
+      pollSemaphore.acquire();
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
       return null;
-    } catch (Exception e) {
-      log.warn("Error while trying to reserve a slot for an activity", e.getCause());
-      return null;
     }
 
     try {
@@ -128,20 +118,9 @@ public ActivityTask poll() {
               ProtobufTimeUtils.toM3Duration(
                   response.getStartedTime(), response.getCurrentAttemptScheduledTime()));
       isSuccessful = true;
-      slotSupplier.markSlotUsed(
-          new ActivitySlotInfo(
-              ActivityPollResponseToInfo.toActivityInfoImpl(
-                  response,
-                  pollRequest.getNamespace(),
-                  pollRequest.getTaskQueue().getNormalName(),
-                  false),
-              pollRequest.getIdentity(),
-              pollRequest.getWorkerVersionCapabilities().getBuildId()),
-          permit);
-      return new ActivityTask(
-          response, () -> slotSupplier.releaseSlot(SlotReleaseReason.taskComplete(), permit));
+      return new ActivityTask(response, pollSemaphore::release);
     } finally {
-      if (!isSuccessful) slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);
+      if (!isSuccessful) pollSemaphore.release();
     }
   }
 }

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

@@ -39,10 +39,9 @@
 import io.temporal.serviceclient.rpcretry.DefaultStubServiceOperationRpcRetryOptions;
 import io.temporal.worker.MetricsType;
 import io.temporal.worker.WorkerMetricsTag;
-import io.temporal.worker.tuning.*;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nonnull;
 import org.slf4j.Logger;
@@ -65,16 +64,16 @@ final class ActivityWorker implements SuspendableWorker {
   private final Scope workerMetricsScope;
   private final GrpcRetryer grpcRetryer;
   private final GrpcRetryer.GrpcRetryerOptions replyGrpcRetryerOptions;
-  private final TrackingSlotSupplier<ActivitySlotInfo> slotSupplier;
+  private final int executorSlots;
+  private final Semaphore executorSlotsSemaphore;
 
   public ActivityWorker(
       @Nonnull WorkflowServiceStubs service,
       @Nonnull String namespace,
       @Nonnull String taskQueue,
       double taskQueueActivitiesPerSecond,
       @Nonnull SingleWorkerOptions options,
-      @Nonnull ActivityTaskHandler handler,
-      @Nonnull TrackingSlotSupplier<ActivitySlotInfo> slotSupplier) {
+      @Nonnull ActivityTaskHandler handler) {
     this.service = Objects.requireNonNull(service);
     this.namespace = Objects.requireNonNull(namespace);
     this.taskQueue = Objects.requireNonNull(taskQueue);
@@ -88,8 +87,8 @@ public ActivityWorker(
     this.replyGrpcRetryerOptions =
         new GrpcRetryer.GrpcRetryerOptions(
             DefaultStubServiceOperationRpcRetryOptions.INSTANCE, null);
-    this.slotSupplier = slotSupplier;
-    this.slotSupplier.setMetricsScope(this.workerMetricsScope);
+    this.executorSlots = options.getTaskExecutorThreadPoolSize();
+    this.executorSlotsSemaphore = new Semaphore(executorSlots);
   }
 
   @Override
@@ -102,7 +101,8 @@ public boolean start() {
               options.getIdentity(),
               new TaskHandlerImpl(handler),
               pollerOptions,
-              slotSupplier.maximumSlots(),
+              options.getTaskExecutorThreadPoolSize(),
+              workerMetricsScope,
               true);
       poller =
           new Poller<>(
@@ -115,7 +115,7 @@ public boolean start() {
                   options.getBuildId(),
                   options.isUsingBuildIdForVersioning(),
                   taskQueueActivitiesPerSecond,
-                  this.slotSupplier,
+                  executorSlotsSemaphore,
                   workerMetricsScope,
                   service.getServerCapabilities()),
               this.pollTaskExecutor,
@@ -131,14 +131,14 @@ public boolean start() {
 
   @Override
   public CompletableFuture<Void> shutdown(ShutdownManager shutdownManager, boolean interruptTasks) {
-    String supplierName = this + "#executorSlots";
+    String semaphoreName = this + "#executorSlotsSemaphore";
     return poller
         .shutdown(shutdownManager, interruptTasks)
         .thenCompose(
             ignore ->
                 !interruptTasks
-                    ? shutdownManager.waitForSupplierPermitsReleasedUnlimited(
-                        slotSupplier, supplierName)
+                    ? shutdownManager.waitForSemaphorePermitsReleaseUntimed(
+                        executorSlotsSemaphore, executorSlots, semaphoreName)
                     : CompletableFuture.completedFuture(null))
         .thenCompose(
             ignore ->
@@ -416,33 +416,23 @@ private void logExceptionDuringResultReporting(
 
   private final class EagerActivityDispatcherImpl implements EagerActivityDispatcher {
     @Override
-    public Optional<SlotPermit> tryReserveActivitySlot(
+    public boolean tryReserveActivitySlot(
         ScheduleActivityTaskCommandAttributesOrBuilder commandAttributes) {
-      if (!WorkerLifecycleState.ACTIVE.equals(ActivityWorker.this.getLifecycleState())
-          || !Objects.equals(
-              commandAttributes.getTaskQueue().getName(), ActivityWorker.this.taskQueue)) {
-        return Optional.empty();
-      }
-      return ActivityWorker.this.slotSupplier.tryReserveSlot(
-          new SlotReservationData(
-              ActivityWorker.this.taskQueue, options.getIdentity(), options.getBuildId()));
+      return WorkerLifecycleState.ACTIVE.equals(ActivityWorker.this.getLifecycleState())
+          && Objects.equals(
+              commandAttributes.getTaskQueue().getName(), ActivityWorker.this.taskQueue)
+          && ActivityWorker.this.executorSlotsSemaphore.tryAcquire();
     }
 
     @Override
-    public void releaseActivitySlotReservations(Iterable<SlotPermit> permits) {
-      for (SlotPermit permit : permits) {
-        ActivityWorker.this.slotSupplier.releaseSlot(SlotReleaseReason.neverUsed(), permit);
-      }
+    public void releaseActivitySlotReservations(int slotCounts) {
+      ActivityWorker.this.executorSlotsSemaphore.release(slotCounts);
     }
 
     @Override
-    public void dispatchActivity(PollActivityTaskQueueResponse activity, SlotPermit permit) {
+    public void dispatchActivity(PollActivityTaskQueueResponse activity) {
       ActivityWorker.this.pollTaskExecutor.process(
-          new ActivityTask(
-              activity,
-              () ->
-                  ActivityWorker.this.slotSupplier.releaseSlot(
-                      SlotReleaseReason.taskComplete(), permit)));
+          new ActivityTask(activity, ActivityWorker.this.executorSlotsSemaphore::release));
     }
   }
 }

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

@@ -22,33 +22,30 @@
 
 import io.temporal.api.command.v1.ScheduleActivityTaskCommandAttributesOrBuilder;
 import io.temporal.api.workflowservice.v1.PollActivityTaskQueueResponse;
-import io.temporal.worker.tuning.SlotPermit;
-import java.util.Optional;
 
 public interface EagerActivityDispatcher {
-  Optional<SlotPermit> tryReserveActivitySlot(
-      ScheduleActivityTaskCommandAttributesOrBuilder commandAttributes);
+  boolean tryReserveActivitySlot(ScheduleActivityTaskCommandAttributesOrBuilder commandAttributes);
 
-  void releaseActivitySlotReservations(Iterable<SlotPermit> permits);
+  void releaseActivitySlotReservations(int slotCounts);
 
-  void dispatchActivity(PollActivityTaskQueueResponse activity, SlotPermit permit);
+  void dispatchActivity(PollActivityTaskQueueResponse activity);
 
   class NoopEagerActivityDispatcher implements EagerActivityDispatcher {
     @Override
-    public Optional<SlotPermit> tryReserveActivitySlot(
+    public boolean tryReserveActivitySlot(
         ScheduleActivityTaskCommandAttributesOrBuilder commandAttributes) {
-      return Optional.empty();
+      return false;
     }
 
     @Override
-    public void releaseActivitySlotReservations(Iterable<SlotPermit> permits) {
-      if (permits.iterator().hasNext())
+    public void releaseActivitySlotReservations(int slotCounts) {
+      if (slotCounts > 0)
         throw new IllegalStateException(
             "Trying to release activity slots on a NoopEagerActivityDispatcher");
     }
 
     @Override
-    public void dispatchActivity(PollActivityTaskQueueResponse activity, SlotPermit permit) {
+    public void dispatchActivity(PollActivityTaskQueueResponse activity) {
       throw new IllegalStateException(
           "Trying to dispatch activity on a NoopEagerActivityDispatcher");
     }