From 88bc19126dc7517c2e3bef7b0cd9f2c06b125c56 Mon Sep 17 00:00:00 2001 From: Sean Huh Date: Tue, 1 Sep 2026 23:49:48 -0700 Subject: [PATCH] Add async concurrency gate and re-evaluation coordinator PiperOrigin-RevId: 974934413 --- runtime/planner/BUILD.bazel | 24 + .../src/main/java/dev/cel/runtime/BUILD.bazel | 1 - .../java/dev/cel/runtime/CelRuntimeImpl.java | 31 +- .../planner/AsyncCompletionCoordinator.java | 324 +++++++++ .../dev/cel/runtime/planner/AsyncGate.java | 239 +++++++ .../java/dev/cel/runtime/planner/BUILD.bazel | 62 ++ .../cel/runtime/CelRuntimeLegacyImplTest.java | 2 + .../AsyncCompletionCoordinatorTest.java | 673 ++++++++++++++++++ .../cel/runtime/planner/AsyncGateTest.java | 497 +++++++++++++ .../java/dev/cel/runtime/planner/BUILD.bazel | 5 + 10 files changed, 1834 insertions(+), 24 deletions(-) create mode 100644 runtime/src/main/java/dev/cel/runtime/planner/AsyncCompletionCoordinator.java create mode 100644 runtime/src/main/java/dev/cel/runtime/planner/AsyncGate.java create mode 100644 runtime/src/test/java/dev/cel/runtime/planner/AsyncCompletionCoordinatorTest.java create mode 100644 runtime/src/test/java/dev/cel/runtime/planner/AsyncGateTest.java diff --git a/runtime/planner/BUILD.bazel b/runtime/planner/BUILD.bazel index 860d413a0..e15fa2989 100644 --- a/runtime/planner/BUILD.bazel +++ b/runtime/planner/BUILD.bazel @@ -21,3 +21,27 @@ java_library( visibility = ["//:internal"], exports = ["//runtime/src/main/java/dev/cel/runtime/planner:planned_program"], ) + +java_library( + name = "async_gate", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate"], +) + +cel_android_library( + name = "async_gate_android", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate_android"], +) + +java_library( + name = "async_completion_coordinator", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator"], +) + +cel_android_library( + name = "async_completion_coordinator_android", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator_android"], +) diff --git a/runtime/src/main/java/dev/cel/runtime/BUILD.bazel b/runtime/src/main/java/dev/cel/runtime/BUILD.bazel index 9518e1601..43b490cd3 100644 --- a/runtime/src/main/java/dev/cel/runtime/BUILD.bazel +++ b/runtime/src/main/java/dev/cel/runtime/BUILD.bazel @@ -855,7 +855,6 @@ java_library( "//runtime:activation", "//runtime:interpretable", "//runtime:proto_message_activation_factory", - "//runtime:resolved_overload", "//runtime/planner:planned_program", "//runtime/planner:program_planner", "//runtime/standard:type", diff --git a/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java b/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java index 857434ba2..11e7b7ee2 100644 --- a/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java +++ b/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java @@ -112,22 +112,7 @@ public Program createProgram(CelAbstractSyntaxTree ast) throws CelEvaluationExce return toRuntimeProgram(planner().plan(ast)); } - private static final CelFunctionResolver EMPTY_FUNCTION_RESOLVER = - new CelFunctionResolver() { - @Override - public Optional findOverloadMatchingArgs( - String functionName, Collection overloadIds, Object[] args) { - return Optional.empty(); - } - - @Override - public Optional findOverloadMatchingArgs( - String functionName, Object[] args) { - return Optional.empty(); - } - }; - - public Program toRuntimeProgram(dev.cel.runtime.Program program) { + private Program toRuntimeProgram(dev.cel.runtime.Program program) { return new Program() { @Override @@ -152,7 +137,7 @@ public Object eval(Message message) throws CelEvaluationException { return plannedProgram.evalOrThrow( plannedProgram.interpretable(), ProtoMessageActivationFactory.fromProto(message, plannedProgram.options()), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, /* partialVars= */ null, /* listener= */ null); } @@ -215,14 +200,14 @@ public ListenableFuture evalAsync(PartialVars partialVars) { @Override public Object trace(CelEvaluationListener listener) throws CelEvaluationException { return ((PlannedProgram) program) - .trace(GlobalResolver.EMPTY, EMPTY_FUNCTION_RESOLVER, null, listener); + .trace(GlobalResolver.EMPTY, CelFunctionResolver.EMPTY, null, listener); } @Override public Object trace(Map mapValue, CelEvaluationListener listener) throws CelEvaluationException { return ((PlannedProgram) program) - .trace(Activation.copyOf(mapValue), EMPTY_FUNCTION_RESOLVER, null, listener); + .trace(Activation.copyOf(mapValue), CelFunctionResolver.EMPTY, null, listener); } @Override @@ -232,7 +217,7 @@ public Object trace(Message message, CelEvaluationListener listener) return plannedProgram.evalOrThrow( plannedProgram.interpretable(), ProtoMessageActivationFactory.fromProto(message, plannedProgram.options()), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, /* partialVars= */ null, listener); } @@ -243,7 +228,7 @@ public Object trace(CelVariableResolver resolver, CelEvaluationListener listener return ((PlannedProgram) program) .trace( (name) -> resolver.find(name).orElse(null), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, null, listener); } @@ -278,13 +263,13 @@ public Object trace(PartialVars partialVars, CelEvaluationListener listener) return ((PlannedProgram) program) .trace( (name) -> partialVars.resolver().find(name).orElse(null), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, partialVars, listener); } @Override - public Object advanceEvaluation(UnknownContext context) throws CelEvaluationException { + public Object advanceEvaluation(UnknownContext context) { throw new UnsupportedOperationException("Unsupported operation."); } }; 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..c7adc5f9b --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/planner/AsyncCompletionCoordinator.java @@ -0,0 +1,324 @@ +// 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 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.CelAsyncEvaluationOptions; +import java.util.ArrayList; +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 { + // CEL-Internal-4 + private final Object lock; + + private final CelAsyncEvaluationOptions options; + private final AsyncGate gate; + private final Executor executor; + + // CEL-Internal-4 + private final Consumer failureCallback; + + @GuardedBy("lock") + private final List completedBatch; + + @GuardedBy("lock") + private @Nullable Runnable continuation; + + @GuardedBy("lock") + private @Nullable ScheduledFuture debounceTimer; + + @GuardedBy("lock") + private boolean isWaiting; + + @GuardedBy("lock") + private boolean isCancelled; + + @GuardedBy("lock") + private long cycleId; + + @GuardedBy("lock") + private long debounceGeneration; + + boolean hasPendingBatch() { + synchronized (lock) { + return !completedBatch.isEmpty(); + } + } + + void notifyCallCompleted(CelAsyncCall call) { + checkNotNull(call, "call must not be null"); + ImmutableList batchSnapshot; + int activeCount; + long currentCycleId; + + synchronized (lock) { + if (isCancelled) { + return; + } + completedBatch.add(call); + if (!isWaiting) { + return; + } + batchSnapshot = ImmutableList.copyOf(completedBatch); + activeCount = gate.activeCount(); + currentCycleId = cycleId; + } + + // Evaluate drain strategy outside monitor lock + CelAsyncDrainAction action = options.drainStrategy().nextAction(batchSnapshot, activeCount); + applyDrainAction(action, currentCycleId, activeCount); + } + + /** + * Registers a continuation callback to run when completions satisfy the drain strategy. + * + * @return true if successfully registered or continuation was dispatched; false if cancelled. + */ + boolean waitForCompletions(Runnable continuationCallback) { + checkNotNull(continuationCallback, "continuationCallback must not be null"); + ImmutableList batchSnapshot; + int activeCount; + long currentCycleId; + + synchronized (lock) { + if (isCancelled) { + return false; + } + if (isWaiting) { + throw new IllegalStateException("Coordinator is already waiting for completions"); + } + this.continuation = continuationCallback; + this.isWaiting = true; + this.cycleId++; + batchSnapshot = ImmutableList.copyOf(completedBatch); + activeCount = gate.activeCount(); + currentCycleId = this.cycleId; + if (batchSnapshot.isEmpty() && activeCount > 0) { + return true; + } + } + + // Evaluate drain strategy outside monitor lock + CelAsyncDrainAction action = options.drainStrategy().nextAction(batchSnapshot, activeCount); + applyDrainAction(action, currentCycleId, activeCount); + return true; + } + + private void applyDrainAction(CelAsyncDrainAction action, long currentCycleId, int activeCount) { + Runnable toRun = null; + ScheduledFuture timerToCancel = null; + long waitNanos = -1; + long scheduledGen = -1; + + synchronized (lock) { + if (!isCancelled && isWaiting && this.cycleId == currentCycleId) { + if (action.shouldReevaluate() || (activeCount == 0 && !completedBatch.isEmpty())) { + timerToCancel = cancelDebounceTimerUnderLock(); + toRun = drainAndResetUnderLock(); + } else if (action.waitDuration().isZero()) { + // Indefinite wait for next completion: cancel any pending timer + timerToCancel = cancelDebounceTimerUnderLock(); + } else { + // Sliding window debounce: reset existing timer and reschedule for new wait duration + timerToCancel = cancelDebounceTimerUnderLock(); + waitNanos = action.waitDuration().toNanos(); + scheduledGen = ++debounceGeneration; + } + } + } + + if (timerToCancel != null) { + timerToCancel.cancel(false); + } + if (toRun != null) { + dispatchContinuation(toRun); + } else if (waitNanos >= 0) { + scheduleDebounce(waitNanos, currentCycleId, scheduledGen); + } + } + + private void dispatchContinuation(Runnable toRun) { + try { + executor.execute(toRun); + } catch (Throwable t) { + failureCallback.accept(t); + } + } + + void cancel() { + ScheduledFuture timerToCancel; + synchronized (lock) { + isCancelled = true; + cycleId++; + debounceGeneration++; + timerToCancel = cancelDebounceTimerUnderLock(); + isWaiting = false; + continuation = null; + completedBatch.clear(); + } + if (timerToCancel != null) { + timerToCancel.cancel(false); + } + gate.cancel(); + } + + private void scheduleDebounce(long nanos, long scheduledCycleId, long scheduledGen) { + ScheduledExecutorService scheduler = options.resolveScheduledExecutorService(); + ScheduledFuture future = + scheduler.schedule( + () -> onDebounceFired(scheduledCycleId, scheduledGen), nanos, NANOSECONDS); + 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); + } + } + + @VisibleForTesting + void onDebounceFired(long firedCycleId) { + long currentGen; + synchronized (lock) { + currentGen = this.debounceGeneration; + } + onDebounceFired(firedCycleId, currentGen); + } + + @VisibleForTesting + void onDebounceFired(long firedCycleId, long firedGen) { + Runnable toRun = null; + synchronized (lock) { + if (!isCancelled + && isWaiting + && this.cycleId == firedCycleId + && this.debounceGeneration == firedGen) { + debounceTimer = null; + toRun = drainAndResetUnderLock(); + } + } + if (toRun != null) { + dispatchContinuation(toRun); + } + } + + @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 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; + } + } + + AsyncCompletionCoordinator( + CelAsyncEvaluationOptions options, + AsyncGate gate, + Executor executor, + Consumer failureCallback) { + this.lock = new Object(); + this.options = checkNotNull(options, "options must not be null"); + this.gate = checkNotNull(gate, "gate must not be null"); + this.executor = checkNotNull(executor, "executor must not be null"); + this.failureCallback = checkNotNull(failureCallback, "failureCallback must not be null"); + this.completedBatch = new ArrayList<>(); + this.isWaiting = false; + this.isCancelled = false; + this.cycleId = 0; + this.debounceGeneration = 0; + } + + AsyncCompletionCoordinator(CelAsyncEvaluationOptions options, AsyncGate gate, Executor executor) { + this( + options, + gate, + executor, + t -> + Thread.currentThread() + .getUncaughtExceptionHandler() + .uncaughtException(Thread.currentThread(), t)); + } +} 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..f0454d478 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/planner/AsyncGate.java @@ -0,0 +1,239 @@ +// 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 com.google.common.annotations.VisibleForTesting; +import javax.annotation.concurrent.ThreadSafe; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executor; +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. */ +@ThreadSafe +final class AsyncGate { + + private final @Nullable Semaphore semaphore; + + // CEL-Internal-4 + private final Queue pendingTasks; + + private final AtomicInteger activeCount; + private final AtomicBoolean cancelled; + + /** + * Represents a single-use permit for an admitted asynchronous task. + * + *

Permit release is strictly idempotent: subsequent calls to {@link #release(Executor)} are + * no-ops and will not double-release permits or decrement active counts below zero. + */ + final class Permit { + private final AtomicBoolean released; + + /** + * Releases the permit held by this task and triggers draining of pending tasks on the executor. + * + *

This method is safe to call multiple times; only the first invocation takes effect. It + * will never throw exceptions (such as {@link java.util.concurrent.RejectedExecutionException}) + * out of cleanup paths. + */ + void release(Executor executor) { + checkNotNull(executor, "executor must not be null"); + if (!released.compareAndSet(false, true)) { + return; + } + activeCount.decrementAndGet(); + if (semaphore != null) { + semaphore.release(); + drainPending(executor); + } + } + + private void markRejected() { + if (released.compareAndSet(false, true)) { + activeCount.decrementAndGet(); + } + } + + private Permit() { + this(false); + } + + private Permit(boolean released) { + this.released = new AtomicBoolean(released); + } + } + + /** A task that accepts its associated {@link Permit}. */ + @FunctionalInterface + interface GatedTask { + void run(Permit permit); + } + + @VisibleForTesting + static final class QueuedTask { + final GatedTask task; + final Permit permit; + + QueuedTask(GatedTask task, Permit permit) { + this.task = checkNotNull(task, "task must not be null"); + this.permit = checkNotNull(permit, "permit must not be null"); + } + } + + /** Cancels the gate, preventing any queued tasks from executing. */ + void cancel() { + cancelled.set(true); + clearPendingTasks(); + } + + /** + * Dispatches a gated task for execution. + * + *

If concurrency permits allow, the task is executed immediately on the calling thread. + * Otherwise, it is enqueued into pending tasks and will be drained on the provided executor when + * a permit is released. + */ + void dispatch(Executor executor, GatedTask task) { + checkNotNull(executor, "executor must not be null"); + checkNotNull(task, "task must not be null"); + if (cancelled.get()) { + return; + } + + activeCount.incrementAndGet(); + Permit permit = new Permit(); + + if (semaphore == null) { + runTask(executor, task, permit); + return; + } + + if (pendingTasks.isEmpty() && semaphore.tryAcquire()) { + runTask(executor, task, permit); + return; + } + + pendingTasks.add(new QueuedTask(task, permit)); + drainPending(executor); + } + + /** + * Dispatches a simple {@link Runnable} task. + * + *

The concurrency permit is held for the duration of {@link Runnable#run()} and automatically + * released upon completion. + */ + void dispatch(Executor executor, Runnable task) { + checkNotNull(executor, "executor must not be null"); + checkNotNull(task, "task must not be null"); + dispatch( + executor, + permit -> { + try { + task.run(); + } finally { + permit.release(executor); + } + }); + } + + @VisibleForTesting + Permit acquirePermitForTesting() { + activeCount.incrementAndGet(); + if (semaphore != null) { + semaphore.acquireUninterruptibly(); + } + return new Permit(); + } + + private void runTask(Executor executor, GatedTask task, Permit permit) { + try { + task.run(permit); + } catch (Throwable t) { + permit.release(executor); + throw t; + } + } + + @VisibleForTesting + void drainPending(Executor executor) { + if (semaphore == null || cancelled.get()) { + clearPendingTasks(); + return; + } + while (!pendingTasks.isEmpty()) { + if (!semaphore.tryAcquire()) { + break; + } + QueuedTask queuedTask = pendingTasks.poll(); + if (queuedTask != null) { + try { + executor.execute( + () -> { + if (cancelled.get()) { + queuedTask.permit.markRejected(); + semaphore.release(); + return; + } + runTask(executor, queuedTask.task, queuedTask.permit); + }); + } catch (Throwable e) { + semaphore.release(); + queuedTask.permit.markRejected(); + clearPendingTasks(); + break; + } + } else { + semaphore.release(); + break; + } + } + } + + private void clearPendingTasks() { + QueuedTask queuedTask; + while ((queuedTask = pendingTasks.poll()) != null) { + queuedTask.permit.markRejected(); + } + } + + /** Returns the current number of active (admitted or queued) tasks. */ + int activeCount() { + return activeCount.get(); + } + + @VisibleForTesting + boolean isCancelled() { + return cancelled.get(); + } + + AsyncGate(int maxConcurrency) { + this(maxConcurrency > 0 ? new Semaphore(maxConcurrency) : null, new ConcurrentLinkedQueue<>()); + } + + @VisibleForTesting + AsyncGate(@Nullable Semaphore semaphore, Queue pendingTasks) { + this.semaphore = semaphore; + this.pendingTasks = checkNotNull(pendingTasks, "pendingTasks must not be null"); + this.activeCount = 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..dcd89dffd 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"], @@ -717,6 +747,38 @@ cel_android_library( ], ) +cel_android_library( + name = "async_gate_android", + 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", + "@maven_android//:com_google_guava_guava", + ], +) + +cel_android_library( + name = "async_completion_coordinator_android", + srcs = ["AsyncCompletionCoordinator.java"], + tags = [ + ], + deps = [ + ":async_gate_android", + "//runtime:async_call_android", + "//runtime:async_drain_strategy_android", + "//runtime:async_options_android", + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", + "@maven//:org_jspecify_jspecify", + "@maven_android//:com_google_guava_guava", + ], +) + cel_android_library( name = "activation_wrapper_android", srcs = ["ActivationWrapper.java"], diff --git a/runtime/src/test/java/dev/cel/runtime/CelRuntimeLegacyImplTest.java b/runtime/src/test/java/dev/cel/runtime/CelRuntimeLegacyImplTest.java index 5bef0c61e..972b9b1d2 100644 --- a/runtime/src/test/java/dev/cel/runtime/CelRuntimeLegacyImplTest.java +++ b/runtime/src/test/java/dev/cel/runtime/CelRuntimeLegacyImplTest.java @@ -106,6 +106,8 @@ public void toRuntimeBuilder_collectionProperties_areImmutable() { } @Test + @SuppressWarnings( + "deprecation") // Tests deprecated setStandardEnvironmentEnabled on legacy builder public void toRuntimeBuilder_optionalProperties() { Function customTypeFactory = (typeName) -> TestAllTypes.newBuilder(); CelStandardFunctions overriddenStandardFunctions = 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..6e42c36db --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/planner/AsyncCompletionCoordinatorTest.java @@ -0,0 +1,673 @@ +// 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 java.time.Duration; +import java.util.ArrayList; +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 notifyCallCompleted_whenWaitingWithPendingActiveCalls_schedulesDebounceTimer() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(1); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(continuationRan.get()).isFalse(); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isTrue(); + assertThat(scheduler.getQueue()).isNotEmpty(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void + notifyCallCompleted_whenWaitingWithDrainAllStrategy_waitsIndefinitelyWithoutSchedulingTimer() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainAll()) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(2); + AsyncGate.Permit permit1 = gate.acquirePermitForTesting(); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + permit1.release(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(continuationRan.get()).isFalse(); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + assertThat(scheduler.getQueue()).isEmpty(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void notifyCallCompleted_whenDebounceTimerPending_resetsDebounceTimerForSlidingWindow() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(2); + gate.acquirePermitForTesting(); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + coordinator.notifyCallCompleted(DUMMY_CALL); + ScheduledFuture firstTimer = (ScheduledFuture) scheduler.getQueue().peek(); + assertThat(firstTimer).isNotNull(); + assertThat(firstTimer.isCancelled()).isFalse(); + + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(firstTimer.isCancelled()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isTrue(); + assertThat(continuationRan.get()).isFalse(); + } 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 = new AsyncGate(1); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + ScheduledFuture scheduledTask = + requireNonNull((ScheduledFuture) scheduler.getQueue().peek()); + + ((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 cancel_cancelsDebounceTimerAndPreventsContinuation() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(1); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + ScheduledFuture scheduledTask = (ScheduledFuture) scheduler.getQueue().peek(); + assertThat(scheduledTask).isNotNull(); + assertThat(scheduledTask.isCancelled()).isFalse(); + + coordinator.cancel(); + + assertThat(coordinator.hasPendingBatch()).isFalse(); + assertThat(coordinator.isWaiting()).isFalse(); + assertThat(coordinator.hasContinuation()).isFalse(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + assertThat(scheduledTask.isCancelled()).isTrue(); + assertThat(continuationRan.get()).isFalse(); + + ((Runnable) scheduledTask).run(); + assertThat(continuationRan.get()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @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 = new AsyncGate(1); + AsyncGate.Permit pass1Permit = gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + AtomicInteger pass1Count = new AtomicInteger(); + coordinator.waitForCompletions(pass1Count::incrementAndGet); + ScheduledFuture pass1Timer = (ScheduledFuture) scheduler.getQueue().peek(); + assertThat(pass1Timer).isNotNull(); + + pass1Permit.release(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + assertThat(pass1Count.get()).isEqualTo(1); + assertThat(coordinator.isWaiting()).isFalse(); + + gate.acquirePermitForTesting(); + AtomicInteger pass2Count = new AtomicInteger(); + coordinator.waitForCompletions(pass2Count::incrementAndGet); + assertThat(coordinator.isWaiting()).isTrue(); + + ((Runnable) pass1Timer).run(); + + assertThat(pass2Count.get()).isEqualTo(0); + assertThat(coordinator.isWaiting()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void waitForCompletions_whenCoordinatorCancelled_returnsFalseAndDoesNotRunContinuation() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.cancel(); + AtomicBoolean ran = new AtomicBoolean(false); + + boolean result = coordinator.waitForCompletions(() -> ran.set(true)); + + assertThat(result).isFalse(); + assertThat(ran.get()).isFalse(); + } + + @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 = new AsyncGate(workerCount); + List permits = new ArrayList<>(); + for (int i = 0; i < workerCount; i++) { + permits.add(gate.acquirePermitForTesting()); + } + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, workers); + 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++) { + int workerIndex = i; + workers.execute( + () -> { + readyLatch.countDown(); + try { + startLatch.await(); + permits.get(workerIndex).release(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + } + + readyLatch.await(5, SECONDS); + startLatch.countDown(); + + assertThat(continuationLatch.await(5, SECONDS)).isTrue(); + + workers.shutdown(); + assertThat(workers.awaitTermination(5, SECONDS)).isTrue(); + assertThat(continuationDispatches.get()).isEqualTo(1); + assertThat(coordinator.isWaiting()).isFalse(); + } finally { + workers.shutdownNow(); + scheduler.shutdownNow(); + } + } + + @Test + public void waitForCompletions_whenAlreadyWaiting_throwsIllegalStateException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.waitForCompletions(() -> {}); + + assertThrows(IllegalStateException.class, () -> coordinator.waitForCompletions(() -> {})); + } + + @Test + public void notifyCallCompleted_whenCancelled_ignoresCompletion() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.cancel(); + + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(coordinator.hasPendingBatch()).isFalse(); + } + + @Test + public void constructorAndMethods_nullArguments_throwsNullPointerException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + + assertThrows( + NullPointerException.class, + () -> new AsyncCompletionCoordinator(null, gate, Runnable::run)); + assertThrows( + NullPointerException.class, + () -> new AsyncCompletionCoordinator(options, null, Runnable::run)); + assertThrows( + NullPointerException.class, () -> new AsyncCompletionCoordinator(options, gate, null)); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + assertThrows(NullPointerException.class, () -> coordinator.notifyCallCompleted(null)); + assertThrows(NullPointerException.class, () -> coordinator.waitForCompletions(null)); + } + + @Test + public void + waitForCompletions_whenDrainStrategySatisfiedImmediately_dispatchesContinuationWithoutTimer() { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainAll()) + .build(); + AsyncGate gate = new AsyncGate(1); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(continuationRan.get()).isTrue(); + assertThat(coordinator.isWaiting()).isFalse(); + assertThat(coordinator.hasPendingBatch()).isFalse(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + } + + @Test + public void constructor_initializesCycleIdToZero() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + assertThat(coordinator.cycleId()).isEqualTo(0); + } + + @Test + public void notifyCallCompleted_whenDebounceTimerPending_cancelsExistingTimerWithoutInterrupt() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(1); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + coordinator.waitForCompletions(() -> {}); + ScheduledFuture firstTimer = (ScheduledFuture) scheduler.getQueue().peek(); + assertThat(firstTimer).isNotNull(); + assertThat(firstTimer.isCancelled()).isFalse(); + + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(firstTimer.isCancelled()).isTrue(); + } finally { + scheduler.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 = new AsyncGate(1); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinatorHolder[0] = coordinator; + coordinator.notifyCallCompleted(DUMMY_CALL); + coordinator.waitForCompletions(() -> {}); + + ScheduledFuture scheduledTask = (ScheduledFuture) scheduler.getQueue().peek(); + assertThat(scheduledTask).isNotNull(); + assertThat(scheduledTask.isCancelled()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void drainAndReset_incrementsCycleId() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.waitForCompletions(() -> {}); + long initialCycleId = coordinator.cycleId(); + + permit.release(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(coordinator.cycleId()).isGreaterThan(initialCycleId); + } + + @Test + public void drainAndReset_clearsContinuation() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.waitForCompletions(() -> {}); + assertThat(coordinator.hasContinuation()).isTrue(); + + permit.release(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(coordinator.hasContinuation()).isFalse(); + } + + @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 = new AsyncGate(1); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, rejectingNullExecutor); + coordinator.waitForCompletions(() -> {}); + + coordinator.onDebounceFired(coordinator.cycleId() - 1); + + assertThat(executedCount.get()).isEqualTo(0); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasContinuation()).isTrue(); + } + + @Test + public void + waitForCompletions_whenNoCallsInFlightAndEmptyBatch_evaluatesDrainStrategyAndInvokesContinuation() { + AsyncGate gate = new AsyncGate(1); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMillis(50))) + .build(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(continuationRan.get()).isTrue(); + assertThat(coordinator.isWaiting()).isFalse(); + } + + @Test + public void applyDrainAction_whenActiveCountZeroAndStrategyWaits_forcesReevaluation() { + CelAsyncDrainStrategy alwaysWaitStrategy = (batch, active) -> CelAsyncDrainAction.waitForMore(); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setDrainStrategy(alwaysWaitStrategy).build(); + AsyncGate gate = new AsyncGate(1); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + coordinator.notifyCallCompleted(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 = new AsyncGate(1); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, rejectingExecutor, failure::set); + coordinator.waitForCompletions(() -> {}); + + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(failure.get()).isInstanceOf(RejectedExecutionException.class); + } + + @Test + @SuppressWarnings("Immutable") + public void + waitForCompletions_whenEmptyBatchAndActiveCallsInFlight_returnsEarlyWithoutEvaluatingStrategy() { + AsyncGate gate = new AsyncGate(1); + gate.acquirePermitForTesting(); + AtomicBoolean strategyEvaluated = new AtomicBoolean(false); + CelAsyncDrainStrategy strategy = + (batch, active) -> { + strategyEvaluated.set(true); + return CelAsyncDrainAction.reevaluate(); + }; + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder().setDrainStrategy(strategy).build(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + boolean result = coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(result).isTrue(); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(continuationRan.get()).isFalse(); + assertThat(strategyEvaluated.get()).isFalse(); + } + + @Test + public void applyDrainAction_whenCycleIdMismatch_doesNotExecuteContinuation() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + gate.acquirePermitForTesting(); + AtomicInteger continuationRan = new AtomicInteger(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.waitForCompletions(continuationRan::incrementAndGet); + coordinator.cancel(); + + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(continuationRan.get()).isEqualTo(0); + assertThat(coordinator.isWaiting()).isFalse(); + } + + @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 = new AsyncGate(2); + gate.acquirePermitForTesting(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicInteger continuationRan = new AtomicInteger(); + coordinator.waitForCompletions(continuationRan::incrementAndGet); + coordinator.notifyCallCompleted(DUMMY_CALL); + coordinator.notifyCallCompleted(DUMMY_CALL); + + coordinator.onDebounceFired(coordinator.cycleId(), 1); + + assertThat(continuationRan.get()).isEqualTo(0); + assertThat(coordinator.isWaiting()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void cancel_cancelsAssociatedGate() { + AsyncGate gate = new AsyncGate(1); + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + coordinator.cancel(); + + assertThat(gate.isCancelled()).isTrue(); + } +} 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..ed11fe968 --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/planner/AsyncGateTest.java @@ -0,0 +1,497 @@ +// 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 static org.junit.Assert.assertThrows; + +import com.google.common.collect.ForwardingQueue; +import java.util.AbstractQueue; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +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.Semaphore; +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 AsyncGateTest { + + @Test + public void unboundedConcurrency_runsImmediatelyAndTracksActive() { + AsyncGate gate = new AsyncGate(0); + AtomicBoolean ran = new AtomicBoolean(false); + AtomicReference permit = new AtomicReference<>(); + + gate.dispatch( + Runnable::run, + p -> { + ran.set(true); + permit.set(p); + }); + + assertThat(ran.get()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + + permit.get().release(Runnable::run); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void dispatch_whenConcurrencyLimitReached_enqueuesTaskAndIncrementsActiveCount() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + AtomicBoolean taskRan = new AtomicBoolean(false); + + gate.dispatch(Runnable::run, () -> taskRan.set(true)); + + assertThat(taskRan.get()).isFalse(); + assertThat(gate.activeCount()).isEqualTo(2); + + permit.release(Runnable::run); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void release_whenTasksEnqueued_drainsPendingTask() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + AtomicBoolean taskRan = new AtomicBoolean(false); + gate.dispatch(Runnable::run, () -> taskRan.set(true)); + + permit.release(Runnable::run); + + assertThat(taskRan.get()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void cancel_withQueuedPendingTasks_clearsPendingTasksImmediately() { + ConcurrentLinkedQueue pendingTasks = new ConcurrentLinkedQueue<>(); + AsyncGate gate = new AsyncGate(new Semaphore(1), pendingTasks); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + gate.dispatch(Runnable::run, () -> {}); + + gate.cancel(); + + assertThat(pendingTasks).isEmpty(); + assertThat(gate.activeCount()).isEqualTo(1); + + permit.release(Runnable::run); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void cancel_whenCancelled_subsequentDispatchIsIgnored() { + AsyncGate gate = new AsyncGate(1); + AtomicBoolean taskRan = new AtomicBoolean(false); + gate.cancel(); + + gate.dispatch(Runnable::run, () -> taskRan.set(true)); + + assertThat(taskRan.get()).isFalse(); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void drainPending_whenCancelled_clearsPendingTasksAndDoesNotRun() { + ConcurrentLinkedQueue pendingTasks = new ConcurrentLinkedQueue<>(); + AsyncGate gate = new AsyncGate(new Semaphore(1), pendingTasks); + AtomicBoolean taskRan = new AtomicBoolean(false); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + gate.cancel(); + pendingTasks.add(new AsyncGate.QueuedTask(p -> taskRan.set(true), permit)); + + permit.release(Runnable::run); + + assertThat(taskRan.get()).isFalse(); + assertThat(pendingTasks).isEmpty(); + } + + @Test + public void dispatch_whenPermitAvailableAfterQueueing_drainPendingRunsTask() { + AtomicBoolean taskRan = new AtomicBoolean(false); + Semaphore semaphore = new Semaphore(1); + ConcurrentLinkedQueue delegate = new ConcurrentLinkedQueue<>(); + Queue queue = + new ForwardingQueue() { + @Override + protected Queue delegate() { + return delegate; + } + + @Override + public boolean add(AsyncGate.QueuedTask r) { + boolean res = super.add(r); + semaphore.release(); + return res; + } + }; + AsyncGate gate = new AsyncGate(semaphore, queue); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + + gate.dispatch(Runnable::run, () -> taskRan.set(true)); + + assertThat(taskRan.get()).isTrue(); + + permit.release(Runnable::run); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void concurrentDrainAndCancel_retainsPermitBalance() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(4); + try { + for (int i = 0; i < 50; i++) { + AsyncGate gate = new AsyncGate(1); + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(2); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + gate.dispatch(executor, p -> p.release(executor)); + + executor.execute( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + permit.release(executor); + doneLatch.countDown(); + } + }); + + executor.execute( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + gate.cancel(); + doneLatch.countDown(); + } + }); + + startLatch.countDown(); + assertThat(doneLatch.await(5, SECONDS)).isTrue(); + for (int retry = 0; retry < 100 && gate.activeCount() > 0; retry++) { + Thread.sleep(5); + } + assertThat(gate.activeCount()).isEqualTo(0); + } + } finally { + executor.shutdown(); + executor.awaitTermination(5, SECONDS); + } + } + + @Test + public void drainPending_whenTaskPolledIsNull_releasesPermitAndBreaks() { + Semaphore semaphore = new Semaphore(1); + AtomicInteger pollCount = new AtomicInteger(); + Queue queue = + new AbstractQueue() { + @Override + public boolean offer(AsyncGate.QueuedTask e) { + return true; + } + + @Override + public AsyncGate.QueuedTask peek() { + return null; + } + + @Override + public Iterator iterator() { + return Collections.emptyIterator(); + } + + @Override + public int size() { + return 1; + } + + @Override + public boolean isEmpty() { + return false; + } + + @Override + public AsyncGate.QueuedTask poll() { + pollCount.incrementAndGet(); + return null; + } + }; + AsyncGate gate = new AsyncGate(semaphore, queue); + + gate.drainPending(Runnable::run); + + assertThat(pollCount.get()).isEqualTo(1); + assertThat(semaphore.availablePermits()).isEqualTo(1); + } + + @Test + public void dispatch_taskThrowsRuntimeException_releasesPermitAndDecrementsActiveCount() { + AsyncGate gate = new AsyncGate(1); + + assertThrows( + RuntimeException.class, + () -> + gate.dispatch( + Runnable::run, + () -> { + throw new RuntimeException("fail"); + })); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void dispatch_unboundedConcurrency_taskThrows_decrementsActiveCountAndRethrows() { + AsyncGate gate = new AsyncGate(0); + + assertThrows( + RuntimeException.class, + () -> + gate.dispatch( + Runnable::run, + () -> { + throw new RuntimeException("fail"); + })); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void drainPending_executorThrows_releasesPermitDecrementsActiveCountAndDoesNotRethrow() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + gate.dispatch(Runnable::run, () -> {}); + Executor rejectingExecutor = + r -> { + throw new RejectedExecutionException("rejected"); + }; + + permit.release(rejectingExecutor); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void dispatch_nullArguments_throwsNullPointerException() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + + assertThrows(NullPointerException.class, () -> gate.dispatch(null, () -> {})); + assertThrows(NullPointerException.class, () -> gate.dispatch(Runnable::run, (Runnable) null)); + assertThrows(NullPointerException.class, () -> gate.dispatch(null, (AsyncGate.GatedTask) null)); + assertThrows(NullPointerException.class, () -> permit.release(null)); + } + + @Test + public void negativeConcurrency_treatedAsUnbounded() { + AsyncGate gate = new AsyncGate(-1); + AtomicBoolean ran = new AtomicBoolean(false); + AtomicReference permit = new AtomicReference<>(); + + gate.dispatch( + Runnable::run, + p -> { + ran.set(true); + permit.set(p); + }); + + assertThat(ran.get()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + + permit.get().release(Runnable::run); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void releasePermit_boundedConcurrency_releasesSemaphorePermit() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + + permit.release(Runnable::run); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void drainPending_taskThrowsInWorkerThread_decrementsActiveCountAndReleasesPermit() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + gate.dispatch( + Runnable::run, + () -> { + throw new RuntimeException("worker thread boom"); + }); + + permit.release(Runnable::run); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void dispatch_whenTasksPending_enqueuesWithoutQueueBarging() { + Semaphore semaphore = new Semaphore(1); + Queue pendingTasks = new ConcurrentLinkedQueue<>(); + AsyncGate gate = new AsyncGate(semaphore, pendingTasks); + AsyncGate.Permit initialPermit = gate.acquirePermitForTesting(); + List order = new ArrayList<>(); + Queue queued = new ArrayDeque<>(); + gate.dispatch(queued::add, () -> order.add("first")); + semaphore.release(); + + gate.dispatch(queued::add, () -> order.add("second")); + + assertThat(queued).hasSize(1); + queued.poll().run(); + assertThat(queued).hasSize(1); + queued.poll().run(); + assertThat(order).containsExactly("first", "second").inOrder(); + + initialPermit.release(Runnable::run); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void release_calledMultipleTimes_isIdempotentAndDoesNotDriveActiveCountNegative() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + + permit.release(Runnable::run); + permit.release(Runnable::run); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void dispatch_taskThrowsAfterReleasingPermit_doesNotDoubleRelease() { + AsyncGate gate = new AsyncGate(1); + + assertThrows( + RuntimeException.class, + () -> + gate.dispatch( + Runnable::run, + permit -> { + permit.release(Runnable::run); + throw new RuntimeException("boom"); + })); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void dispatch_whenTaskQueued_incrementsActiveCount() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit1 = gate.acquirePermitForTesting(); + + gate.dispatch(Runnable::run, () -> {}); + + assertThat(gate.activeCount()).isEqualTo(2); + + permit1.release(Runnable::run); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void release_whenQueuedTaskReleased_decrementsActiveCount() { + AsyncGate gate = new AsyncGate(1); + AtomicReference permit1 = new AtomicReference<>(); + AtomicReference permit2 = new AtomicReference<>(); + gate.dispatch(Runnable::run, permit1::set); + gate.dispatch(Runnable::run, permit2::set); + + permit1.get().release(Runnable::run); + permit2.get().release(Runnable::run); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void dispatch_gatedTaskReceivesPermitDirectly() { + AsyncGate gate = new AsyncGate(1); + AtomicBoolean ran = new AtomicBoolean(false); + AtomicBoolean released = new AtomicBoolean(false); + + gate.dispatch( + Runnable::run, + permit -> { + ran.set(true); + permit.release(Runnable::run); + released.set(true); + }); + + assertThat(ran.get()).isTrue(); + assertThat(released.get()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void drainPending_executorRejects_clearsAllPendingTasksAndDecrementsActiveCount() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + AtomicBoolean task1Ran = new AtomicBoolean(false); + AtomicBoolean task2Ran = new AtomicBoolean(false); + gate.dispatch(Runnable::run, () -> task1Ran.set(true)); + gate.dispatch(Runnable::run, () -> task2Ran.set(true)); + Executor rejectingExecutor = + r -> { + throw new RejectedExecutionException("rejected"); + }; + + permit.release(rejectingExecutor); + + assertThat(task1Ran.get()).isFalse(); + assertThat(task2Ran.get()).isFalse(); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void drainPending_cancelledWhileTaskInExecutorQueue_taskDoesNotRunAndPermitRestored() { + AsyncGate gate = new AsyncGate(1); + AsyncGate.Permit permit = gate.acquirePermitForTesting(); + AtomicBoolean taskRan = new AtomicBoolean(false); + Queue executorQueue = new ArrayDeque<>(); + gate.dispatch(executorQueue::add, () -> taskRan.set(true)); + permit.release(executorQueue::add); + gate.cancel(); + + executorQueue.poll().run(); + + assertThat(taskRan.get()).isFalse(); + assertThat(gate.activeCount()).isEqualTo(0); + } +} 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",