diff --git a/runtime/planner/BUILD.bazel b/runtime/planner/BUILD.bazel index 0a4ef8a84..78d73885a 100644 --- a/runtime/planner/BUILD.bazel +++ b/runtime/planner/BUILD.bazel @@ -22,6 +22,12 @@ java_library( exports = ["//runtime/src/main/java/dev/cel/runtime/planner:planned_program"], ) +cel_android_library( + name = "planned_program_android", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:planned_program_android"], +) + java_library( name = "async_gate", testonly = 1, @@ -29,9 +35,35 @@ java_library( exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate"], ) +cel_android_library( + name = "async_gate_android", + testonly = 1, + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate_android"], +) + java_library( name = "async_completion_coordinator", testonly = 1, visibility = ["//:internal"], exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator"], ) + +cel_android_library( + name = "async_completion_coordinator_android", + testonly = 1, + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator_android"], +) + +java_library( + name = "async_call_state_tracker", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_call_state_tracker"], +) + +cel_android_library( + name = "async_call_state_tracker_android", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_call_state_tracker_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..158c932c7 100644 --- a/runtime/src/main/java/dev/cel/runtime/BUILD.bazel +++ b/runtime/src/main/java/dev/cel/runtime/BUILD.bazel @@ -829,7 +829,6 @@ java_library( ":function_binding", ":function_resolver", ":partial_vars", - ":program", ":proto_message_runtime_equality", ":runtime", ":runtime_equality", @@ -855,7 +854,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", @@ -997,6 +995,7 @@ java_library( "//common/types:type_providers", "//common/values", "//common/values:cel_value_provider", + "//runtime:async_options", "//runtime:evaluation_exception", "//runtime/planner:program_planner", "//runtime/standard:standard_function", @@ -1025,6 +1024,7 @@ cel_android_library( "//common/types:type_providers_android", "//common/values:cel_value_provider_android", "//common/values:values_android", + "//runtime:async_options_android", "//runtime:evaluation_exception", "//runtime/planner:program_planner_android", "//runtime/standard:standard_function_android", diff --git a/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java b/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java index 857434ba2..c3bff8dfd 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(PlannedProgram program) { return new Program() { @Override @@ -148,11 +133,9 @@ public Object eval(Map mapValue, CelFunctionResolver lateBoundFunctio @Override public Object eval(Message message) throws CelEvaluationException { - PlannedProgram plannedProgram = (PlannedProgram) program; - return plannedProgram.evalOrThrow( - plannedProgram.interpretable(), - ProtoMessageActivationFactory.fromProto(message, plannedProgram.options()), - EMPTY_FUNCTION_RESOLVER, + return program.evalOrThrow( + ProtoMessageActivationFactory.fromProto(message, program.options()), + CelFunctionResolver.EMPTY, /* partialVars= */ null, /* listener= */ null); } @@ -190,12 +173,6 @@ public ListenableFuture evalAsync( return program.evalAsync(mapValue, lateBoundFunctionResolver); } - @Override - public ListenableFuture evalAsync(Message message) { - throw new UnsupportedOperationException( - "evalAsync is not supported by this Program implementation."); - } - @Override public ListenableFuture evalAsync(CelVariableResolver resolver) { return program.evalAsync(resolver); @@ -212,27 +189,30 @@ public ListenableFuture evalAsync(PartialVars partialVars) { return program.evalAsync(partialVars); } + @Override + public ListenableFuture evalAsync(Message message) { + throw new UnsupportedOperationException( + "evalAsync is not supported by this Program implementation."); + } + @Override public Object trace(CelEvaluationListener listener) throws CelEvaluationException { - return ((PlannedProgram) program) - .trace(GlobalResolver.EMPTY, EMPTY_FUNCTION_RESOLVER, null, listener); + return program.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); + return program.trace( + Activation.copyOf(mapValue), CelFunctionResolver.EMPTY, null, listener); } @Override public Object trace(Message message, CelEvaluationListener listener) throws CelEvaluationException { - PlannedProgram plannedProgram = (PlannedProgram) program; - return plannedProgram.evalOrThrow( - plannedProgram.interpretable(), - ProtoMessageActivationFactory.fromProto(message, plannedProgram.options()), - EMPTY_FUNCTION_RESOLVER, + return program.evalOrThrow( + ProtoMessageActivationFactory.fromProto(message, program.options()), + CelFunctionResolver.EMPTY, /* partialVars= */ null, listener); } @@ -240,12 +220,8 @@ public Object trace(Message message, CelEvaluationListener listener) @Override public Object trace(CelVariableResolver resolver, CelEvaluationListener listener) throws CelEvaluationException { - return ((PlannedProgram) program) - .trace( - (name) -> resolver.find(name).orElse(null), - EMPTY_FUNCTION_RESOLVER, - null, - listener); + return program.trace( + (name) -> resolver.find(name).orElse(null), CelFunctionResolver.EMPTY, null, listener); } @Override @@ -254,12 +230,8 @@ public Object trace( CelFunctionResolver lateBoundFunctionResolver, CelEvaluationListener listener) throws CelEvaluationException { - return ((PlannedProgram) program) - .trace( - (name) -> resolver.find(name).orElse(null), - lateBoundFunctionResolver, - null, - listener); + return program.trace( + (name) -> resolver.find(name).orElse(null), lateBoundFunctionResolver, null, listener); } @Override @@ -268,23 +240,22 @@ public Object trace( CelFunctionResolver lateBoundFunctionResolver, CelEvaluationListener listener) throws CelEvaluationException { - return ((PlannedProgram) program) - .trace(Activation.copyOf(mapValue), lateBoundFunctionResolver, null, listener); + return program.trace( + Activation.copyOf(mapValue), lateBoundFunctionResolver, null, listener); } @Override public Object trace(PartialVars partialVars, CelEvaluationListener listener) throws CelEvaluationException { - return ((PlannedProgram) program) - .trace( - (name) -> partialVars.resolver().find(name).orElse(null), - EMPTY_FUNCTION_RESOLVER, - partialVars, - listener); + return program.trace( + (name) -> partialVars.resolver().find(name).orElse(null), + CelFunctionResolver.EMPTY, + partialVars, + listener); } @Override - public Object advanceEvaluation(UnknownContext context) throws CelEvaluationException { + public Object advanceEvaluation(UnknownContext context) { throw new UnsupportedOperationException("Unsupported operation."); } }; @@ -347,6 +318,10 @@ public abstract Builder setAsyncEvaluationOptions( @Override public abstract CelValueProvider valueProvider(); + abstract CelAsyncEvaluationOptions asyncEvaluationOptions(); + + abstract Optional asyncExecutor(); + abstract CelStandardFunctions standardFunctions(); abstract ExtensionRegistry extensionRegistry(); @@ -604,7 +579,9 @@ public CelRuntime build() { celValueConverter, container(), options(), - lateBoundFunctionNamesBuilder().build()); + lateBoundFunctionNamesBuilder().build(), + asyncEvaluationOptions(), + asyncExecutor().orElse(null)); setPlanner(planner); setFunctionBindings(ImmutableMap.copyOf(mutableFunctionBindings)); diff --git a/runtime/src/main/java/dev/cel/runtime/LiteRuntimeImpl.java b/runtime/src/main/java/dev/cel/runtime/LiteRuntimeImpl.java index 6572621a6..875626e81 100644 --- a/runtime/src/main/java/dev/cel/runtime/LiteRuntimeImpl.java +++ b/runtime/src/main/java/dev/cel/runtime/LiteRuntimeImpl.java @@ -229,7 +229,10 @@ public CelLiteRuntime build() { celValueProvider.celValueConverter(), container, celOptions, - lateBoundFunctionNamesBuilder.build()); + lateBoundFunctionNamesBuilder.build(), + // TODO: Support async eval in lite runtime. + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); return new LiteRuntimeImpl( planner, diff --git a/runtime/src/main/java/dev/cel/runtime/RuntimeEquality.java b/runtime/src/main/java/dev/cel/runtime/RuntimeEquality.java index 56a8761cd..a9b607e2b 100644 --- a/runtime/src/main/java/dev/cel/runtime/RuntimeEquality.java +++ b/runtime/src/main/java/dev/cel/runtime/RuntimeEquality.java @@ -135,12 +135,12 @@ public Optional findInMap(Map map, Object index) { * comparable even if they are not of the same type, where type differences are usually trivially * false. */ - @SuppressWarnings({"rawtypes", "unchecked"}) + @SuppressWarnings({"rawtypes", "unchecked", "ReferenceEquality"}) public boolean objectEquals(Object x, Object y) { if (celOptions.disableCelStandardEquality()) { return Objects.equals(x, y); } - if (x == y) { + if (x == y && !isNan(x)) { return true; } x = runtimeHelpers.adaptValue(x); @@ -237,7 +237,9 @@ public int hashCode(Object object) { object = runtimeHelpers.adaptValue(object); if (object instanceof Number) { - return Double.hashCode(((Number) object).doubleValue()); + double value = ((Number) object).doubleValue(); + // Normalize -0.0 to 0.0. objectEquals reports the two as equal, so they must hash alike. + return Double.hashCode(value == 0.0d ? 0.0d : value); } if (object instanceof Iterable) { int h = 1; @@ -276,6 +278,10 @@ private static Optional unsignedToLongLossless(UnsignedLong v) { return Optional.empty(); } + private static boolean isNan(Object value) { + return value instanceof Number && Double.isNaN(((Number) value).doubleValue()); + } + RuntimeEquality(RuntimeHelpers runtimeHelpers, CelOptions celOptions) { this.runtimeHelpers = runtimeHelpers; this.celOptions = celOptions; diff --git a/runtime/src/main/java/dev/cel/runtime/planner/AsyncCallRecord.java b/runtime/src/main/java/dev/cel/runtime/planner/AsyncCallRecord.java new file mode 100644 index 000000000..d45a6a95c --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/planner/AsyncCallRecord.java @@ -0,0 +1,289 @@ +// 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.util.concurrent.ListenableFuture; +import javax.annotation.concurrent.ThreadSafe; +import dev.cel.runtime.CelAsyncCall; +import dev.cel.runtime.CelAsyncFunctionOverload; +import dev.cel.runtime.RuntimeEquality; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import org.jspecify.annotations.Nullable; + +/** Tracks the execution state and result of a single asynchronous function call. */ +@ThreadSafe +// CEL-Internal-4 +final class AsyncCallRecord implements CelAsyncCall { + + enum State { + NOT_STARTED, + RUNNING, + SUCCESS, + FAILURE, + CANCELLED + } + + // Type markers keep values of different kinds from colliding in the bucket hash, e.g. the + // string "NaN" and the double NaN. Collisions remain harmless because matches() disambiguates the + // bucket. + private static final int STRING_HASH_MARKER = 's'; + private static final int BOOL_HASH_MARKER = 'b'; + private static final int NUMBER_HASH_MARKER = 'n'; + private static final int COMPLEX_HASH_MARKER = 'x'; + + private final long callId; + private final long exprId; + private final String functionName; + private final String overloadId; + + @SuppressWarnings("Immutable") // Array not mutated after construction + private final Object[] args; + + private final CelAsyncFunctionOverload overload; + + private final Object lock = new Object(); + private final AtomicBoolean completionReported = new AtomicBoolean(false); + private volatile State state = State.NOT_STARTED; + private volatile @Nullable Object result; + private volatile @Nullable Throwable error; + private volatile @Nullable ListenableFuture inFlightFuture; + + static AsyncCallRecord create( + long callId, + long exprId, + String functionName, + String overloadId, + Object[] args, + CelAsyncFunctionOverload overload) { + return new AsyncCallRecord(callId, exprId, functionName, overloadId, args, overload); + } + + /** + * Computes the bucket hash under which a call is tracked. + * + *

This is a bucketing hint, not an identity: calls that {@link #matches} considers identical + * hash alike, but distinct calls may share a bucket. Resolve the exact call via {@link #matches}. + */ + static int hashCall(long exprId, String overloadId, Object[] args) { + int result = 31 * Long.hashCode(exprId) + overloadId.hashCode(); + for (Object arg : args) { + result = result * 31 + hashArg(arg); + } + return result; + } + + /** + * Returns whether this record tracks a call to the same expression node, function, overload, and + * arguments. + * + *

Arguments are compared under CEL equality, except that NaN compares equal to itself so that + * a node re-evaluated with a NaN argument can find its existing record. + */ + boolean matches( + long exprId, + String functionName, + String overloadId, + Object[] args, + RuntimeEquality runtimeEquality) { + if (this.exprId != exprId + || !this.functionName.equals(functionName) + || !this.overloadId.equals(overloadId) + || this.args.length != args.length) { + return false; + } + for (int i = 0; i < this.args.length; i++) { + Object arg = this.args[i]; + Object otherArg = args[i]; + if (!runtimeEquality.objectEquals(arg, otherArg) && !(isNan(arg) && isNan(otherArg))) { + return false; + } + } + return true; + } + + @Override + public long callId() { + return callId; + } + + @Override + public long exprId() { + return exprId; + } + + @Override + public String functionName() { + return functionName; + } + + @Override + public String overloadId() { + return overloadId; + } + + /** + * Transitions the call state from {@link State#NOT_STARTED} to {@link State#RUNNING}. + * + * @return true if the transition succeeded, false if the call was already running, completed, or + * cancelled. + */ + boolean markRunning() { + synchronized (lock) { + if (state != State.NOT_STARTED) { + return false; + } + state = State.RUNNING; + return true; + } + } + + void setInFlightFuture(ListenableFuture future) { + checkNotNull(future); + boolean shouldCancel; + synchronized (lock) { + inFlightFuture = future; + shouldCancel = (state == State.CANCELLED && !future.isDone()); + } + if (shouldCancel) { + future.cancel(/* mayInterruptIfRunning= */ false); + } + } + + boolean cancelInFlight() { + ListenableFuture futureToCancel = null; + synchronized (lock) { + if (!isPending()) { + return false; + } + state = State.CANCELLED; + ListenableFuture future = inFlightFuture; + if (future != null && !future.isDone()) { + futureToCancel = future; + } + } + if (futureToCancel != null) { + futureToCancel.cancel(/* mayInterruptIfRunning= */ false); + } + return true; + } + + /** + * Claims the right to report this call's completion, returning true for the first caller only. + * + *

Tracked separately from {@link State} because a call cancelled after dispatch still holds a + * concurrency permit and must release it exactly once. + */ + boolean markCompletionReported() { + return completionReported.compareAndSet(false, true); + } + + boolean isCancelled() { + return state == State.CANCELLED; + } + + boolean complete(@Nullable Object result) { + synchronized (lock) { + if (!isPending()) { + return false; + } + this.result = result; + state = State.SUCCESS; + return true; + } + } + + boolean fail(Throwable error) { + checkNotNull(error); + synchronized (lock) { + if (!isPending()) { + return false; + } + this.error = error; + state = State.FAILURE; + return true; + } + } + + Object[] args() { + return args.clone(); + } + + CelAsyncFunctionOverload overload() { + return overload; + } + + State state() { + return state; + } + + /** + * Returns the completed result, if present. + * + *

Note: If a call completed successfully with a {@code null} value, this method returns {@code + * Optional.empty()}. Callers should check {@link #state()} to distinguish between a call that has + * not completed and one that succeeded with {@code null}. + */ + Optional result() { + return Optional.ofNullable(result); + } + + Optional error() { + return Optional.ofNullable(error); + } + + private static int hashArg(@Nullable Object arg) { + if (arg instanceof String) { + return STRING_HASH_MARKER * 31 + arg.hashCode(); + } + if (arg instanceof Boolean) { + return BOOL_HASH_MARKER * 31 + arg.hashCode(); + } + if (arg instanceof Number) { + // Hash int, uint, and double through a common double representation so that values CEL + // considers equal (1 == 1u == 1.0) share a bucket. NaN needs no special case because + // Double.hashCode(NaN) is a constant across all double and float NaN representations. + double value = ((Number) arg).doubleValue(); + // Normalize -0.0 to 0.0, which CEL considers equal to 0.0. + return NUMBER_HASH_MARKER * 31 + Double.hashCode(value == 0.0d ? 0.0d : value); + } + return COMPLEX_HASH_MARKER; + } + + private static boolean isNan(@Nullable Object value) { + return value instanceof Number && Double.isNaN(((Number) value).doubleValue()); + } + + private boolean isPending() { + return state == State.NOT_STARTED || state == State.RUNNING; + } + + private AsyncCallRecord( + long callId, + long exprId, + String functionName, + String overloadId, + Object[] args, + CelAsyncFunctionOverload overload) { + this.callId = callId; + this.exprId = exprId; + this.functionName = checkNotNull(functionName); + this.overloadId = checkNotNull(overloadId); + this.args = checkNotNull(args).clone(); + this.overload = checkNotNull(overload); + } +} diff --git a/runtime/src/main/java/dev/cel/runtime/planner/AsyncCallStateTracker.java b/runtime/src/main/java/dev/cel/runtime/planner/AsyncCallStateTracker.java new file mode 100644 index 000000000..ee4b8a5a4 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/planner/AsyncCallStateTracker.java @@ -0,0 +1,298 @@ +// 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.util.concurrent.MoreExecutors.directExecutor; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import javax.annotation.concurrent.ThreadSafe; +import dev.cel.common.exceptions.CelRuntimeException; +import dev.cel.common.values.CelValueConverter; +import dev.cel.runtime.AccumulatedUnknowns; +import dev.cel.runtime.CelAsyncFunctionOverload; +import dev.cel.runtime.CelAsyncObserver; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.InterpreterUtil; +import dev.cel.runtime.RuntimeEquality; +import java.util.Set; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicLong; +import org.jspecify.annotations.Nullable; + +/** + * Tracks the registry and cache of all asynchronous function calls made during an expression + * evaluation. + */ +@ThreadSafe +// CEL-Internal-4 +final class AsyncCallStateTracker { + private final AtomicLong callIdGenerator = new AtomicLong(1); + private final ConcurrentMap> recordsByBucket = + new ConcurrentHashMap<>(); + private final ConcurrentMap recordsById = new ConcurrentHashMap<>(); + private final RuntimeEquality runtimeEquality; + + static AsyncCallStateTracker create(RuntimeEquality runtimeEquality) { + return new AsyncCallStateTracker(runtimeEquality); + } + + /** + * Returns the resolved result for a previously completed call matching {@code (exprId, + * overloadId, args)}, throws a runtime exception if the call failed or was cancelled, or + * registers and returns an {@link AccumulatedUnknowns} with the call's tracking ID if pending. + */ + Object recordOrGet( + long exprId, + String functionName, + String overloadId, + Object[] args, + CelAsyncFunctionOverload overload, + CelValueConverter celValueConverter) { + checkNotNull(functionName); + checkNotNull(overloadId); + checkNotNull(args); + checkNotNull(overload); + checkNotNull(celValueConverter); + int bucketKey = AsyncCallRecord.hashCall(exprId, overloadId, args); + CopyOnWriteArrayList bucket = recordsByBucket.get(bucketKey); + if (bucket != null) { + for (int i = 0; i < bucket.size(); i++) { + AsyncCallRecord existing = bucket.get(i); + if (existing.matches(exprId, functionName, overloadId, args, runtimeEquality)) { + return resolveRecord(existing, celValueConverter); + } + } + } + + bucket = recordsByBucket.computeIfAbsent(bucketKey, k -> new CopyOnWriteArrayList<>()); + AsyncCallRecord record = null; + synchronized (bucket) { + for (int i = 0; i < bucket.size(); i++) { + AsyncCallRecord existing = bucket.get(i); + if (existing.matches(exprId, functionName, overloadId, args, runtimeEquality)) { + record = existing; + break; + } + } + if (record == null) { + long callId = callIdGenerator.getAndIncrement(); + record = AsyncCallRecord.create(callId, exprId, functionName, overloadId, args, overload); + recordsById.put(callId, record); + bucket.add(record); + } + } + + return resolveRecord(record, celValueConverter); + } + + /** + * Launches every not-yet-started call in {@code requiredCallIds}, subject to {@code gate} + * admission control. + * + *

{@code executor} must run or reject each task; one that silently discards tasks strands the + * call's concurrency permit. + */ + void dispatchPendingCalls( + Set requiredCallIds, + Executor executor, + AsyncGate gate, + AsyncCompletionCoordinator coordinator, + @Nullable CelAsyncObserver observer) { + checkNotNull(requiredCallIds); + checkNotNull(executor); + checkNotNull(gate); + checkNotNull(coordinator); + for (Long callId : ImmutableList.sortedCopyOf(requiredCallIds)) { + AsyncCallRecord record = recordsById.get(callId); + if (record != null && record.state() == AsyncCallRecord.State.NOT_STARTED) { + tryLaunch(record, executor, gate, coordinator, observer); + } + } + } + + @VisibleForTesting + void tryLaunch( + AsyncCallRecord record, + Executor executor, + AsyncGate gate, + AsyncCompletionCoordinator coordinator, + @Nullable CelAsyncObserver observer) { + if (!gate.tryAcquire()) { + return; + } + if (!record.markRunning()) { + gate.release(); + return; + } + + try { + if (observer != null) { + observer.onCallStarted(record, ImmutableList.copyOf(record.args())); + } + executor.execute(() -> executeAsyncCall(record, coordinator, observer)); + } catch (RuntimeException e) { + handleFailure(record, e, coordinator, observer); + } + } + + private static void executeAsyncCall( + AsyncCallRecord record, + AsyncCompletionCoordinator coordinator, + @Nullable CelAsyncObserver observer) { + ListenableFuture future; + try { + if (record.isCancelled()) { + throw new CancellationException("Async call was cancelled before dispatch"); + } + future = + checkNotNull( + record.overload().applyAsync(record.args()), + "Async function '%s' returned a null ListenableFuture", + record.functionName()); + record.setInFlightFuture(future); + } catch (CelEvaluationException | RuntimeException e) { + handleFailure(record, e, coordinator, observer); + return; + } + + Futures.addCallback( + future, + new FutureCallback() { + @Override + public void onSuccess(Object result) { + handleSuccess(record, result, coordinator, observer); + } + + @Override + public void onFailure(Throwable t) { + handleFailure(record, t, coordinator, observer); + } + }, + directExecutor()); + } + + private Object resolveRecord(AsyncCallRecord record, CelValueConverter celValueConverter) { + switch (record.state()) { + case SUCCESS: + Object rawResult = record.result().orElseThrow(AssertionError::new); + return InterpreterUtil.maybeAdaptToAccumulatedUnknowns( + celValueConverter.maybeUnwrap(celValueConverter.toRuntimeValue(rawResult))); + case FAILURE: + Throwable error = record.error().orElseThrow(AssertionError::new); + if (error instanceof CelRuntimeException) { + throw (CelRuntimeException) error; + } + String errorMessage = + error.getMessage() != null ? error.getMessage() : error.getClass().getSimpleName(); + throw new IllegalArgumentException( + String.format("Async function '%s' failed: %s", record.functionName(), errorMessage), + error); + case RUNNING: + case NOT_STARTED: + return AccumulatedUnknowns.createForAsyncCall(record.exprId(), record.callId()); + case CANCELLED: + throw new CancellationException( + String.format("Async function '%s' was cancelled", record.functionName())); + } + throw new AssertionError("Unexpected record state: " + record.state()); + } + + boolean hasInFlightCalls() { + for (AsyncCallRecord record : recordsById.values()) { + if (record.state() == AsyncCallRecord.State.RUNNING) { + return true; + } + } + return false; + } + + void cancelInFlight() { + for (AsyncCallRecord record : recordsById.values()) { + record.cancelInFlight(); + } + } + + private static void handleSuccess( + AsyncCallRecord record, + @Nullable Object result, + AsyncCompletionCoordinator coordinator, + @Nullable CelAsyncObserver observer) { + if (result == null) { + handleFailure( + record, + new NullPointerException( + String.format("Async function '%s' returned a null result", record.functionName())), + coordinator, + observer); + return; + } + record.complete(result); + reportCompletion(record, result, /* error= */ null, coordinator, observer); + } + + private static void handleFailure( + AsyncCallRecord record, + Throwable error, + AsyncCompletionCoordinator coordinator, + @Nullable CelAsyncObserver observer) { + record.fail(error); + reportCompletion(record, /* result= */ null, error, coordinator, observer); + } + + /** + * Notifies the observer and completion coordinator of a launched call's terminal outcome at most + * once. + */ + private static void reportCompletion( + AsyncCallRecord record, + @Nullable Object result, + @Nullable Throwable error, + AsyncCompletionCoordinator coordinator, + @Nullable CelAsyncObserver observer) { + if (!record.markCompletionReported()) { + return; + } + try { + if (observer != null) { + observer.onCallFinished(record, result, error); + } + } finally { + coordinator.callCompleted(record); + } + } + + @VisibleForTesting + ConcurrentMap> recordsByBucket() { + return recordsByBucket; + } + + @VisibleForTesting + ConcurrentMap recordsById() { + return recordsById; + } + + private AsyncCallStateTracker(RuntimeEquality runtimeEquality) { + this.runtimeEquality = checkNotNull(runtimeEquality); + } +} 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 d838e8d53..a8882f539 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel +++ b/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel @@ -17,6 +17,7 @@ java_library( ":attribute", ":error_metadata", ":eval_and", + ":eval_async_call", ":eval_attribute", ":eval_binary", ":eval_block", @@ -54,10 +55,11 @@ java_library( "//common/types:type_providers", "//common/values", "//common/values:cel_value_provider", + "//runtime:async_options", "//runtime:dispatcher", "//runtime:evaluation_exception", "//runtime:evaluation_exception_builder", - "//runtime:program", + "//runtime:function_overload", "//runtime:resolved_overload", "@maven//:com_google_code_findbugs_annotations", "@maven//:com_google_errorprone_error_prone_annotations", @@ -81,6 +83,7 @@ java_library( "//common/exceptions:runtime_exception", "//common/values", "//runtime:activation", + "//runtime:async_options", "//runtime:evaluation_exception", "//runtime:evaluation_exception_builder", "//runtime:evaluation_listener", @@ -89,7 +92,6 @@ java_library( "//runtime:interpreter_util", "//runtime:partial_vars", "//runtime:program", - "//runtime:resolved_overload", "//runtime:variable_resolver", "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:com_google_guava_guava", @@ -217,6 +219,46 @@ java_library( ], ) +java_library( + name = "async_call_state_tracker", + srcs = [ + "AsyncCallRecord.java", + "AsyncCallStateTracker.java", + ], + tags = [ + ], + deps = [ + ":async_completion_coordinator", + ":async_gate", + "//common/exceptions:runtime_exception", + "//common/values", + "//runtime:accumulated_unknowns", + "//runtime:async_call", + "//runtime:async_observer", + "//runtime:evaluation_exception", + "//runtime:function_overload", + "//runtime:interpreter_util", + "//runtime:runtime_equality", + "@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 = "eval_async_call", + srcs = ["EvalAsyncCall.java"], + deps = [ + ":planned_interpretable", + "//common/ast", + "//runtime:evaluation_exception", + "//runtime:interpretable", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", + ], +) + java_library( name = "activation_wrapper", srcs = ["ActivationWrapper.java"], @@ -580,6 +622,7 @@ cel_android_library( ":attribute_android", ":error_metadata_android", ":eval_and_android", + ":eval_async_call_android", ":eval_attribute_android", ":eval_binary_android", ":eval_block_android", @@ -617,11 +660,12 @@ cel_android_library( "//common/types:types_android", "//common/values:cel_value_provider_android", "//common/values:values_android", + "//runtime:async_options_android", "//runtime:dispatcher_android", "//runtime:evaluation_exception", "//runtime:evaluation_exception_builder", + "//runtime:function_overload_android", "//runtime:resolved_overload_android", - "//runtime/src/main/java/dev/cel/runtime:program_android", "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:org_jspecify_jspecify", "@maven_android//:com_google_guava_guava", @@ -631,6 +675,8 @@ cel_android_library( cel_android_library( name = "planned_program_android", srcs = ["PlannedProgram.java"], + tags = [ + ], deps = [ ":error_metadata_android", ":localized_evaluation_exception_android", @@ -641,10 +687,10 @@ cel_android_library( "//common/exceptions:runtime_exception", "//common/values:values_android", "//runtime:activation_android", + "//runtime:async_options_android", "//runtime:evaluation_exception", "//runtime:evaluation_exception_builder", "//runtime:interpretable_android", - "//runtime:resolved_overload_android", "//runtime:variable_resolver", "//runtime/src/main/java/dev/cel/runtime:evaluation_listener_android", "//runtime/src/main/java/dev/cel/runtime:function_resolver_android", @@ -747,6 +793,78 @@ 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//: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 = "async_call_state_tracker_android", + srcs = [ + "AsyncCallRecord.java", + "AsyncCallStateTracker.java", + ], + tags = [ + ], + deps = [ + ":async_completion_coordinator_android", + ":async_gate_android", + "//common/exceptions:runtime_exception", + "//common/values:values_android", + "//runtime:async_call_android", + "//runtime:async_observer_android", + "//runtime:evaluation_exception", + "//runtime:function_overload_android", + "//runtime:interpreter_util_android", + "//runtime:runtime_equality_android", + "//runtime/src/main/java/dev/cel/runtime:accumulated_unknowns_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 = "eval_async_call_android", + srcs = ["EvalAsyncCall.java"], + deps = [ + ":planned_interpretable_android", + "//common/ast:ast_android", + "//runtime:evaluation_exception", + "//runtime:interpretable_android", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven_android//:com_google_guava_guava", + ], +) + cel_android_library( name = "activation_wrapper_android", srcs = ["ActivationWrapper.java"], diff --git a/runtime/src/main/java/dev/cel/runtime/planner/EvalAsyncCall.java b/runtime/src/main/java/dev/cel/runtime/planner/EvalAsyncCall.java new file mode 100644 index 000000000..3b88564aa --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/planner/EvalAsyncCall.java @@ -0,0 +1,47 @@ +// 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.errorprone.annotations.Immutable; +import dev.cel.common.ast.CelExpr; +import dev.cel.runtime.CelEvaluationException; +import dev.cel.runtime.GlobalResolver; + +/** Evaluates an asynchronous function call within a planned program. */ +@Immutable +final class EvalAsyncCall extends PlannedInterpretable { + + private final String functionName; + + static EvalAsyncCall create(CelExpr expr, String functionName) { + return new EvalAsyncCall(expr, functionName); + } + + @Override + Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) throws CelEvaluationException { + throw new CelEvaluationException( + String.format( + "Async function '%s' evaluated in synchronous mode. Asynchronous functions are only" + + " supported via evalAsync.", + functionName)); + } + + private EvalAsyncCall(CelExpr expr, String functionName) { + super(expr); + this.functionName = checkNotNull(functionName); + } +} diff --git a/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java b/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java index f7f3d7f01..2f007923e 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java @@ -16,23 +16,23 @@ import com.google.auto.value.AutoValue; import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.ListeningExecutorService; import com.google.errorprone.annotations.Immutable; import dev.cel.common.CelOptions; import dev.cel.common.annotations.Internal; import dev.cel.common.exceptions.CelRuntimeException; import dev.cel.common.values.ErrorValue; import dev.cel.runtime.Activation; +import dev.cel.runtime.CelAsyncEvaluationOptions; import dev.cel.runtime.CelEvaluationException; import dev.cel.runtime.CelEvaluationExceptionBuilder; import dev.cel.runtime.CelEvaluationListener; import dev.cel.runtime.CelFunctionResolver; -import dev.cel.runtime.CelResolvedOverload; import dev.cel.runtime.CelVariableResolver; import dev.cel.runtime.GlobalResolver; import dev.cel.runtime.InterpreterUtil; import dev.cel.runtime.PartialVars; import dev.cel.runtime.Program; -import java.util.Collection; import java.util.Map; import java.util.Optional; import org.jspecify.annotations.Nullable; @@ -47,33 +47,37 @@ @AutoValue public abstract class PlannedProgram implements Program { - 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 abstract PlannedInterpretable interpretable(); + abstract PlannedInterpretable interpretable(); abstract ErrorMetadata metadata(); public abstract CelOptions options(); + // CelAsyncEvaluationOptions is an immutable value object. + @SuppressWarnings("Immutable") + @AutoValue.CopyAnnotations + abstract CelAsyncEvaluationOptions asyncOptions(); + + // The executor service is an externally managed, thread-safe asynchronous execution pool. + @SuppressWarnings("Immutable") + @AutoValue.CopyAnnotations + abstract Optional asyncExecutor(); + + static PlannedProgram create( + PlannedInterpretable interpretable, + ErrorMetadata metadata, + CelOptions options, + CelAsyncEvaluationOptions asyncOptions, + @Nullable ListeningExecutorService asyncExecutor) { + return new AutoValue_PlannedProgram( + interpretable, metadata, options, asyncOptions, Optional.ofNullable(asyncExecutor)); + } + @Override public Object eval() throws CelEvaluationException { return evalOrThrow( - interpretable(), GlobalResolver.EMPTY, - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, /* partialVars= */ null, /* listener= */ null); } @@ -81,9 +85,8 @@ public Object eval() throws CelEvaluationException { @Override public Object eval(Map mapValue) throws CelEvaluationException { return evalOrThrow( - interpretable(), Activation.copyOf(mapValue), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, /* partialVars= */ null, /* listener= */ null); } @@ -92,7 +95,6 @@ public Object eval(Map mapValue) throws CelEvaluationException { public Object eval(Map mapValue, CelFunctionResolver lateBoundFunctionResolver) throws CelEvaluationException { return evalOrThrow( - interpretable(), Activation.copyOf(mapValue), lateBoundFunctionResolver, /* partialVars= */ null, @@ -102,9 +104,8 @@ public Object eval(Map mapValue, CelFunctionResolver lateBoundFunctio @Override public Object eval(CelVariableResolver resolver) throws CelEvaluationException { return evalOrThrow( - interpretable(), (name) -> resolver.find(name).orElse(null), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, /* partialVars= */ null, /* listener= */ null); } @@ -113,7 +114,6 @@ public Object eval(CelVariableResolver resolver) throws CelEvaluationException { public Object eval(CelVariableResolver resolver, CelFunctionResolver lateBoundFunctionResolver) throws CelEvaluationException { return evalOrThrow( - interpretable(), (name) -> resolver.find(name).orElse(null), lateBoundFunctionResolver, /* partialVars= */ null, @@ -123,9 +123,8 @@ public Object eval(CelVariableResolver resolver, CelFunctionResolver lateBoundFu @Override public Object eval(PartialVars partialVars) throws CelEvaluationException { return evalOrThrow( - interpretable(), (name) -> partialVars.resolver().find(name).orElse(null), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, partialVars, /* listener= */ null); } @@ -163,7 +162,6 @@ public ListenableFuture evalAsync(PartialVars partialVars) { } public Object evalOrThrow( - PlannedInterpretable interpretable, GlobalResolver resolver, CelFunctionResolver functionResolver, @Nullable PartialVars partialVars, @@ -172,7 +170,7 @@ public Object evalOrThrow( try { ExecutionFrame frame = ExecutionFrame.create(functionResolver, options(), partialVars, listener); - Object evalResult = interpretable.eval(resolver, frame); + Object evalResult = interpretable().eval(resolver, frame); if (evalResult instanceof ErrorValue) { ErrorValue errorValue = (ErrorValue) evalResult; throw newCelEvaluationException(errorValue.exprId(), errorValue.value()); @@ -180,20 +178,23 @@ public Object evalOrThrow( return InterpreterUtil.maybeAdaptToCelUnknownSet(evalResult); } catch (RuntimeException e) { - throw newCelEvaluationException(interpretable.expr().id(), e); + throw newCelEvaluationException(interpretable().expr().id(), e); } } public Object trace( GlobalResolver resolver, CelFunctionResolver functionResolver, - PartialVars partialVars, - CelEvaluationListener listener) + @Nullable PartialVars partialVars, + @Nullable CelEvaluationListener listener) throws CelEvaluationException { - return evalOrThrow(interpretable(), resolver, functionResolver, partialVars, listener); + return evalOrThrow(resolver, functionResolver, partialVars, listener); } - private CelEvaluationException newCelEvaluationException(long exprId, Exception e) { + private CelEvaluationException newCelEvaluationException(long exprId, Throwable e) { + if (e instanceof CelEvaluationException) { + return (CelEvaluationException) e; + } CelEvaluationExceptionBuilder builder; if (e instanceof LocalizedEvaluationException) { // Use the localized expr ID (most specific error location) @@ -201,8 +202,7 @@ private CelEvaluationException newCelEvaluationException(long exprId, Exception exprId = localized.exprId(); Throwable cause = localized.getCause(); if (cause instanceof CelRuntimeException) { - builder = - CelEvaluationExceptionBuilder.newBuilder((CelRuntimeException) localized.getCause()); + builder = CelEvaluationExceptionBuilder.newBuilder((CelRuntimeException) cause); } else { builder = CelEvaluationExceptionBuilder.newBuilder(cause.getMessage()).setCause(cause); } @@ -220,8 +220,5 @@ private CelEvaluationException newCelEvaluationException(long exprId, Exception return builder.setMetadata(metadata(), exprId).build(); } - static Program create( - PlannedInterpretable interpretable, ErrorMetadata metadata, CelOptions options) { - return new AutoValue_PlannedProgram(interpretable, metadata, options); - } + PlannedProgram() {} } diff --git a/runtime/src/main/java/dev/cel/runtime/planner/ProgramPlanner.java b/runtime/src/main/java/dev/cel/runtime/planner/ProgramPlanner.java index 23a6e5dec..d538e1c43 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/ProgramPlanner.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/ProgramPlanner.java @@ -21,6 +21,7 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; +import com.google.common.util.concurrent.ListeningExecutorService; import com.google.errorprone.annotations.CheckReturnValue; import com.google.errorprone.annotations.Immutable; import dev.cel.common.CelAbstractSyntaxTree; @@ -47,11 +48,13 @@ import dev.cel.common.types.TypeType; import dev.cel.common.values.CelValueConverter; import dev.cel.common.values.CelValueProvider; +import dev.cel.runtime.CelAsyncEvaluationOptions; +import dev.cel.runtime.CelAsyncFunctionOverload; import dev.cel.runtime.CelEvaluationException; import dev.cel.runtime.CelEvaluationExceptionBuilder; import dev.cel.runtime.CelResolvedOverload; import dev.cel.runtime.DefaultDispatcher; -import dev.cel.runtime.Program; +import java.util.Arrays; import java.util.HashMap; import java.util.NoSuchElementException; import java.util.Optional; @@ -73,11 +76,19 @@ public final class ProgramPlanner { private final CelValueConverter celValueConverter; private final ImmutableSet lateBoundFunctionNames; + // CelAsyncEvaluationOptions is an immutable value object. + @SuppressWarnings("Immutable") + private final CelAsyncEvaluationOptions asyncOptions; + + // The executor service is an externally managed, thread-safe asynchronous execution pool. + @SuppressWarnings("Immutable") + private final @Nullable ListeningExecutorService asyncExecutor; + /** - * Plans a {@link Program} from the provided parsed-only or type-checked {@link + * Plans a {@link PlannedProgram} from the provided parsed-only or type-checked {@link * CelAbstractSyntaxTree}. */ - public Program plan(CelAbstractSyntaxTree ast) throws CelEvaluationException { + public PlannedProgram plan(CelAbstractSyntaxTree ast) throws CelEvaluationException { PlannedInterpretable plannedInterpretable; ErrorMetadata errorMetadata = ErrorMetadata.create(ast.getSource().getPositionsMap(), ast.getSource().getDescription()); @@ -94,7 +105,8 @@ public Program plan(CelAbstractSyntaxTree ast) throws CelEvaluationException { .build(); } - return PlannedProgram.create(plannedInterpretable, errorMetadata, options); + return PlannedProgram.create( + plannedInterpretable, errorMetadata, options, asyncOptions, asyncExecutor); } private PlannedInterpretable plan(CelExpr celExpr, PlannerContext ctx) { @@ -117,9 +129,8 @@ private PlannedInterpretable plan(CelExpr celExpr, PlannerContext ctx) { return planComprehension(celExpr, ctx); case NOT_SET: throw new UnsupportedOperationException("Unsupported kind: " + celExpr.getKind()); - default: - throw new UnsupportedOperationException("Unexpected kind: " + celExpr.getKind()); } + throw new UnsupportedOperationException("Unexpected kind: " + celExpr.getKind()); } private PlannedInterpretable planSelect(CelExpr celExpr, PlannerContext ctx) { @@ -320,6 +331,10 @@ private PlannedInterpretable planCall(CelExpr expr, PlannerContext ctx) { expr, functionName, overloadIds, evaluatedArgs, celValueConverter); } + if (resolvedOverload.getDefinition() instanceof CelAsyncFunctionOverload) { + return EvalAsyncCall.create(expr, functionName); + } + switch (argCount) { case 0: return EvalZeroArity.create(expr, functionName, resolvedOverload, celValueConverter); @@ -353,9 +368,7 @@ private PlannedInterpretable planBlock(CelBlock celBlock, PlannerContext ctx) { ImmutableList indices = celBlock.indices(); PlannedInterpretable[] slotExprs = new PlannedInterpretable[indices.size()]; - for (int i = 0; i < slotExprs.length; i++) { - slotExprs[i] = plan(indices.get(i), ctx); - } + Arrays.setAll(slotExprs, i -> plan(indices.get(i), ctx)); PlannedInterpretable resultExpr = plan(celBlock.result(), ctx); return EvalBlock.create(celBlock.expr(), slotExprs, resultExpr); } @@ -695,15 +708,18 @@ private boolean isLocalVar(String name) { return localVars.containsKey(name); } - private PlannerContext(CelAbstractSyntaxTree ast) { - this.ast = checkNotNull(ast); - } - static PlannerContext create(CelAbstractSyntaxTree ast) { return new PlannerContext(ast); } + + private PlannerContext(CelAbstractSyntaxTree ast) { + this.ast = checkNotNull(ast); + } } + // Internal API: ProgramPlanner is marked @Internal for the CEL runtime engine and requires all + // engine dependencies for planning. + @SuppressWarnings("TooManyParameters") public static ProgramPlanner newPlanner( CelTypeProvider typeProvider, CelValueProvider valueProvider, @@ -711,7 +727,9 @@ public static ProgramPlanner newPlanner( CelValueConverter celValueConverter, CelContainer container, CelOptions options, - ImmutableSet lateBoundFunctionNames) { + ImmutableSet lateBoundFunctionNames, + CelAsyncEvaluationOptions asyncOptions, + @Nullable ListeningExecutorService asyncExecutor) { return new ProgramPlanner( typeProvider, valueProvider, @@ -719,7 +737,9 @@ public static ProgramPlanner newPlanner( celValueConverter, container, options, - lateBoundFunctionNames); + lateBoundFunctionNames, + asyncOptions, + asyncExecutor); } private ProgramPlanner( @@ -729,7 +749,9 @@ private ProgramPlanner( CelValueConverter celValueConverter, CelContainer container, CelOptions options, - ImmutableSet lateBoundFunctionNames) { + ImmutableSet lateBoundFunctionNames, + CelAsyncEvaluationOptions asyncOptions, + @Nullable ListeningExecutorService asyncExecutor) { this.typeProvider = typeProvider; this.valueProvider = valueProvider; this.dispatcher = dispatcher; @@ -737,6 +759,8 @@ private ProgramPlanner( this.container = container; this.options = options; this.lateBoundFunctionNames = lateBoundFunctionNames; + this.asyncOptions = checkNotNull(asyncOptions); + this.asyncExecutor = asyncExecutor; this.attributeFactory = AttributeFactory.newAttributeFactory(container, typeProvider, celValueConverter); } diff --git a/runtime/src/test/java/dev/cel/runtime/RuntimeEqualityTest.java b/runtime/src/test/java/dev/cel/runtime/RuntimeEqualityTest.java index 00e55873c..50c39ed58 100644 --- a/runtime/src/test/java/dev/cel/runtime/RuntimeEqualityTest.java +++ b/runtime/src/test/java/dev/cel/runtime/RuntimeEqualityTest.java @@ -37,6 +37,7 @@ public void objectEquals_and_hashCode() { assertEqualityAndHashCode(runtimeEquality, 2, 2L); assertEqualityAndHashCode(runtimeEquality, 3, 3.0); assertEqualityAndHashCode(runtimeEquality, 4, UnsignedLong.valueOf(4)); + assertEqualityAndHashCode(runtimeEquality, 0.0d, -0.0d); assertEqualityAndHashCode( runtimeEquality, ImmutableList.of(1, 2, 3), @@ -56,12 +57,24 @@ private void assertEqualityAndHashCode(RuntimeEquality runtimeEquality, Object o public void objectEquals_messageLite_throws() { RuntimeEquality runtimeEquality = RuntimeEquality.create(RuntimeHelpers.create(), CelOptions.DEFAULT); + TestAllTypes.Builder builder = TestAllTypes.newBuilder(); + TestAllTypes defaultInstance = TestAllTypes.getDefaultInstance(); // Unimplemented until CelLiteDescriptor is available. - assertThrows( - UnsupportedOperationException.class, - () -> - runtimeEquality.objectEquals( - TestAllTypes.newBuilder(), TestAllTypes.getDefaultInstance())); + UnsupportedOperationException e = + assertThrows( + UnsupportedOperationException.class, + () -> runtimeEquality.objectEquals(builder, defaultInstance)); + + assertThat(e).hasMessageThat().contains("Not implemented yet"); + } + + @Test + public void objectEquals_nanWithIdenticalReference_returnsFalse() { + RuntimeEquality runtimeEquality = + RuntimeEquality.create(RuntimeHelpers.create(), CelOptions.DEFAULT); + Double nan = Double.NaN; + + assertThat(runtimeEquality.objectEquals(nan, nan)).isFalse(); } } diff --git a/runtime/src/test/java/dev/cel/runtime/planner/AsyncCallRecordTest.java b/runtime/src/test/java/dev/cel/runtime/planner/AsyncCallRecordTest.java new file mode 100644 index 000000000..0dacb48ca --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/planner/AsyncCallRecordTest.java @@ -0,0 +1,615 @@ +// 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 com.google.common.util.concurrent.Futures.immediateFuture; +import static org.junit.Assert.assertThrows; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.primitives.UnsignedLong; +import com.google.common.util.concurrent.SettableFuture; +import com.google.testing.junit.testparameterinjector.TestParameter; +import com.google.testing.junit.testparameterinjector.TestParameterInjector; +import dev.cel.common.CelOptions; +import dev.cel.common.values.NullValue; +import dev.cel.runtime.CelAsyncFunctionOverload; +import dev.cel.runtime.RuntimeEquality; +import dev.cel.runtime.RuntimeHelpers; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; +import org.junit.runner.RunWith; + +@RunWith(TestParameterInjector.class) +public final class AsyncCallRecordTest { + + private static final CelAsyncFunctionOverload DUMMY_OVERLOAD = args -> immediateFuture("ok"); + + private static final RuntimeEquality RUNTIME_EQUALITY = + RuntimeEquality.create(RuntimeHelpers.create(), CelOptions.DEFAULT); + + /** Each constant differs from the base call in exactly one component. */ + @SuppressWarnings("ImmutableEnumChecker") + private enum CallMismatch { + EXPR_ID(20L, "myFunc", "myFunc_overload", ImmutableList.of(1L, "a")), + FUNCTION_NAME(10L, "otherFunc", "myFunc_overload", ImmutableList.of(1L, "a")), + OVERLOAD_ID(10L, "myFunc", "other_overload", ImmutableList.of(1L, "a")), + ARITY_FEWER(10L, "myFunc", "myFunc_overload", ImmutableList.of(1L)), + ARITY_MORE(10L, "myFunc", "myFunc_overload", ImmutableList.of(1L, "a", "extra")), + ARG_VALUE(10L, "myFunc", "myFunc_overload", ImmutableList.of(2L, "a")); + + private final long exprId; + private final String functionName; + private final String overloadId; + private final ImmutableList args; + + boolean matchesAgainst(AsyncCallRecord record) { + return record.matches(exprId, functionName, overloadId, args.toArray(), RUNTIME_EQUALITY); + } + + CallMismatch(long exprId, String functionName, String overloadId, ImmutableList args) { + this.exprId = exprId; + this.functionName = functionName; + this.overloadId = overloadId; + this.args = args; + } + } + + @Test + public void initialValues_matchConstructor() { + AsyncCallRecord record = + AsyncCallRecord.create( + 1L, 10L, "myFunc", "myFunc_overload", new Object[] {"arg1"}, DUMMY_OVERLOAD); + + assertThat(record.callId()).isEqualTo(1L); + assertThat(record.exprId()).isEqualTo(10L); + assertThat(record.functionName()).isEqualTo("myFunc"); + assertThat(record.overloadId()).isEqualTo("myFunc_overload"); + assertThat(record.args()).asList().containsExactly("arg1"); + assertThat(record.overload()).isEqualTo(DUMMY_OVERLOAD); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.NOT_STARTED); + assertThat(record.isCancelled()).isFalse(); + assertThat(record.result()).isEmpty(); + assertThat(record.error()).isEmpty(); + } + + @Test + public void create_nullArguments_throwsNullPointerException() { + assertThrows( + NullPointerException.class, + () -> AsyncCallRecord.create(1L, 10L, null, "overload", new Object[0], DUMMY_OVERLOAD)); + assertThrows( + NullPointerException.class, + () -> AsyncCallRecord.create(1L, 10L, "func", null, new Object[0], DUMMY_OVERLOAD)); + assertThrows( + NullPointerException.class, + () -> AsyncCallRecord.create(1L, 10L, "func", "overload", null, DUMMY_OVERLOAD)); + assertThrows( + NullPointerException.class, + () -> AsyncCallRecord.create(1L, 10L, "func", "overload", new Object[0], null)); + } + + @Test + public void hashCall_celEqualNumericArgs_shareBucket() { + // 1 == 1u == 1.0 in CEL, so all three must land in the same bucket. + int intHash = AsyncCallRecord.hashCall(10L, "ov", new Object[] {1L}); + + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {UnsignedLong.ONE})) + .isEqualTo(intHash); + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {1.0d})).isEqualTo(intHash); + } + + @Test + public void hashCall_signedZeroArgs_shareBucket() { + int zeroHash = AsyncCallRecord.hashCall(10L, "ov", new Object[] {0.0d}); + + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {-0.0d})).isEqualTo(zeroHash); + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {0L})).isEqualTo(zeroHash); + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {UnsignedLong.ZERO})) + .isEqualTo(zeroHash); + } + + @Test + public void hashCall_distinctNanRepresentations_shareBucket() { + int nanHash = AsyncCallRecord.hashCall(10L, "ov", new Object[] {Double.NaN}); + + assertThat( + AsyncCallRecord.hashCall( + 10L, "ov", new Object[] {Double.longBitsToDouble(0x7ff8000000000001L)})) + .isEqualTo(nanHash); + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {Float.NaN})).isEqualTo(nanHash); + } + + @Test + public void hashCall_distinctCalls_produceDistinctBuckets() { + // Distinctness is best-effort (hashCall is a bucketing hint), but verifies that call site + // components and argument boundaries are salted. + int base = AsyncCallRecord.hashCall(10L, "ov", new Object[] {"a", "bc"}); + + assertThat(AsyncCallRecord.hashCall(11L, "ov", new Object[] {"a", "bc"})).isNotEqualTo(base); + assertThat(AsyncCallRecord.hashCall(10L, "other", new Object[] {"a", "bc"})).isNotEqualTo(base); + // Arguments are mixed in separately, so ("a", "bc") does not collide with ("ab", "c"). + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {"ab", "c"})).isNotEqualTo(base); + // Type markers keep the string "NaN" distinct from the double NaN. + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {Double.NaN})) + .isNotEqualTo(AsyncCallRecord.hashCall(10L, "ov", new Object[] {"NaN"})); + } + + @Test + public void hashCall_complexArgs_shareSingleBucket() { + // Complex values are deliberately excluded from the hash because they need a richer + // equivalence than a value hash can express. They share one bucket and are separated by + // matches() instead. + int listHash = AsyncCallRecord.hashCall(10L, "ov", new Object[] {ImmutableList.of(1L)}); + + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {ImmutableList.of(2L)})) + .isEqualTo(listHash); + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {ImmutableMap.of("k", "v")})) + .isEqualTo(listHash); + assertThat(AsyncCallRecord.hashCall(10L, "ov", new Object[] {NullValue.NULL_VALUE})) + .isEqualTo(listHash); + } + + @Test + public void matches_identicalCall_returnsTrue() { + AsyncCallRecord record = recordWithArgs(1L, "a"); + + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", 1L, "a")).isTrue(); + } + + @Test + public void matches_mismatchedComponent_returnsFalse(@TestParameter CallMismatch mismatch) { + AsyncCallRecord record = recordWithArgs(1L, "a"); + + assertThat(mismatch.matchesAgainst(record)).isFalse(); + } + + @Test + public void matches_crossTypeNumericArgs_returnsTrue() { + AsyncCallRecord record = recordWithArgs(1L); + + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", 1.0d)).isTrue(); + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", UnsignedLong.ONE)).isTrue(); + } + + @Test + public void matches_signedZeroArgs_returnsTrue() { + AsyncCallRecord record = recordWithArgs(0.0d); + + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", -0.0d)).isTrue(); + } + + @Test + public void matches_nanArgs_returnsTrue() { + // CEL defines NaN != NaN. Without the override a node re-evaluated with a NaN argument would + // never find its record and would dispatch a fresh call on every pass. + AsyncCallRecord record = recordWithArgs(Double.NaN); + + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", Double.NaN)).isTrue(); + } + + @Test + public void matches_floatNanArgs_returnsTrue() { + AsyncCallRecord record = recordWithArgs(Float.NaN); + + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", Float.NaN)).isTrue(); + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", Double.NaN)).isTrue(); + } + + @Test + public void matches_nanVsNonNan_returnsFalse() { + AsyncCallRecord record = recordWithArgs(Double.NaN); + + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", 0.0d)).isFalse(); + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", 1.0d)).isFalse(); + } + + @Test + public void matches_celEqualCollectionArgs_returnsTrue() { + AsyncCallRecord record = recordWithArgs(ImmutableList.of(1L, 2L), ImmutableMap.of(1L, "v")); + + assertThat( + matches( + record, + 10L, + "myFunc", + "myFunc_overload", + ImmutableList.of(1.0d, 2.0d), + ImmutableMap.of(UnsignedLong.ONE, "v"))) + .isTrue(); + } + + @Test + public void matches_differingCollectionArgs_returnsFalse() { + AsyncCallRecord record = recordWithArgs(ImmutableList.of(1L, 2L), ImmutableMap.of(1L, "v")); + + assertThat( + matches( + record, + 10L, + "myFunc", + "myFunc_overload", + ImmutableList.of(1L, 3L), + ImmutableMap.of(1L, "v"))) + .isFalse(); + } + + @Test + public void matches_differentArgTypes_returnsFalse() { + AsyncCallRecord record = recordWithArgs("notANumber"); + + assertThat(matches(record, 10L, "myFunc", "myFunc_overload", 1L)).isFalse(); + } + + @Test + public void markRunning_successFromNotStarted() { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.NOT_STARTED); + + boolean marked = record.markRunning(); + + assertThat(marked).isTrue(); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.RUNNING); + } + + @Test + public void markRunning_whenAlreadyTerminalOrRunning_returnsFalse( + @TestParameter({"RUNNING", "SUCCESS", "FAILURE", "CANCELLED"}) AsyncCallRecord.State state) { + AsyncCallRecord record = createRecord(state); + + boolean marked = record.markRunning(); + + assertThat(marked).isFalse(); + assertThat(record.state()).isEqualTo(state); + } + + @Test + public void concurrentMarkRunning_exactlyOneSucceeds() throws Exception { + int numThreads = 4; + for (int i = 0; i < 100; i++) { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.NOT_STARTED); + CountDownLatch startLatch = new CountDownLatch(1); + AtomicInteger successCount = new AtomicInteger(); + Thread[] threads = new Thread[numThreads]; + + for (int t = 0; t < numThreads; t++) { + threads[t] = + new Thread( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + if (record.markRunning()) { + successCount.incrementAndGet(); + } + }); + threads[t].start(); + } + + startLatch.countDown(); + for (Thread thread : threads) { + thread.join(); + } + + assertThat(successCount.get()).isEqualTo(1); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.RUNNING); + } + } + + @Test + public void concurrentMarkRunningAndCancelInFlight_alwaysCancelsSuccessfully() throws Exception { + for (int i = 0; i < 100; i++) { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.NOT_STARTED); + CountDownLatch startLatch = new CountDownLatch(1); + AtomicBoolean markedRunning = new AtomicBoolean(); + AtomicBoolean cancelled = new AtomicBoolean(); + + Thread t1 = + new Thread( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + markedRunning.set(record.markRunning()); + }); + Thread t2 = + new Thread( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + cancelled.set(record.cancelInFlight()); + }); + + t1.start(); + t2.start(); + startLatch.countDown(); + t1.join(); + t2.join(); + + assertThat(cancelled.get()).isTrue(); + assertThat(record.isCancelled()).isTrue(); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.CANCELLED); + } + } + + @Test + public void complete_successFromActiveState( + @TestParameter({"NOT_STARTED", "RUNNING"}) AsyncCallRecord.State state) { + AsyncCallRecord record = createRecord(state); + + boolean completed = record.complete("successResult"); + + assertThat(completed).isTrue(); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.SUCCESS); + assertThat(record.result()).hasValue("successResult"); + assertThat(record.error()).isEmpty(); + } + + @Test + public void complete_whenAlreadyCompleted_returnsFalseAndDoesNotOverwrite() { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.RUNNING); + record.complete("firstResult"); + + boolean secondCompleted = record.complete("secondResult"); + + assertThat(secondCompleted).isFalse(); + assertThat(record.result()).hasValue("firstResult"); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.SUCCESS); + } + + @Test + public void fail_successFromActiveState( + @TestParameter({"NOT_STARTED", "RUNNING"}) AsyncCallRecord.State state) { + AsyncCallRecord record = createRecord(state); + RuntimeException error = new RuntimeException("test error"); + + boolean failed = record.fail(error); + + assertThat(failed).isTrue(); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.FAILURE); + assertThat(record.error()).hasValue(error); + assertThat(record.result()).isEmpty(); + } + + @Test + public void fail_whenAlreadyCompleted_returnsFalseAndDoesNotOverwrite() { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.RUNNING); + record.complete("firstResult"); + + boolean failed = record.fail(new RuntimeException("subsequent failure")); + + assertThat(failed).isFalse(); + assertThat(record.result()).hasValue("firstResult"); + assertThat(record.error()).isEmpty(); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.SUCCESS); + } + + @Test + public void cancelInFlight_activeState_cancelsFutureAndTransitionsToCancelled( + @TestParameter({"NOT_STARTED", "RUNNING"}) AsyncCallRecord.State state) { + AsyncCallRecord record = createRecord(state); + SettableFuture future = SettableFuture.create(); + record.setInFlightFuture(future); + + boolean cancelled = record.cancelInFlight(); + + assertThat(cancelled).isTrue(); + assertThat(record.isCancelled()).isTrue(); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.CANCELLED); + assertThat(future.isCancelled()).isTrue(); + } + + @Test + public void cancelInFlight_completedOrFailed_returnsFalseAndPreservesState( + @TestParameter({"SUCCESS", "FAILURE"}) AsyncCallRecord.State state) { + AsyncCallRecord record = createRecord(state); + SettableFuture future = SettableFuture.create(); + record.setInFlightFuture(future); + + boolean cancelled = record.cancelInFlight(); + + assertThat(cancelled).isFalse(); + assertThat(record.isCancelled()).isFalse(); + assertThat(record.state()).isEqualTo(state); + assertThat(future.isCancelled()).isFalse(); + } + + @Test + public void cancelInFlight_alreadyCancelled_returnsFalse() { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.CANCELLED); + + boolean cancelled = record.cancelInFlight(); + + assertThat(cancelled).isFalse(); + assertThat(record.isCancelled()).isTrue(); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.CANCELLED); + } + + @Test + public void setInFlightFuture_afterCancelled_cancelsImmediately() { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.CANCELLED); + SettableFuture future = SettableFuture.create(); + + record.setInFlightFuture(future); + + assertThat(future.isCancelled()).isTrue(); + } + + @Test + public void markCompletionReported_onlyFirstCallerSucceeds() { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.RUNNING); + + boolean firstReport = record.markCompletionReported(); + boolean secondReport = record.markCompletionReported(); + + assertThat(firstReport).isTrue(); + assertThat(secondReport).isFalse(); + } + + @Test + public void markCompletionReported_afterCancellation_stillSucceedsOnce() { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.RUNNING); + record.cancelInFlight(); + + boolean firstReport = record.markCompletionReported(); + boolean secondReport = record.markCompletionReported(); + + assertThat(firstReport).isTrue(); + assertThat(secondReport).isFalse(); + } + + @Test + public void concurrentCancellationAndSetInFlightFuture_futureIsAlwaysCancelled() + throws Exception { + for (int i = 0; i < 100; i++) { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.RUNNING); + SettableFuture future = SettableFuture.create(); + CountDownLatch startLatch = new CountDownLatch(1); + AtomicBoolean cancelled = new AtomicBoolean(); + + Thread cancelThread = + new Thread( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + cancelled.set(record.cancelInFlight()); + }); + Thread setFutureThread = + new Thread( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + record.setInFlightFuture(future); + }); + + cancelThread.start(); + setFutureThread.start(); + startLatch.countDown(); + cancelThread.join(); + setFutureThread.join(); + + assertThat(record.isCancelled()).isTrue(); + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.CANCELLED); + assertThat(future.isCancelled()).isTrue(); + } + } + + @Test + public void concurrentCompleteAndCancelInFlight_exactlyOneWinner() throws Exception { + for (int i = 0; i < 100; i++) { + AsyncCallRecord record = createRecord(AsyncCallRecord.State.RUNNING); + CountDownLatch startLatch = new CountDownLatch(1); + AtomicBoolean completed = new AtomicBoolean(); + AtomicBoolean cancelled = new AtomicBoolean(); + + Thread completeThread = + new Thread( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + completed.set(record.complete("success")); + }); + Thread cancelThread = + new Thread( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + cancelled.set(record.cancelInFlight()); + }); + + completeThread.start(); + cancelThread.start(); + startLatch.countDown(); + completeThread.join(); + cancelThread.join(); + + assertThat(completed.get() ^ cancelled.get()).isTrue(); + if (completed.get()) { + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.SUCCESS); + assertThat(record.result()).hasValue("success"); + } else { + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.CANCELLED); + assertThat(record.result()).isEmpty(); + } + } + } + + @Test + public void args_returnsDefensiveCopy() { + AsyncCallRecord record = + AsyncCallRecord.create( + 1L, 10L, "myFunc", "myFunc_overload", new Object[] {"original"}, DUMMY_OVERLOAD); + + Object[] returnedArgs = record.args(); + returnedArgs[0] = "mutated"; + + assertThat(record.args()).asList().containsExactly("original"); + } + + private static AsyncCallRecord recordWithArgs(Object... args) { + return AsyncCallRecord.create(1L, 10L, "myFunc", "myFunc_overload", args, DUMMY_OVERLOAD); + } + + private static boolean matches( + AsyncCallRecord record, long exprId, String functionName, String overloadId, Object... args) { + return record.matches(exprId, functionName, overloadId, args, RUNTIME_EQUALITY); + } + + private static AsyncCallRecord createRecord(AsyncCallRecord.State state) { + AsyncCallRecord record = + AsyncCallRecord.create(1L, 10L, "myFunc", "myFunc_overload", new Object[0], DUMMY_OVERLOAD); + switch (state) { + case NOT_STARTED: + break; + case RUNNING: + record.markRunning(); + break; + case SUCCESS: + record.complete("success"); + break; + case FAILURE: + record.fail(new RuntimeException("failed")); + break; + case CANCELLED: + record.cancelInFlight(); + break; + } + return record; + } +} diff --git a/runtime/src/test/java/dev/cel/runtime/planner/AsyncCallStateTrackerTest.java b/runtime/src/test/java/dev/cel/runtime/planner/AsyncCallStateTrackerTest.java new file mode 100644 index 000000000..f222ba3e2 --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/planner/AsyncCallStateTrackerTest.java @@ -0,0 +1,828 @@ +// 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 com.google.common.collect.ImmutableList.toImmutableList; +import static com.google.common.collect.Iterables.getOnlyElement; +import static com.google.common.truth.Truth.assertThat; +import static com.google.common.util.concurrent.Futures.immediateFailedFuture; +import static com.google.common.util.concurrent.Futures.immediateFuture; +import static com.google.common.util.concurrent.MoreExecutors.directExecutor; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertThrows; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; +import com.google.common.primitives.UnsignedLong; +import com.google.common.util.concurrent.ForwardingListenableFuture.SimpleForwardingListenableFuture; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.SettableFuture; +import com.google.errorprone.annotations.Immutable; +import javax.annotation.concurrent.ThreadSafe; +import com.google.testing.junit.testparameterinjector.TestParameter; +import com.google.testing.junit.testparameterinjector.TestParameterInjector; +import dev.cel.common.CelOptions; +import dev.cel.common.exceptions.CelDivideByZeroException; +import dev.cel.common.exceptions.CelRuntimeException; +import dev.cel.common.values.CelValueConverter; +import dev.cel.common.values.NullValue; +import dev.cel.runtime.AccumulatedUnknowns; +import dev.cel.runtime.CelAsyncCall; +import dev.cel.runtime.CelAsyncEvaluationOptions; +import dev.cel.runtime.CelAsyncFunctionOverload; +import dev.cel.runtime.CelAsyncObserver; +import dev.cel.runtime.RuntimeEquality; +import dev.cel.runtime.RuntimeHelpers; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CopyOnWriteArrayList; +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.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.jspecify.annotations.Nullable; +import org.junit.Test; +import org.junit.runner.RunWith; + +@RunWith(TestParameterInjector.class) +@SuppressWarnings("Immutable") +public final class AsyncCallStateTrackerTest { + + private final RuntimeEquality runtimeEquality = + RuntimeEquality.create(RuntimeHelpers.create(), CelOptions.DEFAULT); + private final Executor directExecutor = directExecutor(); + private final AsyncCallStateTracker tracker = AsyncCallStateTracker.create(runtimeEquality); + private final AsyncGate gate = AsyncGate.create(1); + private final AsyncCompletionCoordinator coordinator = newCoordinator(gate, directExecutor); + private final RecordingObserver observer = new RecordingObserver(); + + @Test + public void dispatchPendingCalls_onlyLaunchesRequiredCallIds() throws Exception { + AtomicBoolean call1Executed = new AtomicBoolean(false); + AtomicBoolean call2Executed = new AtomicBoolean(false); + AccumulatedUnknowns unk1 = + recordCall( + 1L, + "func1", + "a", + args -> { + call1Executed.set(true); + return immediateFuture("res1"); + }); + recordCall( + 2L, + "func2", + "b", + args -> { + call2Executed.set(true); + return immediateFuture("res2"); + }); + + tracker.dispatchPendingCalls( + unk1.callIds(), directExecutor, gate, coordinator, /* observer= */ null); + + assertThat(call1Executed.get()).isTrue(); + assertThat(call2Executed.get()).isFalse(); + } + + @Test + public void dispatchPendingCalls_cancelledWhileQueued_abortsOverloadAndNotifiesObserver( + @TestParameter boolean withObserver) throws Exception { + List queuedTasks = new ArrayList<>(); + AtomicBoolean overloadExecuted = new AtomicBoolean(false); + AccumulatedUnknowns unknowns = + recordDefaultCall( + args -> { + overloadExecuted.set(true); + return immediateFuture("ok"); + }); + tracker.dispatchPendingCalls( + unknowns.callIds(), queuedTasks::add, gate, coordinator, withObserver ? observer : null); + + tracker.cancelInFlight(); + queuedTasks.get(0).run(); + + assertThat(overloadExecuted.get()).isFalse(); + assertThat(gate.activeCount()).isEqualTo(0); + assertThat(tracker.hasInFlightCalls()).isFalse(); + if (withObserver) { + assertThat(observer.startedCalls()).hasSize(1); + assertThat(getOnlyElement(observer.finishedCalls()).error) + .isInstanceOf(CancellationException.class); + } + } + + @Test + public void dispatchPendingCalls_whenGateFull_defersUntilPermitReleased( + @TestParameter boolean releaseAndRetry) throws Exception { + checkState(gate.tryAcquire(), "Failed to acquire permit"); + List queuedTasks = new ArrayList<>(); + AtomicBoolean overloadCalled = new AtomicBoolean(false); + AccumulatedUnknowns unknowns = + recordDefaultCall( + args -> { + overloadCalled.set(true); + return immediateFuture("ok"); + }); + + tracker.dispatchPendingCalls( + unknowns.callIds(), queuedTasks::add, gate, coordinator, /* observer= */ null); + if (releaseAndRetry) { + gate.release(); + tracker.dispatchPendingCalls( + unknowns.callIds(), queuedTasks::add, gate, coordinator, /* observer= */ null); + queuedTasks.get(0).run(); + } + + assertThat(overloadCalled.get()).isEqualTo(releaseAndRetry); + assertThat(gate.activeCount()).isEqualTo(releaseAndRetry ? 0 : 1); + assertThat(tracker.hasInFlightCalls()).isFalse(); + } + + @Test + public void recordOrGet_existingKey_reusesCallIdAndAllocatesNewIdForDistinctKey() + throws Exception { + AccumulatedUnknowns first = recordCall(10L, "fn", "x", args -> SettableFuture.create()); + AccumulatedUnknowns second = recordCall(10L, "fn", "x", args -> SettableFuture.create()); + AccumulatedUnknowns third = recordCall(20L, "fn", "y", args -> SettableFuture.create()); + + assertThat(first.callIds()).containsExactly(1L); + assertThat(second.callIds()).containsExactly(1L); + assertThat(third.callIds()).containsExactly(2L); + } + + @Test + public void recordOrGet_bucketHashCollision_disambiguatesViaMatches() throws Exception { + // Both arguments are complex types (lists) so hashArg yields COMPLEX_HASH_MARKER for both, + // causing a bucket collision under the same (exprId, overloadId). + AccumulatedUnknowns first = + recordCall(10L, "fn", ImmutableList.of("a"), args -> SettableFuture.create()); + AccumulatedUnknowns second = + recordCall(10L, "fn", ImmutableList.of("b"), args -> SettableFuture.create()); + AccumulatedUnknowns firstAgain = + recordCall(10L, "fn", ImmutableList.of("a"), args -> SettableFuture.create()); + AccumulatedUnknowns secondAgain = + recordCall(10L, "fn", ImmutableList.of("b"), args -> SettableFuture.create()); + + assertThat(first.callIds()).containsExactly(1L); + assertThat(second.callIds()).containsExactly(2L); + assertThat(firstAgain.callIds()).containsExactly(1L); + assertThat(secondAgain.callIds()).containsExactly(2L); + } + + @Test + public void recordOrGet_celEqualArguments_reusesCallId() throws Exception { + AccumulatedUnknowns first = recordCall(10L, "fn", 1L, args -> SettableFuture.create()); + AccumulatedUnknowns second = recordCall(10L, "fn", 1.0d, args -> SettableFuture.create()); + + assertThat(first.callIds()).containsExactly(1L); + assertThat(second.callIds()).containsExactly(1L); + } + + @Test + public void recordOrGet_nanArguments_reusesCallId() throws Exception { + AccumulatedUnknowns first = recordCall(10L, "fn", Double.NaN, args -> SettableFuture.create()); + AccumulatedUnknowns second = recordCall(10L, "fn", Float.NaN, args -> SettableFuture.create()); + + assertThat(first.callIds()).containsExactly(1L); + assertThat(second.callIds()).containsExactly(1L); + } + + @Test + public void recordOrGet_signedZeroArguments_reusesCallId() throws Exception { + AccumulatedUnknowns first = recordCall(10L, "fn", 0.0d, args -> SettableFuture.create()); + AccumulatedUnknowns second = recordCall(10L, "fn", -0.0d, args -> SettableFuture.create()); + + assertThat(first.callIds()).containsExactly(1L); + assertThat(second.callIds()).containsExactly(1L); + } + + @Test + public void recordOrGet_unsignedLongAndLongArguments_reusesCallIdWhenEqual() throws Exception { + AccumulatedUnknowns first = + recordCall(10L, "fn", UnsignedLong.valueOf(42L), args -> SettableFuture.create()); + AccumulatedUnknowns second = recordCall(10L, "fn", 42L, args -> SettableFuture.create()); + + assertThat(first.callIds()).containsExactly(1L); + assertThat(second.callIds()).containsExactly(1L); + } + + @Test + public void recordOrGet_optimisticRead_doesNotBlockOnBucketLock() throws Exception { + CelAsyncFunctionOverload overload = args -> SettableFuture.create(); + // Seeds a second entry in the same bucket so the optimistic read iterates past one element. + recordCall(1L, "fn", ImmutableList.of("a"), overload); + AccumulatedUnknowns second = recordCall(1L, "fn", ImmutableList.of("b"), overload); + int bucketKey = + AsyncCallRecord.hashCall(1L, "fn_overload", new Object[] {ImmutableList.of("b")}); + CopyOnWriteArrayList bucket = tracker.recordsByBucket().get(bucketKey); + checkNotNull(bucket, "Bucket must not be null"); + checkState(bucket.size() >= 2, "Bucket must contain at least two colliding calls"); + + CountDownLatch lockAcquired = new CountDownLatch(1); + CountDownLatch releaseLock = new CountDownLatch(1); + Thread blockerThread = + new Thread( + () -> { + synchronized (bucket) { + lockAcquired.countDown(); + try { + releaseLock.await(5, SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }); + SettableFuture readResult = SettableFuture.create(); + Thread readerThread = + new Thread( + () -> { + try { + // Read the second colliding record to ensure iteration covers multiple elements + // lock-free + readResult.set(recordCall(1L, "fn", ImmutableList.of("b"), overload)); + } catch (Throwable t) { + readResult.setException(t); + } + }); + try { + blockerThread.start(); + checkState(lockAcquired.await(5, SECONDS), "blockerThread failed to acquire bucket lock"); + + readerThread.start(); + AccumulatedUnknowns result = readResult.get(1, SECONDS); + + assertThat(result.callIds()).containsExactlyElementsIn(second.callIds()); + } finally { + releaseLock.countDown(); + blockerThread.join(5000); + readerThread.join(5000); + } + } + + @Test + public void recordOrGet_concurrentRegistrationRace_reusesExistingRecordInSlowPath() + throws Exception { + CelAsyncFunctionOverload overload = args -> SettableFuture.create(); + Object[] args = new Object[] {1L}; + int bucketKey = AsyncCallRecord.hashCall(1L, "fn_overload", args); + CopyOnWriteArrayList bucket = new CopyOnWriteArrayList<>(); + tracker.recordsByBucket().put(bucketKey, bucket); + + CountDownLatch blockerLocked = new CountDownLatch(1); + CountDownLatch populateAndRelease = new CountDownLatch(1); + Thread blockerThread = + new Thread( + () -> { + synchronized (bucket) { + blockerLocked.countDown(); + try { + populateAndRelease.await(5, SECONDS); + AsyncCallRecord preExisting = + AsyncCallRecord.create(99L, 1L, "fn", "fn_overload", args, overload); + tracker.recordsById().put(99L, preExisting); + bucket.add(preExisting); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }); + SettableFuture callerResult = SettableFuture.create(); + Thread callerThread = + new Thread( + () -> { + try { + callerResult.set( + (AccumulatedUnknowns) + tracker.recordOrGet( + 1L, + "fn", + "fn_overload", + args, + overload, + CelValueConverter.getDefaultInstance())); + } catch (Throwable t) { + callerResult.setException(t); + } + }); + try { + blockerThread.start(); + checkState(blockerLocked.await(5, SECONDS), "blockerThread failed to acquire lock"); + + callerThread.start(); + long deadline = System.currentTimeMillis() + 5000; + while (callerThread.getState() != Thread.State.BLOCKED) { + if (callerThread.getState() == Thread.State.TERMINATED) { + callerResult.get(); + throw new AssertionError("callerThread terminated unexpectedly without blocking"); + } + if (System.currentTimeMillis() > deadline) { + throw new AssertionError( + "callerThread never entered BLOCKED state; state is " + callerThread.getState()); + } + Thread.sleep(10); + } + populateAndRelease.countDown(); + AccumulatedUnknowns unknowns = callerResult.get(5, SECONDS); + + assertThat(unknowns.callIds()).containsExactly(99L); + } finally { + populateAndRelease.countDown(); + blockerThread.join(5000); + callerThread.join(5000); + } + } + + @Test + public void recordOrGet_concurrentBucketHashCollision_registersAllCallsSafely() throws Exception { + List results = new CopyOnWriteArrayList<>(); + + runConcurrently( + 16, + () -> { + int id = results.size(); + results.add( + recordCall( + 10L, "fn", ImmutableList.of("arg_" + id), args -> SettableFuture.create())); + }); + + assertThat(results).hasSize(16); + assertThat(tracker.hasInFlightCalls()).isFalse(); + } + + private enum OverloadFailureMode { + THROWS_SYNCHRONOUSLY, + FAILED_FUTURE, + RETURNS_NULL_FUTURE, + FUTURE_COMPLETES_WITH_NULL + } + + @Test + public void dispatchPendingCalls_overloadFails_notifiesObserverAndReleasesPermit( + @TestParameter OverloadFailureMode failureMode, @TestParameter boolean withObserver) + throws Exception { + RuntimeException expectedError = new RuntimeException("fail"); + AccumulatedUnknowns unknowns = + recordDefaultCall( + args -> { + switch (failureMode) { + case THROWS_SYNCHRONOUSLY: + throw expectedError; + case FAILED_FUTURE: + return immediateFailedFuture(expectedError); + case RETURNS_NULL_FUTURE: + return null; + case FUTURE_COMPLETES_WITH_NULL: + return immediateFuture(null); + } + throw new AssertionError(); + }); + + tracker.dispatchPendingCalls( + unknowns.callIds(), directExecutor, gate, coordinator, withObserver ? observer : null); + + boolean isNullFailure = + failureMode == OverloadFailureMode.RETURNS_NULL_FUTURE + || failureMode == OverloadFailureMode.FUTURE_COMPLETES_WITH_NULL; + assertThat(gate.activeCount()).isEqualTo(0); + assertThat(tracker.hasInFlightCalls()).isFalse(); + if (withObserver) { + assertThat(getOnlyElement(observer.startedArgs())).containsExactly("x"); + Throwable recordedError = getOnlyElement(observer.finishedCalls()).error; + if (isNullFailure) { + assertThat(recordedError).isInstanceOf(NullPointerException.class); + } else { + assertThat(recordedError).isSameInstanceAs(expectedError); + } + } + IllegalArgumentException evalException = + assertThrows( + IllegalArgumentException.class, + () -> getDefaultCall(args -> immediateFuture("unused"))); + if (isNullFailure) { + assertThat(evalException).hasCauseThat().isInstanceOf(NullPointerException.class); + } else { + assertThat(evalException).hasCauseThat().isSameInstanceAs(expectedError); + } + } + + @Test + public void dispatchPendingCalls_executorRejection_releasesPermitAndFailsRecord() + throws Exception { + AccumulatedUnknowns unknowns = recordDefaultCall(args -> immediateFuture("done")); + Executor rejectingExecutor = + cmd -> { + throw new RejectedExecutionException("pool full"); + }; + + tracker.dispatchPendingCalls( + unknowns.callIds(), rejectingExecutor, gate, coordinator, /* observer= */ null); + + assertThat(gate.activeCount()).isEqualTo(0); + assertThat(tracker.hasInFlightCalls()).isFalse(); + } + + @Test + public void recordOrGet_afterSuccess_returnsResolvedValueAndNotifiesObserver( + @TestParameter boolean nullSentinelValues) throws Exception { + Object arg = nullSentinelValues ? NullValue.NULL_VALUE : "x"; + Object expectedResult = nullSentinelValues ? NullValue.NULL_VALUE : "syncSuccess"; + SettableFuture future = SettableFuture.create(); + AccumulatedUnknowns unknowns = recordCall(10L, "fn", arg, args -> future); + tracker.dispatchPendingCalls(unknowns.callIds(), directExecutor, gate, coordinator, observer); + future.set(expectedResult); + + Object result = recordOrGetCall(10L, "fn", arg, args -> future); + + FinishedCall finished = getOnlyElement(observer.finishedCalls()); + assertThat(result).isEqualTo(expectedResult); + assertThat(getOnlyElement(observer.startedCalls()).call.functionName()).isEqualTo("fn"); + assertThat(getOnlyElement(observer.startedArgs())).containsExactly(arg); + assertThat(finished.result).isEqualTo(expectedResult); + assertThat(finished.call.functionName()).isEqualTo("fn"); + assertThat(gate.activeCount()).isEqualTo(0); + assertThat(tracker.hasInFlightCalls()).isFalse(); + } + + @Test + public void dispatchPendingCalls_concurrentRace_threadContentionHandledSafely() throws Exception { + AtomicInteger callsDispatched = new AtomicInteger(0); + AccumulatedUnknowns unknowns = + recordDefaultCall( + args -> { + callsDispatched.incrementAndGet(); + return immediateFuture("result"); + }); + + runConcurrently(8, () -> dispatchDefaultCalls(unknowns)); + + assertThat(callsDispatched.get()).isEqualTo(1); + assertThat(gate.activeCount()).isEqualTo(0); + assertThat(tracker.hasInFlightCalls()).isFalse(); + } + + @Test + public void cancelInFlight_beforeFutureCompletes_releasesPermitAndNotifiesObserver( + @TestParameter boolean succeedsAfterCancel) throws Exception { + SettableFuture underlyingFuture = SettableFuture.create(); + dispatchCallWithObserver(args -> nonCancellableFuture(underlyingFuture)); + RuntimeException lateError = new RuntimeException("late_failure"); + + tracker.cancelInFlight(); + if (succeedsAfterCancel) { + underlyingFuture.set("late_success"); + } else { + underlyingFuture.setException(lateError); + } + + assertThat(observer.startedCalls()).hasSize(1); + FinishedCall finished = getOnlyElement(observer.finishedCalls()); + assertThat(finished.result).isEqualTo(succeedsAfterCancel ? "late_success" : null); + assertThat(finished.error).isEqualTo(succeedsAfterCancel ? null : lateError); + assertThat(finished.call.functionName()).isEqualTo("fn"); + assertThat(gate.activeCount()).isEqualTo(0); + assertThat(tracker.hasInFlightCalls()).isFalse(); + } + + @Test + public void recordOrGet_whenInFlight_returnsAccumulatedUnknownsWithSameCallId() throws Exception { + SettableFuture pendingFuture = SettableFuture.create(); + AccumulatedUnknowns initial = recordDefaultCall(args -> pendingFuture); + dispatchDefaultCalls(initial); + + AccumulatedUnknowns whileRunning = recordDefaultCall(args -> pendingFuture); + + assertThat(whileRunning.callIds()).containsExactlyElementsIn(initial.callIds()); + assertThat(tracker.hasInFlightCalls()).isTrue(); + } + + @Test + public void recordOrGet_concurrentRegistrationSameKey_deduplicatesToSingleCallId() + throws Exception { + List results = new CopyOnWriteArrayList<>(); + + runConcurrently(16, () -> results.add(recordDefaultCall(args -> immediateFuture("done")))); + + assertThat(results).hasSize(16); + long canonicalCallId = getOnlyElement(results.get(0).callIds()); + for (AccumulatedUnknowns result : results) { + assertThat(result.callIds()).containsExactly(canonicalCallId); + } + } + + @Test + public void tryLaunch_whenRecordCannotTransitionToRunning_releasesPermitWithoutDispatch( + @TestParameter boolean alreadyRunning) { + AtomicInteger tasksExecuted = new AtomicInteger(0); + AsyncCallRecord record = + defaultRecord( + args -> { + tasksExecuted.incrementAndGet(); + return immediateFuture("done"); + }); + if (alreadyRunning) { + checkState(record.markRunning()); + } else { + record.cancelInFlight(); + } + + tracker.tryLaunch(record, directExecutor, gate, coordinator, observer); + + assertThat(gate.activeCount()).isEqualTo(0); + assertThat(tasksExecuted.get()).isEqualTo(0); + assertThat(observer.startedCalls()).isEmpty(); + } + + @Test + public void tryLaunch_whenFutureNotifiesListenersTwice_releasesGatePermitOnce() { + AsyncGate twoPermitGate = AsyncGate.create(2); + checkState(twoPermitGate.tryAcquire()); + AsyncCallRecord record = defaultRecord(args -> doubleNotifyingFuture()); + + tracker.tryLaunch( + record, + directExecutor, + twoPermitGate, + newCoordinator(twoPermitGate, directExecutor), + observer); + + assertThat(observer.finishedCalls()).hasSize(1); + assertThat(twoPermitGate.activeCount()).isEqualTo(1); + } + + @Test + public void tryLaunch_observerThrowsOnStart_failsRecordAndReleasesPermit() { + RuntimeException expected = new RuntimeException("observer start failure"); + AtomicReference reportedError = new AtomicReference<>(); + CelAsyncObserver throwingObserver = + new CelAsyncObserver() { + @Override + public void onCallStarted(CelAsyncCall call, ImmutableList args) { + throw expected; + } + + @Override + public void onCallFinished( + CelAsyncCall call, @Nullable Object result, @Nullable Throwable error) { + reportedError.set(error); + } + }; + AsyncCallRecord record = defaultRecord(args -> immediateFuture("done")); + + tracker.tryLaunch(record, directExecutor, gate, coordinator, throwingObserver); + + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.FAILURE); + assertThat(record.error()).hasValue(expected); + assertThat(reportedError.get()).isSameInstanceAs(expected); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void tryLaunch_observerThrowsOnFinish_preservesResultAndReleasesPermit() { + CelAsyncObserver throwingObserver = + new CelAsyncObserver() { + @Override + public void onCallStarted(CelAsyncCall call, ImmutableList args) {} + + @Override + public void onCallFinished( + CelAsyncCall call, @Nullable Object result, @Nullable Throwable error) { + throw new RuntimeException("observer finish failure"); + } + }; + AsyncCallRecord record = defaultRecord(args -> immediateFuture("done")); + + tracker.tryLaunch(record, directExecutor, gate, coordinator, throwingObserver); + + assertThat(record.state()).isEqualTo(AsyncCallRecord.State.SUCCESS); + assertThat(record.result()).hasValue("done"); + assertThat(gate.activeCount()).isEqualTo(0); + } + + private enum FailureCase { + CHECKED_OR_RUNTIME(new IllegalArgumentException("computation failed"), "computation failed"), + NULL_MESSAGE(new IllegalStateException((String) null), "IllegalStateException"), + CEL_RUNTIME_EXCEPTION(new CelDivideByZeroException(), "/ by zero"), + CANCELLED(new CancellationException(), "was cancelled"); + + private final Throwable cause; + private final String expectedMessage; + + FailureCase(Throwable cause, String expectedMessage) { + this.cause = cause; + this.expectedMessage = expectedMessage; + } + } + + @Test + public void recordOrGet_whenRecordFailed_throwsExpectedException( + @TestParameter FailureCase failureCase) throws Exception { + ListenableFuture future = immediateFailedFuture(failureCase.cause); + if (failureCase == FailureCase.CANCELLED) { + recordDefaultCall(args -> future); + tracker.cancelInFlight(); + CancellationException e = + assertThrows(CancellationException.class, () -> recordDefaultCall(args -> future)); + assertThat(e).hasMessageThat().contains(failureCase.expectedMessage); + } else if (failureCase == FailureCase.CEL_RUNTIME_EXCEPTION) { + dispatchDefaultCalls(recordDefaultCall(args -> future)); + CelRuntimeException e = + assertThrows(CelRuntimeException.class, () -> recordDefaultCall(args -> future)); + assertThat(e).isSameInstanceAs(failureCase.cause); + } else { + dispatchDefaultCalls(recordDefaultCall(args -> future)); + IllegalArgumentException e = + assertThrows(IllegalArgumentException.class, () -> recordDefaultCall(args -> future)); + assertThat(e).hasMessageThat().contains(failureCase.expectedMessage); + assertThat(e).hasCauseThat().isInstanceOf(failureCase.cause.getClass()); + } + } + + @Test + public void recordOrGet_nullCelValueConverter_throwsNullPointerException() { + assertThrows( + NullPointerException.class, + () -> + tracker.recordOrGet( + 10L, + "fn", + "fn_overload", + new Object[] {"x"}, + args -> immediateFuture("done"), + null)); + } + + @Test + public void dispatchPendingCalls_withUnknownCallId_doesNotThrow() { + tracker.dispatchPendingCalls( + ImmutableSet.of(9999L), directExecutor, gate, coordinator, /* observer= */ null); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + private void dispatchCallWithObserver(CelAsyncFunctionOverload overload) { + tracker.dispatchPendingCalls( + recordDefaultCall(overload).callIds(), directExecutor, gate, coordinator, observer); + } + + private void dispatchDefaultCalls(AccumulatedUnknowns unknowns) { + tracker.dispatchPendingCalls( + unknowns.callIds(), directExecutor, gate, coordinator, /* observer= */ null); + } + + private AccumulatedUnknowns recordDefaultCall(CelAsyncFunctionOverload overload) { + return (AccumulatedUnknowns) getDefaultCall(overload); + } + + private Object getDefaultCall(CelAsyncFunctionOverload overload) { + return recordOrGetCall(10L, "fn", "x", overload); + } + + private AccumulatedUnknowns recordCall( + long exprId, String functionName, Object arg, CelAsyncFunctionOverload overload) { + return (AccumulatedUnknowns) recordOrGetCall(exprId, functionName, arg, overload); + } + + private Object recordOrGetCall( + long exprId, String functionName, Object arg, CelAsyncFunctionOverload overload) { + return tracker.recordOrGet( + exprId, + functionName, + functionName + "_overload", + new Object[] {arg}, + overload, + CelValueConverter.getDefaultInstance()); + } + + private static AsyncCallRecord defaultRecord(CelAsyncFunctionOverload overload) { + return AsyncCallRecord.create(100L, 10L, "fn", "fn_overload", new Object[] {"x"}, overload); + } + + private static AsyncCompletionCoordinator newCoordinator(AsyncGate gate, Executor executor) { + return AsyncCompletionCoordinator.create( + CelAsyncEvaluationOptions.defaultOptions(), gate, executor, t -> {}); + } + + private interface ThrowingRunnable { + void run() throws Exception; + } + + private static void runConcurrently(int threads, ThrowingRunnable action) + throws InterruptedException { + ExecutorService pool = Executors.newFixedThreadPool(threads); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch done = new CountDownLatch(threads); + for (int i = 0; i < threads; i++) { + pool.execute( + () -> { + try { + start.await(); + action.run(); + } catch (Exception e) { + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } + throw new AssertionError(e); + } finally { + done.countDown(); + } + }); + } + start.countDown(); + assertThat(done.await(5, SECONDS)).isTrue(); + pool.shutdown(); + } + + private static ListenableFuture nonCancellableFuture(ListenableFuture delegate) { + return new SimpleForwardingListenableFuture(delegate) { + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + return false; + } + }; + } + + private static ListenableFuture doubleNotifyingFuture() { + return new SimpleForwardingListenableFuture(immediateFuture("done")) { + @Override + public void addListener(Runnable listener, Executor executor) { + super.addListener(listener, executor); + super.addListener(listener, executor); + } + }; + } + + @ThreadSafe + private static final class RecordingObserver implements CelAsyncObserver { + private final CopyOnWriteArrayList startedCalls = new CopyOnWriteArrayList<>(); + private final CopyOnWriteArrayList finishedCalls = new CopyOnWriteArrayList<>(); + + @Override + public void onCallStarted(CelAsyncCall call, ImmutableList args) { + startedCalls.add(new StartedCall(call, args)); + } + + @Override + public void onCallFinished( + CelAsyncCall call, @Nullable Object result, @Nullable Throwable error) { + finishedCalls.add(new FinishedCall(call, result, error)); + } + + ImmutableList startedCalls() { + return ImmutableList.copyOf(startedCalls); + } + + ImmutableList> startedArgs() { + return startedCalls.stream().map(s -> s.args).collect(toImmutableList()); + } + + ImmutableList finishedCalls() { + return ImmutableList.copyOf(finishedCalls); + } + } + + @Immutable + @SuppressWarnings("Immutable") + private static final class StartedCall { + private final CelAsyncCall call; + private final ImmutableList args; + + private StartedCall(CelAsyncCall call, ImmutableList args) { + this.call = call; + this.args = args; + } + } + + @Immutable + @SuppressWarnings("Immutable") + private static final class FinishedCall { + private final CelAsyncCall call; + private final @Nullable Object result; + private final @Nullable Throwable error; + + private FinishedCall(CelAsyncCall call, @Nullable Object result, @Nullable Throwable error) { + this.call = call; + this.result = result; + this.error = error; + } + } +} 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 38d1d0d70..53240ff87 100644 --- a/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel +++ b/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel @@ -16,13 +16,15 @@ java_library( "//:java_truth", "//common:cel_ast", "//common:cel_descriptor_util", + "//common:cel_function_decl", + "//common:cel_overload_decl", "//common:cel_source", - "//common:compiler_common", "//common:container", "//common:error_codes", "//common:options", "//common/ast", "//common/exceptions:divide_by_zero", + "//common/exceptions:runtime_exception", "//common/internal:cel_descriptor_pools", "//common/internal:default_message_factory", "//common/internal:dynamic_proto", @@ -40,9 +42,7 @@ java_library( "//extensions", "//parser:macro", "//runtime", - "//runtime:async_call", - "//runtime:async_drain_strategy", - "//runtime:async_options", + "//runtime:accumulated_unknowns", "//runtime:descriptor_type_resolver", "//runtime:dispatcher", "//runtime:function_binding", @@ -52,14 +52,19 @@ java_library( "//runtime:runtime_helpers", "//runtime:standard_functions", "//runtime:unknown_attributes", + "//runtime/planner:async_call_state_tracker", "//runtime/planner:async_completion_coordinator", "//runtime/planner:async_gate", + "//runtime/planner:planned_program", "//runtime/planner:program_planner", "//runtime/standard:type", "@cel_spec//proto/cel/expr/conformance/proto3:test_all_types_java_proto", + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:com_google_guava_guava", "@maven//:com_google_testparameterinjector_test_parameter_injector", "@maven//:junit_junit", + "@maven//:org_jspecify_jspecify", ], ) diff --git a/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerTest.java b/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerTest.java index a3b1e3596..ebf8e1cdb 100644 --- a/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerTest.java +++ b/runtime/src/test/java/dev/cel/runtime/planner/ProgramPlannerTest.java @@ -14,7 +14,9 @@ package dev.cel.runtime.planner; +import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.truth.Truth.assertThat; +import static com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService; import static dev.cel.common.CelFunctionDecl.newFunctionDeclaration; import static dev.cel.common.CelOverloadDecl.newGlobalOverload; import static dev.cel.common.CelOverloadDecl.newMemberOverload; @@ -26,6 +28,8 @@ import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; import com.google.common.primitives.UnsignedLong; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListeningExecutorService; import com.google.testing.junit.testparameterinjector.TestParameter; import com.google.testing.junit.testparameterinjector.TestParameterInjector; import com.google.testing.junit.testparameterinjector.TestParameters; @@ -66,6 +70,7 @@ import dev.cel.expr.conformance.proto3.TestAllTypes.NestedMessage; import dev.cel.extensions.CelExtensions; import dev.cel.parser.CelStandardMacro; +import dev.cel.runtime.CelAsyncEvaluationOptions; import dev.cel.runtime.CelAttribute; import dev.cel.runtime.CelAttributePattern; import dev.cel.runtime.CelEvaluationException; @@ -92,7 +97,9 @@ public final class ProgramPlannerTest { private static final CelTypeProvider TYPE_PROVIDER = new CombinedCelTypeProvider( DefaultTypeProvider.getInstance(), - new ProtoMessageTypeProvider(ImmutableSet.of(TestAllTypes.getDescriptor()))); + ProtoMessageTypeProvider.newBuilder() + .addDescriptors(ImmutableSet.of(TestAllTypes.getDescriptor())) + .build()); private static final RuntimeEquality RUNTIME_EQUALITY = RuntimeEquality.create(RuntimeHelpers.create(), CEL_OPTIONS); private static final CelDescriptorPool DESCRIPTOR_POOL = @@ -119,7 +126,9 @@ public final class ProgramPlannerTest { CEL_VALUE_CONVERTER, CEL_CONTAINER, CEL_OPTIONS, - ImmutableSet.of("late_bound_func")); + ImmutableSet.of("late_bound_func"), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); private static final CelCompiler CEL_COMPILER = CelCompilerFactory.standardCelCompilerBuilder() @@ -255,9 +264,7 @@ private static DefaultDispatcher newDispatcher() { private static void addBindingsToDispatcher( DefaultDispatcher.Builder builder, ImmutableCollection overloadBindings) { - if (overloadBindings.isEmpty()) { - throw new IllegalArgumentException("Invalid bindings"); - } + checkArgument(!overloadBindings.isEmpty(), "Invalid bindings"); overloadBindings.forEach( overload -> @@ -320,7 +327,9 @@ public void plan_ident_enumContainer() throws Exception { CEL_VALUE_CONVERTER, container, CEL_OPTIONS, - ImmutableSet.of()); + ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); Program program = planner.plan(ast); @@ -519,7 +528,7 @@ public void plan_call_throws() throws Exception { .hasMessageThat() .contains("evaluation error at :5: Function 'error' failed with arg(s) ''"); assertThat(e).hasCauseThat().isInstanceOf(IllegalArgumentException.class); - assertThat(e.getCause()).hasMessageThat().contains("Intentional error"); + assertThat(e).hasCauseThat().hasMessageThat().contains("Intentional error"); } @Test @@ -1023,7 +1032,9 @@ public void plan_comprehension_iterationLimit_throws(String expression) throws E CEL_VALUE_CONVERTER, CEL_CONTAINER, options, - ImmutableSet.of()); + ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); CelAbstractSyntaxTree ast = compile(expression); Program program = planner.plan(ast); @@ -1044,7 +1055,9 @@ public void plan_comprehension_iterationLimit_success() throws Exception { CEL_VALUE_CONVERTER, CEL_CONTAINER, options, - /* lateBoundFunctionNames= */ ImmutableSet.of()); + /* lateBoundFunctionNames= */ ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); CelAbstractSyntaxTree ast = compile("[1, 2, 3].map(x, [1, 2].map(y, x + y))"); Program program = planner.plan(ast); @@ -1202,6 +1215,90 @@ public void plan_foldMap_withUnknownLoopCondition_earlyReturn() throws Exception CelUnknownSet.create(ImmutableSet.of(CelAttribute.create("unk")), ImmutableSet.of(7L))); } + @Test + public void newPlanner_withAsyncOptionsAndExecutor_plansSuccessfully() throws Exception { + ListeningExecutorService executor = newDirectExecutorService(); + try { + CelAsyncEvaluationOptions asyncOptions = + CelAsyncEvaluationOptions.builder().setMaxIterations(5).build(); + ProgramPlanner planner = + ProgramPlanner.newPlanner( + TYPE_PROVIDER, + VALUE_PROVIDER, + newDispatcher(), + CEL_VALUE_CONVERTER, + CEL_CONTAINER, + CEL_OPTIONS, + ImmutableSet.of(), + asyncOptions, + executor); + CelAbstractSyntaxTree ast = compile("1 + 2"); + + PlannedProgram program = planner.plan(ast); + + assertThat(program.eval()).isEqualTo(3L); + assertThat(program.asyncOptions()).isEqualTo(asyncOptions); + assertThat(program.asyncExecutor()).hasValue(executor); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void newPlanner_nullAsyncOptions_throwsNullPointerException() { + DefaultDispatcher dispatcher = newDispatcher(); + + assertThrows( + NullPointerException.class, + () -> + ProgramPlanner.newPlanner( + TYPE_PROVIDER, + VALUE_PROVIDER, + dispatcher, + CEL_VALUE_CONVERTER, + CEL_CONTAINER, + CEL_OPTIONS, + ImmutableSet.of(), + /* asyncOptions= */ null, + /* asyncExecutor= */ null)); + } + + @Test + public void plan_asyncFunction_evalSynchronously_throwsCelEvaluationException() throws Exception { + CelCompiler compiler = + CelCompilerFactory.standardCelCompilerBuilder() + .addFunctionDeclarations( + newFunctionDeclaration( + "asyncSquare", + newGlobalOverload("asyncSquare_int", SimpleType.INT, SimpleType.INT))) + .build(); + CelAbstractSyntaxTree ast = compiler.compile("asyncSquare(5)").getAst(); + DefaultDispatcher.Builder dispatcher = DefaultDispatcher.newBuilder(); + addBindingsToDispatcher( + dispatcher, + ImmutableList.of( + CelFunctionBinding.fromAsync( + "asyncSquare_int", Long.class, (Long arg) -> Futures.immediateFuture(arg * arg)))); + ProgramPlanner planner = + ProgramPlanner.newPlanner( + TYPE_PROVIDER, + VALUE_PROVIDER, + dispatcher.build(), + CEL_VALUE_CONVERTER, + CEL_CONTAINER, + CEL_OPTIONS, + ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); + Program program = planner.plan(ast); + + CelEvaluationException e = assertThrows(CelEvaluationException.class, program::eval); + + assertThat(e) + .hasMessageThat() + .contains("Async function 'asyncSquare' evaluated in synchronous mode."); + } + @Test public void plan_binaryFunction_withUnknownArg() throws Exception { CelCompiler compiler = @@ -1264,7 +1361,9 @@ public void localShadowIdentifier_inSelect() throws Exception { CEL_VALUE_CONVERTER, CelContainer.ofName("cel.example"), CEL_OPTIONS, - /* lateBoundFunctionNames= */ ImmutableSet.of()); + /* lateBoundFunctionNames= */ ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); CelAbstractSyntaxTree ast = compile(celCompiler, "[{'z': 0}].exists(y, y.z == 0)"); Program program = planner.plan(ast); @@ -1289,7 +1388,9 @@ public void localShadowIdentifier_inSelect_globalDisambiguation() throws Excepti CEL_VALUE_CONVERTER, CelContainer.ofName("y"), CEL_OPTIONS, - /* lateBoundFunctionNames= */ ImmutableSet.of()); + /* lateBoundFunctionNames= */ ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); CelAbstractSyntaxTree ast = compile(celCompiler, "[{'z': 0}].exists(y, y.z == 0 && .y.z == 1)"); Program program = planner.plan(ast); @@ -1313,7 +1414,9 @@ public void localShadowIdentifier_withGlobalDisambiguation() throws Exception { CEL_VALUE_CONVERTER, CelContainer.newBuilder().build(), CEL_OPTIONS, - /* lateBoundFunctionNames= */ ImmutableSet.of()); + /* lateBoundFunctionNames= */ ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); CelAbstractSyntaxTree ast = compile(celCompiler, "[0].exists(x, x == 0 && .x == 1)"); Program program = planner.plan(ast); @@ -1337,7 +1440,9 @@ public void localDoubleShadowIdentifier_withGlobalDisambiguation() throws Except CEL_VALUE_CONVERTER, CelContainer.newBuilder().build(), CEL_OPTIONS, - /* lateBoundFunctionNames= */ ImmutableSet.of()); + /* lateBoundFunctionNames= */ ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); CelAbstractSyntaxTree ast = compile(celCompiler, "[0].exists(x, [x+1].exists(x, x == .x))"); Program program = planner.plan(ast); @@ -1377,7 +1482,9 @@ public void plan_customFunctionReturningUnknown_fieldSelection() throws Exceptio CEL_VALUE_CONVERTER, CEL_CONTAINER, CEL_OPTIONS, - ImmutableSet.of()); + ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); Program program = planner.plan(ast); @@ -1417,7 +1524,9 @@ public void plan_customFunctionReturningUnknown_binaryOperation() throws Excepti CEL_VALUE_CONVERTER, CEL_CONTAINER, CEL_OPTIONS, - ImmutableSet.of()); + ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); Program program = planner.plan(ast); @@ -1450,7 +1559,9 @@ public void plan_variableAsCelUnknownSet_propagatesUnknown() throws Exception { CEL_VALUE_CONVERTER, CEL_CONTAINER, CEL_OPTIONS, - ImmutableSet.of()); + ImmutableSet.of(), + CelAsyncEvaluationOptions.defaultOptions(), + /* asyncExecutor= */ null); ImmutableMap vars = ImmutableMap.of(