diff --git a/runtime/planner/BUILD.bazel b/runtime/planner/BUILD.bazel index 860d413a0..0a4ef8a84 100644 --- a/runtime/planner/BUILD.bazel +++ b/runtime/planner/BUILD.bazel @@ -21,3 +21,17 @@ java_library( visibility = ["//:internal"], exports = ["//runtime/src/main/java/dev/cel/runtime/planner:planned_program"], ) + +java_library( + name = "async_gate", + testonly = 1, + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate"], +) + +java_library( + name = "async_completion_coordinator", + testonly = 1, + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator"], +) diff --git a/runtime/src/main/java/dev/cel/runtime/planner/AsyncCompletionCoordinator.java b/runtime/src/main/java/dev/cel/runtime/planner/AsyncCompletionCoordinator.java new file mode 100644 index 000000000..1b6f8bc5d --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/planner/AsyncCompletionCoordinator.java @@ -0,0 +1,513 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License 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 dev.cel.runtime.planner; + +import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkState; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import javax.annotation.concurrent.ThreadSafe; +import com.google.errorprone.annotations.concurrent.GuardedBy; +import dev.cel.runtime.CelAsyncCall; +import dev.cel.runtime.CelAsyncDrainAction; +import dev.cel.runtime.CelAsyncDrainStrategy; +import dev.cel.runtime.CelAsyncEvaluationOptions; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.function.Consumer; +import org.jspecify.annotations.Nullable; + +/** + * Coordinates asynchronous call completion notifications, debouncing, and re-evaluation dispatch. + */ +@ThreadSafe +final class AsyncCompletionCoordinator { + + /** Represents the result of attempting to wait for asynchronous completions. */ + enum WaitResult { + /** Continuation registered; execution will resume asynchronously when work arrives. */ + REGISTERED, + /** Drain strategy satisfied immediately; caller should reevaluate now via loop trampoline. */ + REEVALUATE_NOW, + /** No calls in flight and no completions pending; evaluation cannot make further progress. */ + NO_OUTSTANDING_WORK, + /** The coordinator has been cancelled. */ + CANCELLED + } + + // CEL-Internal-4 + private final Object lock; + + private final CelAsyncEvaluationOptions options; + private final AsyncGate gate; + private final Executor continuationExecutor; + + // CEL-Internal-4 + private final Consumer failureCallback; + + @GuardedBy("lock") + private final List completedBatch; + + @GuardedBy("lock") + private int inFlightCount; + + @GuardedBy("lock") + private boolean isWaiting; + + @GuardedBy("lock") + private boolean isCancelled; + + @GuardedBy("lock") + private @Nullable Runnable continuation; + + @GuardedBy("lock") + private @Nullable ScheduledFuture debounceTimer; + + private final ThreadLocal> continuationTrampoline; + + @GuardedBy("lock") + private long cycleId; + + @GuardedBy("lock") + private long debounceGeneration; + + @GuardedBy("lock") + private boolean failureReported; + + static AsyncCompletionCoordinator create( + CelAsyncEvaluationOptions options, + AsyncGate gate, + Executor continuationExecutor, + Consumer failureCallback) { + return new AsyncCompletionCoordinator(options, gate, continuationExecutor, failureCallback); + } + + /** Notifies the coordinator that an asynchronous call has been initiated. */ + void callStarted() { + synchronized (lock) { + if (!isCancelled) { + inFlightCount++; + } + } + } + + /** + * Notifies the coordinator that an asynchronous call has finished. + * + *

Decrements in-flight tracking, appends the call to the current batch, and evaluates the + * configured {@link CelAsyncDrainStrategy} if currently waiting. + */ + void callCompleted(CelAsyncCall call) { + checkNotNull(call, "call must not be null"); + ImmutableList batchSnapshot; + int inFlightSnapshot; + long currentCycleId; + long currentGen; + + synchronized (lock) { + if (isCancelled) { + return; + } + checkState(inFlightCount > 0, "callCompleted called with no calls in flight"); + completedBatch.add(call); + inFlightCount--; + if (!isWaiting) { + return; + } + batchSnapshot = ImmutableList.copyOf(completedBatch); + inFlightSnapshot = inFlightCount; + currentCycleId = cycleId; + currentGen = ++debounceGeneration; + } + + CelAsyncDrainAction action; + try { + action = + checkNotNull( + options.drainStrategy().nextAction(batchSnapshot, inFlightSnapshot), + "drainStrategy must not return null"); + } catch (Throwable t) { + failAndCancel(t); + return; + } + applyDrainAction(action, currentCycleId, currentGen, inFlightSnapshot); + } + + /** + * Waits for pending asynchronous completions or triggers immediate re-evaluation. + * + * @param continuationCallback callback invoked when the drain strategy allows re-evaluation. + * @return {@link WaitResult} indicating how the caller should proceed. + */ + WaitResult waitForCompletions(Runnable continuationCallback) { + checkNotNull(continuationCallback, "continuationCallback must not be null"); + ImmutableList batchSnapshot; + int inFlightSnapshot; + long currentCycleId; + long currentGen; + + synchronized (lock) { + if (isCancelled) { + return WaitResult.CANCELLED; + } + checkState(!isWaiting, "Coordinator is already waiting for completions"); + + if (inFlightCount == 0 && completedBatch.isEmpty()) { + return WaitResult.NO_OUTSTANDING_WORK; + } + + this.isWaiting = true; + this.continuation = continuationCallback; + + if (completedBatch.isEmpty()) { + return WaitResult.REGISTERED; + } + + batchSnapshot = ImmutableList.copyOf(completedBatch); + inFlightSnapshot = inFlightCount; + currentCycleId = this.cycleId; + currentGen = ++this.debounceGeneration; + } + + CelAsyncDrainAction action; + try { + action = + checkNotNull( + options.drainStrategy().nextAction(batchSnapshot, inFlightSnapshot), + "drainStrategy must not return null"); + } catch (Throwable t) { + failAndCancel(t); + return WaitResult.CANCELLED; + } + + ScheduledFuture timerToCancel = null; + boolean reevaluateNow = false; + synchronized (lock) { + if (isCancelled) { + return WaitResult.CANCELLED; + } + if (this.cycleId != currentCycleId) { + return WaitResult.REGISTERED; + } + + if (this.debounceGeneration == currentGen) { + if (action.shouldReevaluate() || inFlightCount == 0) { + timerToCancel = cancelDebounceTimerUnderLock(); + drainAndResetUnderLock(); + reevaluateNow = true; + } + } else { + return WaitResult.REGISTERED; + } + } + if (timerToCancel != null) { + timerToCancel.cancel(false); + } + if (reevaluateNow) { + return WaitResult.REEVALUATE_NOW; + } + + Duration waitDuration = action.waitDuration(); + if (!waitDuration.isZero()) { + scheduleDebounce(waitDuration.toNanos(), currentCycleId, currentGen); + } else { + ScheduledFuture timer = null; + synchronized (lock) { + if (!isCancelled + && isWaiting + && this.cycleId == currentCycleId + && this.debounceGeneration == currentGen) { + timer = cancelDebounceTimerUnderLock(); + } + } + if (timer != null) { + timer.cancel(false); + } + } + return WaitResult.REGISTERED; + } + + private void applyDrainAction( + CelAsyncDrainAction action, long expectedCycleId, long expectedGen, int inFlightSnapshot) { + if (action.shouldReevaluate() || inFlightSnapshot == 0) { + Runnable toRun = null; + ScheduledFuture timerToCancel = null; + synchronized (lock) { + if (!isCancelled + && isWaiting + && this.cycleId == expectedCycleId + && this.debounceGeneration == expectedGen) { + timerToCancel = cancelDebounceTimerUnderLock(); + toRun = drainAndResetUnderLock(); + } + } + if (timerToCancel != null) { + timerToCancel.cancel(false); + } + if (toRun != null) { + dispatchContinuation(toRun); + } + return; + } + + Duration waitDuration = action.waitDuration(); + if (!waitDuration.isZero()) { + scheduleDebounce(waitDuration.toNanos(), expectedCycleId, expectedGen); + } else { + ScheduledFuture timerToCancel = null; + synchronized (lock) { + if (!isCancelled + && isWaiting + && this.cycleId == expectedCycleId + && this.debounceGeneration == expectedGen) { + timerToCancel = cancelDebounceTimerUnderLock(); + } + } + if (timerToCancel != null) { + timerToCancel.cancel(false); + } + } + } + + private void scheduleDebounce(long nanos, long scheduledCycleId, long scheduledGen) { + ScheduledExecutorService scheduler; + try { + scheduler = options.resolveScheduledExecutorService(); + } catch (Throwable t) { + failAndCancel(t); + return; + } + + ScheduledFuture future; + try { + future = + scheduler.schedule( + () -> onDebounceFired(scheduledCycleId, scheduledGen), nanos, NANOSECONDS); + } catch (Throwable t) { + failAndCancel(t); + return; + } + + ScheduledFuture redundantFuture = null; + synchronized (lock) { + if (!isCancelled + && isWaiting + && this.cycleId == scheduledCycleId + && this.debounceGeneration == scheduledGen) { + if (debounceTimer != null) { + redundantFuture = debounceTimer; + } + debounceTimer = future; + } else { + redundantFuture = future; + } + } + if (redundantFuture != null) { + redundantFuture.cancel(false); + } + } + + void onDebounceFired(long firedCycleId, long firedGen) { + Runnable toRun = null; + ScheduledFuture timerToCancel = null; + synchronized (lock) { + if (!isCancelled + && isWaiting + && this.cycleId == firedCycleId + && this.debounceGeneration == firedGen) { + timerToCancel = cancelDebounceTimerUnderLock(); + toRun = drainAndResetUnderLock(); + } + } + if (timerToCancel != null) { + timerToCancel.cancel(false); + } + if (toRun != null) { + dispatchContinuation(toRun); + } + } + + private void dispatchContinuation(@Nullable Runnable run) { + if (run == null) { + return; + } + Deque queue = continuationTrampoline.get(); + queue.add(run); + if (queue.size() > 1) { + return; + } + try { + while (!queue.isEmpty()) { + Runnable next = queue.peek(); + try { + continuationExecutor.execute(next); + } catch (Throwable t) { + failAndCancel(t); + } finally { + queue.poll(); + } + } + } finally { + queue.clear(); + continuationTrampoline.remove(); + } + } + + /** + * Cancels the coordinator and associated concurrency gate. + * + *

Any registered continuation callback is discarded without being executed. The caller or + * owner of this coordinator is responsible for completing or failing the outer evaluation future + * itself; calling {@code cancel()} does not notify the continuation callback. + */ + void cancel() { + ScheduledFuture timerToCancel; + synchronized (lock) { + if (isCancelled) { + return; + } + isCancelled = true; + isWaiting = false; + continuation = null; + completedBatch.clear(); + inFlightCount = 0; + timerToCancel = cancelDebounceTimerUnderLock(); + } + if (timerToCancel != null) { + timerToCancel.cancel(false); + } + gate.cancel(); + } + + private void failAndCancel(Throwable t) { + synchronized (lock) { + if (failureReported) { + return; + } + failureReported = true; + } + cancel(); + failureCallback.accept(t); + } + + @GuardedBy("lock") + private @Nullable Runnable drainAndResetUnderLock() { + cycleId++; + debounceGeneration++; + isWaiting = false; + completedBatch.clear(); + Runnable run = continuation; + continuation = null; + return run; + } + + @GuardedBy("lock") + private @Nullable ScheduledFuture cancelDebounceTimerUnderLock() { + ScheduledFuture timer = debounceTimer; + debounceTimer = null; + return timer; + } + + @VisibleForTesting + boolean hasPendingBatch() { + synchronized (lock) { + return !completedBatch.isEmpty(); + } + } + + @VisibleForTesting + int inFlightCount() { + synchronized (lock) { + return inFlightCount; + } + } + + @VisibleForTesting + boolean isWaiting() { + synchronized (lock) { + return isWaiting; + } + } + + @VisibleForTesting + boolean hasContinuation() { + synchronized (lock) { + return continuation != null; + } + } + + @VisibleForTesting + boolean hasScheduledDebounceTimer() { + synchronized (lock) { + return debounceTimer != null; + } + } + + @VisibleForTesting + long cycleId() { + synchronized (lock) { + return cycleId; + } + } + + @VisibleForTesting + long debounceGeneration() { + synchronized (lock) { + return debounceGeneration; + } + } + + @VisibleForTesting + boolean isCancelled() { + synchronized (lock) { + return isCancelled; + } + } + + private AsyncCompletionCoordinator( + CelAsyncEvaluationOptions options, + AsyncGate gate, + Executor continuationExecutor, + Consumer failureCallback) { + this.options = checkNotNull(options, "options must not be null"); + this.gate = checkNotNull(gate, "gate must not be null"); + this.continuationExecutor = + checkNotNull(continuationExecutor, "continuationExecutor must not be null"); + this.failureCallback = checkNotNull(failureCallback, "failureCallback must not be null"); + this.lock = new Object(); + this.completedBatch = new ArrayList<>(); + this.continuationTrampoline = + new ThreadLocal>() { + @Override + protected Deque initialValue() { + return new ArrayDeque<>(); + } + }; + this.isWaiting = false; + this.isCancelled = false; + this.failureReported = false; + this.cycleId = 0; + this.debounceGeneration = 0; + this.inFlightCount = 0; + } +} diff --git a/runtime/src/main/java/dev/cel/runtime/planner/AsyncGate.java b/runtime/src/main/java/dev/cel/runtime/planner/AsyncGate.java new file mode 100644 index 000000000..d284e5286 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/planner/AsyncGate.java @@ -0,0 +1,112 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License 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 dev.cel.runtime.planner; + +import com.google.common.annotations.VisibleForTesting; +import com.google.errorprone.annotations.CheckReturnValue; +import javax.annotation.concurrent.ThreadSafe; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.jspecify.annotations.Nullable; + +/** + * Regulates the number of concurrent asynchronous function executions based on maxConcurrency. + * + *

A {@code maxConcurrency} value of {@code 0} or less represents unbounded concurrency (no limit + * on concurrent executions). + */ +@ThreadSafe +final class AsyncGate { + + /** Null when {@code maxConcurrency <= 0}, indicating unbounded concurrency (no throttling). */ + private final @Nullable Semaphore semaphore; + + private final AtomicInteger activeCalls; + private final AtomicBoolean cancelled; + + /** + * Creates an {@link AsyncGate} regulating concurrent asynchronous calls. + * + * @param maxConcurrency the maximum number of concurrent executions allowed. A value of {@code 0} + * or less indicates unbounded concurrency (no concurrency limit). + */ + static AsyncGate create(int maxConcurrency) { + return new AsyncGate(maxConcurrency); + } + + /** + * Attempts to acquire a concurrency slot for an asynchronous call. + * + * @return true if a slot was acquired; false if the gate is cancelled or at maximum concurrency. + */ + @CheckReturnValue + boolean tryAcquire() { + if (semaphore != null && !semaphore.tryAcquire()) { + return false; + } + if (cancelled.get()) { + if (semaphore != null) { + semaphore.release(); + } + return false; + } + activeCalls.incrementAndGet(); + return true; + } + + /** Releases a previously acquired concurrency slot and decrements the active call count. */ + void release() { + while (true) { + int current = activeCalls.get(); + if (current <= 0) { + return; + } + if (activeCalls.compareAndSet(current, current - 1)) { + if (semaphore != null) { + semaphore.release(); + } + return; + } + } + } + + /** Cancels the gate, preventing any future calls from acquiring permits. */ + void cancel() { + cancelled.set(true); + } + + /** Returns true if the gate has been cancelled. */ + boolean isCancelled() { + return cancelled.get(); + } + + /** Returns the current number of active in-flight calls. */ + int activeCount() { + return activeCalls.get(); + } + + /** Returns the number of currently available permits, or -1 if unbounded. */ + @VisibleForTesting + int availablePermits() { + return semaphore != null ? semaphore.availablePermits() : -1; + } + + private AsyncGate(int maxConcurrency) { + this.semaphore = maxConcurrency > 0 ? new Semaphore(maxConcurrency) : null; + this.activeCalls = new AtomicInteger(); + this.cancelled = new AtomicBoolean(false); + } +} diff --git a/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel b/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel index d4dbb1659..d838e8d53 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel +++ b/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel @@ -187,6 +187,36 @@ java_library( ], ) +java_library( + name = "async_gate", + srcs = ["AsyncGate.java"], + tags = [ + ], + deps = [ + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", + "@maven//:org_jspecify_jspecify", + ], +) + +java_library( + name = "async_completion_coordinator", + srcs = ["AsyncCompletionCoordinator.java"], + tags = [ + ], + deps = [ + ":async_gate", + "//runtime:async_call", + "//runtime:async_drain_strategy", + "//runtime:async_options", + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", + "@maven//:org_jspecify_jspecify", + ], +) + java_library( name = "activation_wrapper", srcs = ["ActivationWrapper.java"], diff --git a/runtime/src/test/java/dev/cel/runtime/planner/AsyncCompletionCoordinatorTest.java b/runtime/src/test/java/dev/cel/runtime/planner/AsyncCompletionCoordinatorTest.java new file mode 100644 index 000000000..4539c7f5b --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/planner/AsyncCompletionCoordinatorTest.java @@ -0,0 +1,1137 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License 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 dev.cel.runtime.planner; + +import static com.google.common.truth.Truth.assertThat; +import static java.util.Objects.requireNonNull; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertThrows; + +import dev.cel.runtime.CelAsyncCall; +import dev.cel.runtime.CelAsyncDrainAction; +import dev.cel.runtime.CelAsyncDrainStrategy; +import dev.cel.runtime.CelAsyncEvaluationOptions; +import dev.cel.runtime.planner.AsyncCompletionCoordinator.WaitResult; +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public final class AsyncCompletionCoordinatorTest { + + private static final CelAsyncCall DUMMY_CALL = + new CelAsyncCall() { + @Override + public long callId() { + return 1L; + } + + @Override + public long exprId() { + return 10L; + } + + @Override + public String functionName() { + return "testFn"; + } + + @Override + public String overloadId() { + return "testFn_overload"; + } + }; + + @Test + public void waitForCompletions_whenNoCallsInFlightAndEmptyBatch_returnsNoOutstandingWork() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + WaitResult result = coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(result).isEqualTo(WaitResult.NO_OUTSTANDING_WORK); + assertThat(continuationRan.get()).isFalse(); + assertThat(coordinator.isWaiting()).isFalse(); + } + + @Test + public void waitForCompletions_whenCallsInFlightAndEmptyBatch_returnsRegistered() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + WaitResult result = coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(result).isEqualTo(WaitResult.REGISTERED); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(continuationRan.get()).isFalse(); + } + + @Test + public void + waitForCompletions_whenDrainStrategySatisfiedImmediately_returnsReevaluateNowWithoutDispatch() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainAll()) + .build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callCompleted(DUMMY_CALL); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + WaitResult result = coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(result).isEqualTo(WaitResult.REEVALUATE_NOW); + assertThat(continuationRan.get()).isFalse(); + assertThat(coordinator.isWaiting()).isFalse(); + assertThat(coordinator.hasPendingBatch()).isFalse(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + } + + @Test + public void waitForCompletions_whenDebounceRequested_schedulesTimerAndReturnsRegistered() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + AsyncGate gate = AsyncGate.create(2); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.callCompleted(DUMMY_CALL); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + WaitResult result = coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(result).isEqualTo(WaitResult.REGISTERED); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isTrue(); + assertThat(continuationRan.get()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void waitForCompletions_whenAlreadyWaiting_throwsIllegalStateException() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + IllegalStateException thrown = + assertThrows(IllegalStateException.class, () -> coordinator.waitForCompletions(() -> {})); + + assertThat(thrown).hasMessageThat().contains("Coordinator is already waiting for completions"); + } + + @Test + public void waitForCompletions_whenCoordinatorCancelled_returnsCancelled() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.cancel(); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + WaitResult result = coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(result).isEqualTo(WaitResult.CANCELLED); + assertThat(continuationRan.get()).isFalse(); + } + + @Test + public void callStarted_incrementsInFlightCount() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + + coordinator.callStarted(); + coordinator.callStarted(); + + assertThat(coordinator.inFlightCount()).isEqualTo(2); + } + + @Test + public void callCompleted_decrementsInFlightCountAndAddsToBatch() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(coordinator.inFlightCount()).isEqualTo(1); + assertThat(coordinator.hasPendingBatch()).isTrue(); + } + + @Test + public void callCompleted_whenCancelled_ignoresCall() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.cancel(); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(coordinator.inFlightCount()).isEqualTo(0); + assertThat(coordinator.hasPendingBatch()).isFalse(); + } + + @Test + public void callCompleted_whenWaitingWithPendingCalls_schedulesDebounceTimer() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(continuationRan.get()).isFalse(); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isTrue(); + assertThat(scheduler.getQueue()).isNotEmpty(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void callCompleted_whenWaitingWithDrainAllStrategy_waitsWhileCallsRemainInFlight() { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainAll()) + .build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(continuationRan.get()).isFalse(); + assertThat(coordinator.isWaiting()).isTrue(); + } + + @Test + public void + callCompleted_whenWaitingWithDrainAllStrategy_triggersContinuationWhenFinalCallCompletes() { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainAll()) + .build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + coordinator.callCompleted(DUMMY_CALL); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(continuationRan.get()).isTrue(); + assertThat(coordinator.isWaiting()).isFalse(); + } + + @Test + public void callCompleted_whenDebounceTimerPending_resetsDebounceTimerForSlidingWindow() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(3); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + coordinator.callCompleted(DUMMY_CALL); + ScheduledFuture firstTimer = (ScheduledFuture) scheduler.getQueue().peek(); + coordinator.callCompleted(DUMMY_CALL); + + assertThat(firstTimer).isNotNull(); + assertThat(firstTimer.isCancelled()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void onDebounceFired_whenWaiting_triggersContinuation() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + coordinator.callCompleted(DUMMY_CALL); + ScheduledFuture scheduledTask = (ScheduledFuture) scheduler.getQueue().peek(); + + assertThat(scheduledTask).isNotNull(); + ((Runnable) scheduledTask).run(); + + assertThat(continuationRan.get()).isTrue(); + assertThat(coordinator.isWaiting()).isFalse(); + assertThat(coordinator.hasPendingBatch()).isFalse(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + assertThat(coordinator.hasContinuation()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void onDebounceFired_whenCycleMismatch_doesNotExecuteContinuation() { + AtomicInteger executedCount = new AtomicInteger(); + Executor rejectingNullExecutor = + task -> { + requireNonNull(task, "task must not be null"); + executedCount.incrementAndGet(); + task.run(); + }; + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = AsyncGate.create(1); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, rejectingNullExecutor, t -> {}); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + coordinator.onDebounceFired(coordinator.cycleId() - 1, coordinator.debounceGeneration()); + + assertThat(executedCount.get()).isEqualTo(0); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasContinuation()).isTrue(); + } + + @Test + public void onDebounceFired_whenDebounceGenerationMismatch_doesNotExecuteContinuation() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(3); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.callStarted(); + AtomicInteger continuationRan = new AtomicInteger(); + coordinator.waitForCompletions(continuationRan::incrementAndGet); + coordinator.callCompleted(DUMMY_CALL); + long staleGen = coordinator.debounceGeneration(); + coordinator.callCompleted(DUMMY_CALL); + + coordinator.onDebounceFired(coordinator.cycleId(), staleGen); + + assertThat(continuationRan.get()).isEqualTo(0); + assertThat(coordinator.isWaiting()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void cancel_cancelsDebounceTimerAndPreventsContinuation() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + coordinator.callCompleted(DUMMY_CALL); + ScheduledFuture scheduledTask = (ScheduledFuture) scheduler.getQueue().peek(); + + coordinator.cancel(); + + assertThat(coordinator.hasPendingBatch()).isFalse(); + assertThat(coordinator.isWaiting()).isFalse(); + assertThat(coordinator.hasContinuation()).isFalse(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + assertThat(scheduledTask).isNotNull(); + assertThat(scheduledTask.isCancelled()).isTrue(); + assertThat(continuationRan.get()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void onDebounceFired_whenCancelled_doesNotTriggerContinuation() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + coordinator.callCompleted(DUMMY_CALL); + ScheduledFuture scheduledTask = (ScheduledFuture) scheduler.getQueue().peek(); + coordinator.cancel(); + + assertThat(scheduledTask).isNotNull(); + ((Runnable) scheduledTask).run(); + + assertThat(continuationRan.get()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void cancel_cancelsAssociatedGate() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + + coordinator.cancel(); + + assertThat(gate.isCancelled()).isTrue(); + } + + @Test + public void applyDrainAction_whenInFlightZeroAndStrategyWaits_forcesReevaluation() { + CelAsyncDrainStrategy alwaysWaitStrategy = (batch, active) -> CelAsyncDrainAction.waitForMore(); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setDrainStrategy(alwaysWaitStrategy).build(); + AsyncGate gate = AsyncGate.create(1); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(continuationRan.get()).isTrue(); + assertThat(coordinator.isWaiting()).isFalse(); + } + + @Test + public void dispatchContinuation_whenExecutorThrows_invokesFailureCallback() { + Executor rejectingExecutor = + r -> { + throw new RejectedExecutionException("rejected"); + }; + AtomicReference failure = new AtomicReference<>(); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = AsyncGate.create(1); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, rejectingExecutor, failure::set); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(failure.get()).isInstanceOf(RejectedExecutionException.class); + } + + @Test + public void scheduleDebounce_whenSchedulerThrows_invokesFailureCallback() { + ScheduledThreadPoolExecutor rejectingScheduler = + new ScheduledThreadPoolExecutor(1) { + @Override + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + throw new RejectedExecutionException("scheduler rejected"); + } + }; + try { + AtomicReference failure = new AtomicReference<>(); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(rejectingScheduler) + .build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, failure::set); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(failure.get()).isInstanceOf(RejectedExecutionException.class); + } finally { + rejectingScheduler.shutdownNow(); + } + } + + @Test + public void + scheduleDebounce_whenCoordinatorCancelledConcurrently_cancelsScheduledFutureWithoutInterrupt() { + AtomicBoolean cancelledInsideScheduler = new AtomicBoolean(false); + AsyncCompletionCoordinator[] coordinatorHolder = new AsyncCompletionCoordinator[1]; + ScheduledThreadPoolExecutor scheduler = + new ScheduledThreadPoolExecutor(1) { + @Override + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + ScheduledFuture task = super.schedule(command, delay, unit); + if (coordinatorHolder[0] != null && !cancelledInsideScheduler.get()) { + cancelledInsideScheduler.set(true); + coordinatorHolder[0].cancel(); + } + return task; + } + }; + + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinatorHolder[0] = coordinator; + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.callCompleted(DUMMY_CALL); + coordinator.waitForCompletions(() -> {}); + + ScheduledFuture scheduledTask = (ScheduledFuture) scheduler.getQueue().peek(); + + assertThat(scheduledTask).isNotNull(); + assertThat(scheduledTask.isCancelled()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void multiThreadedConcurrentCompletions_retainsSingleContinuationDispatch() + throws Exception { + int workerCount = 10; + ExecutorService workers = Executors.newFixedThreadPool(workerCount); + ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainAll()) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(workerCount); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, workers, t -> {}); + for (int i = 0; i < workerCount; i++) { + coordinator.callStarted(); + } + AtomicInteger continuationDispatches = new AtomicInteger(); + CountDownLatch continuationLatch = new CountDownLatch(1); + CountDownLatch readyLatch = new CountDownLatch(workerCount); + CountDownLatch startLatch = new CountDownLatch(1); + + coordinator.waitForCompletions( + () -> { + continuationDispatches.incrementAndGet(); + continuationLatch.countDown(); + }); + + for (int i = 0; i < workerCount; i++) { + workers.execute( + () -> { + readyLatch.countDown(); + try { + startLatch.await(); + coordinator.callCompleted(DUMMY_CALL); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + } + + readyLatch.await(5, SECONDS); + startLatch.countDown(); + boolean continuationReached = continuationLatch.await(5, SECONDS); + workers.shutdown(); + boolean workersTerminated = workers.awaitTermination(5, SECONDS); + + assertThat(continuationReached).isTrue(); + assertThat(workersTerminated).isTrue(); + assertThat(continuationDispatches.get()).isEqualTo(1); + assertThat(coordinator.isWaiting()).isFalse(); + } finally { + workers.shutdownNow(); + scheduler.shutdownNow(); + } + } + + @Test + public void create_nullOptions_throwsNullPointerException() { + AsyncGate gate = AsyncGate.create(1); + + NullPointerException thrown = + assertThrows( + NullPointerException.class, + () -> AsyncCompletionCoordinator.create(null, gate, Runnable::run, t -> {})); + + assertThat(thrown).hasMessageThat().contains("options must not be null"); + } + + @Test + public void create_nullGate_throwsNullPointerException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + + NullPointerException thrown = + assertThrows( + NullPointerException.class, + () -> AsyncCompletionCoordinator.create(options, null, Runnable::run, t -> {})); + + assertThat(thrown).hasMessageThat().contains("gate must not be null"); + } + + @Test + public void create_nullExecutor_throwsNullPointerException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = AsyncGate.create(1); + + NullPointerException thrown = + assertThrows( + NullPointerException.class, + () -> AsyncCompletionCoordinator.create(options, gate, null, t -> {})); + + assertThat(thrown).hasMessageThat().contains("continuationExecutor must not be null"); + } + + @Test + public void create_nullFailureCallback_throwsNullPointerException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = AsyncGate.create(1); + + NullPointerException thrown = + assertThrows( + NullPointerException.class, + () -> AsyncCompletionCoordinator.create(options, gate, Runnable::run, null)); + + assertThat(thrown).hasMessageThat().contains("failureCallback must not be null"); + } + + @Test + public void callCompleted_nullCall_throwsNullPointerException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = AsyncGate.create(1); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + + NullPointerException thrown = + assertThrows(NullPointerException.class, () -> coordinator.callCompleted(null)); + + assertThat(thrown).hasMessageThat().contains("call must not be null"); + } + + @Test + public void waitForCompletions_nullContinuation_throwsNullPointerException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = AsyncGate.create(1); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + + NullPointerException thrown = + assertThrows(NullPointerException.class, () -> coordinator.waitForCompletions(null)); + + assertThat(thrown).hasMessageThat().contains("continuationCallback must not be null"); + } + + @Test + public void staleTimerFromPreviousPass_doesNotTriggerContinuationOnSubsequentPass() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.callStarted(); + AtomicInteger pass1Count = new AtomicInteger(); + coordinator.waitForCompletions(pass1Count::incrementAndGet); + coordinator.callCompleted(DUMMY_CALL); + ScheduledFuture pass1Timer = (ScheduledFuture) scheduler.getQueue().peek(); + coordinator.callCompleted(DUMMY_CALL); + coordinator.callStarted(); + AtomicInteger pass2Count = new AtomicInteger(); + coordinator.waitForCompletions(pass2Count::incrementAndGet); + + assertThat(pass1Timer).isNotNull(); + ((Runnable) pass1Timer).run(); + + assertThat(pass2Count.get()).isEqualTo(0); + assertThat(coordinator.isWaiting()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void drainAndReset_incrementsCycleIdAndClearsContinuation() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = AsyncGate.create(1); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + long initialCycleId = coordinator.cycleId(); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(coordinator.cycleId()).isGreaterThan(initialCycleId); + assertThat(coordinator.hasContinuation()).isFalse(); + } + + @Test + public void + waitForCompletions_lastCallCompletesDuringDrainStrategyEvaluation_executesContinuationAndReturnsRegistered() { + AtomicReference coordinatorRef = new AtomicReference<>(); + CelAsyncDrainStrategy racingStrategy = new RacingDrainStrategy(coordinatorRef); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setDrainStrategy(racingStrategy).build(); + AsyncGate gate = AsyncGate.create(2); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinatorRef.set(coordinator); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.callCompleted(DUMMY_CALL); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + WaitResult result = coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(result).isEqualTo(WaitResult.REGISTERED); + assertThat(continuationRan.get()).isTrue(); + assertThat(coordinator.isWaiting()).isFalse(); + } + + @Test + public void dispatchContinuation_directExecutorReentrantCompletions_doesNotCauseStackOverflow() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = AsyncGate.create(1); + AtomicInteger step = new AtomicInteger(); + int targetSteps = 1000; + AtomicReference coordinatorRef = new AtomicReference<>(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinatorRef.set(coordinator); + coordinator.callStarted(); + coordinator.waitForCompletions( + new Runnable() { + @Override + public void run() { + if (step.incrementAndGet() < targetSteps) { + coordinatorRef.get().callStarted(); + coordinatorRef.get().waitForCompletions(this); + coordinatorRef.get().callCompleted(DUMMY_CALL); + } + } + }); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(step.get()).isEqualTo(targetSteps); + } + + @Test + public void dispatchContinuation_nestedCoordinatorsOnSameThread_doesNotHijackExecutor() { + AtomicBoolean coordinator2ExecutorUsed = new AtomicBoolean(false); + AsyncGate gate1 = AsyncGate.create(1); + AsyncGate gate2 = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator1 = + AsyncCompletionCoordinator.create(options, gate1, Runnable::run, t -> {}); + AsyncCompletionCoordinator coordinator2 = + AsyncCompletionCoordinator.create( + options, + gate2, + task -> { + coordinator2ExecutorUsed.set(true); + task.run(); + }, + t -> {}); + coordinator1.callStarted(); + coordinator2.callStarted(); + coordinator1.waitForCompletions( + () -> { + coordinator2.waitForCompletions(() -> {}); + coordinator2.callCompleted(DUMMY_CALL); + }); + + coordinator1.callCompleted(DUMMY_CALL); + + assertThat(coordinator2ExecutorUsed.get()).isTrue(); + } + + @Test + public void callStarted_whenCancelled_doesNotIncrementInFlightCount() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinator.cancel(); + + coordinator.callStarted(); + + assertThat(coordinator.inFlightCount()).isEqualTo(0); + } + + @Test + public void callCompleted_withoutPriorCallStarted_throwsIllegalStateException() { + AsyncGate gate = AsyncGate.create(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + + IllegalStateException thrown = + assertThrows(IllegalStateException.class, () -> coordinator.callCompleted(DUMMY_CALL)); + + assertThat(thrown).hasMessageThat().contains("callCompleted called with no calls in flight"); + } + + @Test + public void callCompleted_whenDrainStrategyThrows_invokesFailureCallbackAndCancels() { + RuntimeException failure = new RuntimeException("strategy failed"); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(new FailingDrainStrategy(failure)) + .build(); + AsyncGate gate = AsyncGate.create(1); + AtomicReference capturedFailure = new AtomicReference<>(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, capturedFailure::set); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(capturedFailure.get()).isSameInstanceAs(failure); + assertThat(coordinator.isCancelled()).isTrue(); + assertThat(gate.isCancelled()).isTrue(); + } + + @Test + public void waitForCompletions_whenDrainStrategyThrows_invokesFailureCallbackAndCancels() { + RuntimeException failure = new RuntimeException("strategy failed"); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(new FailingDrainStrategy(failure)) + .build(); + AsyncGate gate = AsyncGate.create(1); + AtomicReference capturedFailure = new AtomicReference<>(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, capturedFailure::set); + coordinator.callStarted(); + coordinator.callCompleted(DUMMY_CALL); + coordinator.callStarted(); + + WaitResult result = coordinator.waitForCompletions(() -> {}); + + assertThat(result).isEqualTo(WaitResult.CANCELLED); + assertThat(capturedFailure.get()).isSameInstanceAs(failure); + assertThat(coordinator.isCancelled()).isTrue(); + } + + @Test + public void scheduleDebounce_whenSchedulerThrows_cancelsCoordinator() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + scheduler.shutdown(); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(2); + AtomicReference capturedFailure = new AtomicReference<>(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, capturedFailure::set); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(capturedFailure.get()).isInstanceOf(RejectedExecutionException.class); + assertThat(coordinator.isCancelled()).isTrue(); + assertThat(gate.isCancelled()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void waitForCompletions_whenIntermediateCallArrivesDuringStrategyEval_preservesDebounce() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + AtomicReference coordinatorRef = new AtomicReference<>(); + CelAsyncDrainStrategy racingStrategy = + new SingleShotRacingDrainStrategy( + coordinatorRef, CelAsyncDrainAction.waitDuration(Duration.ofMinutes(5))); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(racingStrategy) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(3); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinatorRef.set(coordinator); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.callCompleted(DUMMY_CALL); + + WaitResult result = coordinator.waitForCompletions(() -> {}); + + assertThat(result).isEqualTo(WaitResult.REGISTERED); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void dispatchContinuation_whenExecutorThrows_cancelsCoordinatorAndNotifiesCallback() { + RejectedExecutionException failure = new RejectedExecutionException("rejected"); + Executor rejectingExecutor = + task -> { + throw failure; + }; + AsyncGate gate = AsyncGate.create(1); + AtomicReference capturedFailure = new AtomicReference<>(); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, rejectingExecutor, capturedFailure::set); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(capturedFailure.get()).isSameInstanceAs(failure); + assertThat(coordinator.isCancelled()).isTrue(); + assertThat(gate.isCancelled()).isTrue(); + } + + @Test + public void failAndCancel_concurrentFailures_notifiesCallbackAtMostOnce() { + RuntimeException failure1 = new RuntimeException("error 1"); + AtomicInteger callbackCount = new AtomicInteger(); + AsyncGate gate = AsyncGate.create(2); + FailingDrainStrategy failingStrategy = new FailingDrainStrategy(failure1); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setDrainStrategy(failingStrategy).build(); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create( + options, gate, Runnable::run, t -> callbackCount.incrementAndGet()); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(callbackCount.get()).isEqualTo(1); + assertThat(coordinator.isCancelled()).isTrue(); + } + + @Test + public void applyDrainAction_whenGenerationStale_discardsStaleAction() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + AtomicReference coordinatorRef = new AtomicReference<>(); + StaleReevaluateRacingDrainStrategy strategy = + new StaleReevaluateRacingDrainStrategy(coordinatorRef); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(strategy) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = AsyncGate.create(3); + AsyncCompletionCoordinator coordinator = + AsyncCompletionCoordinator.create(options, gate, Runnable::run, t -> {}); + coordinatorRef.set(coordinator); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.callStarted(); + coordinator.waitForCompletions(() -> {}); + + coordinator.callCompleted(DUMMY_CALL); + + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + private static final class StaleReevaluateRacingDrainStrategy implements CelAsyncDrainStrategy { + @SuppressWarnings("Immutable") + private final AtomicReference coordinatorRef; + + @SuppressWarnings("Immutable") + private final AtomicBoolean first = new AtomicBoolean(true); + + @Override + public CelAsyncDrainAction nextAction(List batch, int active) { + if (first.compareAndSet(true, false)) { + coordinatorRef.get().callCompleted(DUMMY_CALL); + return CelAsyncDrainAction.reevaluate(); + } + return CelAsyncDrainAction.waitDuration(Duration.ofMinutes(5)); + } + + private StaleReevaluateRacingDrainStrategy( + AtomicReference coordinatorRef) { + this.coordinatorRef = coordinatorRef; + } + } + + private static final class SingleShotRacingDrainStrategy implements CelAsyncDrainStrategy { + @SuppressWarnings("Immutable") + private final AtomicReference coordinatorRef; + + @SuppressWarnings("Immutable") + private final AtomicBoolean completed; + + private final CelAsyncDrainAction returnAction; + + @Override + public CelAsyncDrainAction nextAction(List batch, int active) { + if (completed.compareAndSet(false, true)) { + coordinatorRef.get().callCompleted(DUMMY_CALL); + } + return returnAction; + } + + private SingleShotRacingDrainStrategy( + AtomicReference coordinatorRef, + CelAsyncDrainAction returnAction) { + this.coordinatorRef = coordinatorRef; + this.completed = new AtomicBoolean(false); + this.returnAction = returnAction; + } + } + + private static final class FailingDrainStrategy implements CelAsyncDrainStrategy { + @SuppressWarnings("Immutable") + private final RuntimeException failure; + + @Override + public CelAsyncDrainAction nextAction(List batch, int active) { + throw failure; + } + + private FailingDrainStrategy(RuntimeException failure) { + this.failure = failure; + } + } + + private static final class RacingDrainStrategy implements CelAsyncDrainStrategy { + @SuppressWarnings("Immutable") + private final AtomicReference coordinatorRef; + + @Override + public CelAsyncDrainAction nextAction(List batch, int active) { + if (active > 0) { + coordinatorRef.get().callCompleted(DUMMY_CALL); + } + return CelAsyncDrainAction.waitForMore(); + } + + private RacingDrainStrategy(AtomicReference coordinatorRef) { + this.coordinatorRef = coordinatorRef; + } + } +} diff --git a/runtime/src/test/java/dev/cel/runtime/planner/AsyncGateTest.java b/runtime/src/test/java/dev/cel/runtime/planner/AsyncGateTest.java new file mode 100644 index 000000000..857297185 --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/planner/AsyncGateTest.java @@ -0,0 +1,249 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License 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 dev.cel.runtime.planner; + +import static com.google.common.truth.Truth.assertThat; +import static java.util.concurrent.TimeUnit.SECONDS; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public final class AsyncGateTest { + + @Test + public void tryAcquire_withAvailablePermits_returnsTrueAndIncrementsActiveCount() { + AsyncGate gate = AsyncGate.create(2); + + boolean firstAcquired = gate.tryAcquire(); + boolean secondAcquired = gate.tryAcquire(); + + assertThat(firstAcquired).isTrue(); + assertThat(secondAcquired).isTrue(); + assertThat(gate.activeCount()).isEqualTo(2); + assertThat(gate.availablePermits()).isEqualTo(0); + } + + @Test + public void tryAcquire_atMaxConcurrency_returnsFalseAndDoesNotIncrementActiveCount() { + AsyncGate gate = AsyncGate.create(1); + assertThat(gate.tryAcquire()).isTrue(); + + boolean acquired = gate.tryAcquire(); + + assertThat(acquired).isFalse(); + assertThat(gate.activeCount()).isEqualTo(1); + assertThat(gate.availablePermits()).isEqualTo(0); + } + + @Test + public void tryAcquire_unbounded_alwaysSucceeds() { + AsyncGate gate = AsyncGate.create(0); + + boolean first = gate.tryAcquire(); + boolean second = gate.tryAcquire(); + boolean third = gate.tryAcquire(); + + assertThat(first).isTrue(); + assertThat(second).isTrue(); + assertThat(third).isTrue(); + assertThat(gate.activeCount()).isEqualTo(3); + assertThat(gate.availablePermits()).isEqualTo(-1); + } + + @Test + public void tryAcquire_negativeMaxConcurrency_treatedAsUnbounded() { + AsyncGate gate = AsyncGate.create(-1); + + boolean acquired = gate.tryAcquire(); + + assertThat(acquired).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + assertThat(gate.availablePermits()).isEqualTo(-1); + } + + @Test + public void tryAcquire_whenCancelled_returnsFalse() { + AsyncGate gate = AsyncGate.create(2); + gate.cancel(); + + boolean acquired = gate.tryAcquire(); + + assertThat(acquired).isFalse(); + assertThat(gate.activeCount()).isEqualTo(0); + assertThat(gate.availablePermits()).isEqualTo(2); + } + + @Test + public void tryAcquire_unboundedWhenCancelled_returnsFalse() { + AsyncGate gate = AsyncGate.create(0); + gate.cancel(); + + boolean acquired = gate.tryAcquire(); + + assertThat(acquired).isFalse(); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void release_decrementsActiveCountAndFreesPermit() { + AsyncGate gate = AsyncGate.create(2); + assertThat(gate.tryAcquire()).isTrue(); + assertThat(gate.tryAcquire()).isTrue(); + + gate.release(); + + assertThat(gate.activeCount()).isEqualTo(1); + assertThat(gate.availablePermits()).isEqualTo(1); + } + + @Test + public void release_unbounded_decrementsActiveCount() { + AsyncGate gate = AsyncGate.create(0); + assertThat(gate.tryAcquire()).isTrue(); + + gate.release(); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void release_allowsSubsequentTryAcquire() { + AsyncGate gate = AsyncGate.create(1); + assertThat(gate.tryAcquire()).isTrue(); + + gate.release(); + + assertThat(gate.tryAcquire()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + } + + @Test + public void release_withoutPriorAcquire_doesNotExceedMaxConcurrency() { + AsyncGate gate = AsyncGate.create(2); + + gate.release(); + + assertThat(gate.availablePermits()).isEqualTo(2); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void release_withoutPriorAcquire_doesNotUnderflowActiveCount() { + AsyncGate gate = AsyncGate.create(0); + + gate.release(); + gate.release(); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void release_calledMoreThanAcquires_onlyReleasesAcquiredPermits() { + AsyncGate gate = AsyncGate.create(1); + assertThat(gate.tryAcquire()).isTrue(); + + gate.release(); + gate.release(); + + assertThat(gate.availablePermits()).isEqualTo(1); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void cancel_setsIsCancelledToTrue() { + AsyncGate gate = AsyncGate.create(1); + + gate.cancel(); + + assertThat(gate.isCancelled()).isTrue(); + } + + @Test + public void cancel_idempotent() { + AsyncGate gate = AsyncGate.create(1); + + gate.cancel(); + gate.cancel(); + + assertThat(gate.isCancelled()).isTrue(); + } + + @Test + public void create_factoryMethod_returnsConfiguredGate() { + AsyncGate gate = AsyncGate.create(5); + + assertThat(gate.availablePermits()).isEqualTo(5); + assertThat(gate.activeCount()).isEqualTo(0); + assertThat(gate.isCancelled()).isFalse(); + } + + @Test + public void create_withMaxInteger_initializesCorrectly() { + AsyncGate gate = AsyncGate.create(Integer.MAX_VALUE); + + assertThat(gate.availablePermits()).isEqualTo(Integer.MAX_VALUE); + assertThat(gate.tryAcquire()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + } + + @Test + public void concurrentTryAcquireAndRelease_neverExceedsMaxConcurrency() + throws InterruptedException { + int maxConcurrency = 4; + int taskCount = 32; + AsyncGate gate = AsyncGate.create(maxConcurrency); + AtomicInteger peakConcurrency = new AtomicInteger(); + ExecutorService executor = Executors.newFixedThreadPool(8); + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(taskCount); + + try { + for (int i = 0; i < taskCount; i++) { + executor.execute( + () -> { + try { + startLatch.await(); + while (!gate.tryAcquire()) { + Thread.sleep(1); + } + int current = gate.activeCount(); + peakConcurrency.accumulateAndGet(current, Math::max); + Thread.sleep(2); + gate.release(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }); + } + + startLatch.countDown(); + boolean completed = doneLatch.await(5, SECONDS); + + assertThat(completed).isTrue(); + assertThat(peakConcurrency.get()).isAtMost(maxConcurrency); + assertThat(gate.activeCount()).isEqualTo(0); + } finally { + executor.shutdown(); + } + } +} diff --git a/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel b/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel index 9116818dc..38d1d0d70 100644 --- a/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel +++ b/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel @@ -40,6 +40,9 @@ java_library( "//extensions", "//parser:macro", "//runtime", + "//runtime:async_call", + "//runtime:async_drain_strategy", + "//runtime:async_options", "//runtime:descriptor_type_resolver", "//runtime:dispatcher", "//runtime:function_binding", @@ -49,6 +52,8 @@ java_library( "//runtime:runtime_helpers", "//runtime:standard_functions", "//runtime:unknown_attributes", + "//runtime/planner:async_completion_coordinator", + "//runtime/planner:async_gate", "//runtime/planner:program_planner", "//runtime/standard:type", "@cel_spec//proto/cel/expr/conformance/proto3:test_all_types_java_proto",