diff --git a/.bazelrc b/.bazelrc index f6e2f39c0..34a59ec39 100644 --- a/.bazelrc +++ b/.bazelrc @@ -16,7 +16,7 @@ build --java_language_version=11 common --javacopt=-Xlint:-options # Remove flag once https://github.com/google/cel-spec/issues/508 and rules_jvm_external is fixed. -common --incompatible_autoload_externally=proto_library,cc_proto_library,java_proto_library,java_test +common --incompatible_autoload_externally=proto_library,cc_proto_library,java_proto_library,java_test,java_import # Limit repository cache size by not caching extracted repository contents build --repo_contents_cache= diff --git a/publish/BUILD.bazel b/publish/BUILD.bazel index 7fd15a769..2fb948cea 100644 --- a/publish/BUILD.bazel +++ b/publish/BUILD.bazel @@ -29,6 +29,10 @@ COMMON_TARGETS = [ # keep sorted RUNTIME_TARGETS = [ "//runtime/src/main/java/dev/cel/runtime", + "//runtime/src/main/java/dev/cel/runtime:async_call", + "//runtime/src/main/java/dev/cel/runtime:async_drain_strategy", + "//runtime/src/main/java/dev/cel/runtime:async_observer", + "//runtime/src/main/java/dev/cel/runtime:async_options", "//runtime/src/main/java/dev/cel/runtime:base", "//runtime/src/main/java/dev/cel/runtime:interpreter", "//runtime/src/main/java/dev/cel/runtime:late_function_binding", diff --git a/runtime/BUILD.bazel b/runtime/BUILD.bazel index c87fadca9..e1acc4261 100644 --- a/runtime/BUILD.bazel +++ b/runtime/BUILD.bazel @@ -9,6 +9,10 @@ package( java_library( name = "runtime", exports = [ + ":async_call", + ":async_drain_strategy", + ":async_observer", + ":async_options", ":descriptor_message_provider", ":evaluation_exception", ":function_overload", @@ -340,6 +344,11 @@ java_library( exports = ["//runtime/src/main/java/dev/cel/runtime:function_overload"], ) +cel_android_library( + name = "function_overload_android", + exports = ["//runtime/src/main/java/dev/cel/runtime:function_overload_android"], +) + java_library( name = "descriptor_message_provider", visibility = ["//:internal"], @@ -379,3 +388,43 @@ cel_android_library( name = "partial_vars_android", exports = ["//runtime/src/main/java/dev/cel/runtime:partial_vars_android"], ) + +java_library( + name = "async_call", + exports = ["//runtime/src/main/java/dev/cel/runtime:async_call"], +) + +cel_android_library( + name = "async_call_android", + exports = ["//runtime/src/main/java/dev/cel/runtime:async_call_android"], +) + +java_library( + name = "async_drain_strategy", + exports = ["//runtime/src/main/java/dev/cel/runtime:async_drain_strategy"], +) + +cel_android_library( + name = "async_drain_strategy_android", + exports = ["//runtime/src/main/java/dev/cel/runtime:async_drain_strategy_android"], +) + +java_library( + name = "async_observer", + exports = ["//runtime/src/main/java/dev/cel/runtime:async_observer"], +) + +cel_android_library( + name = "async_observer_android", + exports = ["//runtime/src/main/java/dev/cel/runtime:async_observer_android"], +) + +java_library( + name = "async_options", + exports = ["//runtime/src/main/java/dev/cel/runtime:async_options"], +) + +cel_android_library( + name = "async_options_android", + exports = ["//runtime/src/main/java/dev/cel/runtime:async_options_android"], +) diff --git a/runtime/planner/BUILD.bazel b/runtime/planner/BUILD.bazel index 860d413a0..e15fa2989 100644 --- a/runtime/planner/BUILD.bazel +++ b/runtime/planner/BUILD.bazel @@ -21,3 +21,27 @@ java_library( visibility = ["//:internal"], exports = ["//runtime/src/main/java/dev/cel/runtime/planner:planned_program"], ) + +java_library( + name = "async_gate", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate"], +) + +cel_android_library( + name = "async_gate_android", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate_android"], +) + +java_library( + name = "async_completion_coordinator", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator"], +) + +cel_android_library( + name = "async_completion_coordinator_android", + visibility = ["//:internal"], + exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator_android"], +) diff --git a/runtime/src/main/java/dev/cel/runtime/BUILD.bazel b/runtime/src/main/java/dev/cel/runtime/BUILD.bazel index 489bb64d8..43b490cd3 100644 --- a/runtime/src/main/java/dev/cel/runtime/BUILD.bazel +++ b/runtime/src/main/java/dev/cel/runtime/BUILD.bazel @@ -784,6 +784,7 @@ cel_android_library( java_library( name = "function_overload", srcs = [ + "CelAsyncFunctionOverload.java", "CelFunctionOverload.java", "OptimizedFunctionOverload.java", ], @@ -800,9 +801,12 @@ java_library( cel_android_library( name = "function_overload_android", srcs = [ + "CelAsyncFunctionOverload.java", "CelFunctionOverload.java", "OptimizedFunctionOverload.java", ], + tags = [ + ], deps = [ ":evaluation_exception", ":unknown_attributes_android", @@ -817,6 +821,7 @@ java_library( tags = [ ], deps = [ + ":async_options", ":descriptor_type_resolver", ":dispatcher", ":evaluation_exception", @@ -850,7 +855,6 @@ java_library( "//runtime:activation", "//runtime:interpretable", "//runtime:proto_message_activation_factory", - "//runtime:resolved_overload", "//runtime/planner:planned_program", "//runtime/planner:program_planner", "//runtime/standard:type", @@ -867,6 +871,7 @@ java_library( tags = [ ], deps = [ + ":async_options", ":descriptor_message_provider", ":descriptor_type_resolver", ":dispatcher", @@ -922,6 +927,7 @@ java_library( ], deps = [ ":activation", + ":async_options", ":evaluation_exception", ":evaluation_listener", ":function_binding", @@ -946,6 +952,7 @@ java_library( "@maven//:com_google_errorprone_error_prone_annotations", "@maven//:com_google_guava_guava", "@maven//:com_google_protobuf_protobuf_java", + "@maven//:org_jspecify_jspecify", ], ) @@ -1277,6 +1284,118 @@ cel_android_library( ], ) +java_library( + name = "async_call", + srcs = ["CelAsyncCall.java"], + tags = [ + ], + deps = [ + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", + ], +) + +cel_android_library( + name = "async_call_android", + srcs = ["CelAsyncCall.java"], + tags = [ + ], + deps = [ + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", + ], +) + +java_library( + name = "async_drain_strategy", + srcs = [ + "CelAsyncDrainAction.java", + "CelAsyncDrainStrategy.java", + ], + tags = [ + ], + deps = [ + ":async_call", + "//:auto_value", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", + ], +) + +cel_android_library( + name = "async_drain_strategy_android", + srcs = [ + "CelAsyncDrainAction.java", + "CelAsyncDrainStrategy.java", + ], + tags = [ + ], + deps = [ + ":async_call_android", + "//:auto_value", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven_android//:com_google_guava_guava", + ], +) + +java_library( + name = "async_observer", + srcs = ["CelAsyncObserver.java"], + tags = [ + ], + deps = [ + ":async_call", + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", + "@maven//:org_jspecify_jspecify", + ], +) + +cel_android_library( + name = "async_observer_android", + srcs = ["CelAsyncObserver.java"], + tags = [ + ], + deps = [ + ":async_call_android", + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:org_jspecify_jspecify", + "@maven_android//:com_google_guava_guava", + ], +) + +java_library( + name = "async_options", + srcs = ["CelAsyncEvaluationOptions.java"], + tags = [ + ], + deps = [ + ":async_drain_strategy", + ":async_observer", + "//:auto_value", + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:org_jspecify_jspecify", + ], +) + +cel_android_library( + name = "async_options_android", + srcs = ["CelAsyncEvaluationOptions.java"], + tags = [ + ], + deps = [ + ":async_drain_strategy_android", + ":async_observer_android", + "//:auto_value", + "@maven//:com_google_code_findbugs_annotations", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:org_jspecify_jspecify", + ], +) + java_library( name = "program", srcs = ["Program.java"], @@ -1288,6 +1407,7 @@ java_library( ":partial_vars", ":variable_resolver", "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", ], ) @@ -1302,6 +1422,7 @@ cel_android_library( ":partial_vars_android", ":variable_resolver", "@maven//:com_google_errorprone_error_prone_annotations", + "@maven_android//:com_google_guava_guava", ], ) diff --git a/runtime/src/main/java/dev/cel/runtime/CelAsyncCall.java b/runtime/src/main/java/dev/cel/runtime/CelAsyncCall.java new file mode 100644 index 000000000..1582f12a1 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/CelAsyncCall.java @@ -0,0 +1,34 @@ +// 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; + +import javax.annotation.concurrent.ThreadSafe; + +/** Describes a pending or completed asynchronous function call. */ +@ThreadSafe +public interface CelAsyncCall { + + /** Returns the unique incremental tracking ID assigned to this call. */ + long callId(); + + /** Returns the AST expression node ID where the call is located. */ + long exprId(); + + /** Returns the name of the function being invoked. */ + String functionName(); + + /** Returns the specific overload ID being invoked. */ + String overloadId(); +} diff --git a/runtime/src/main/java/dev/cel/runtime/CelAsyncDrainAction.java b/runtime/src/main/java/dev/cel/runtime/CelAsyncDrainAction.java new file mode 100644 index 000000000..845db25a3 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/CelAsyncDrainAction.java @@ -0,0 +1,57 @@ +// 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; + +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkNotNull; + +import com.google.auto.value.AutoValue; +import com.google.errorprone.annotations.Immutable; +import java.time.Duration; + +/** Dictates what asynchronous evaluation should do after inspecting completions. */ +@AutoValue +@Immutable +public abstract class CelAsyncDrainAction { + + CelAsyncDrainAction() {} + + /** Indicates that the AST should be re-evaluated immediately. */ + public abstract boolean shouldReevaluate(); + + /** + * Indicates how long the evaluator should wait for additional completions before deciding to + * re-evaluate. A duration of ZERO with reevaluate=false means wait indefinitely for the next + * completion. + */ + public abstract Duration waitDuration(); + + public static CelAsyncDrainAction waitDuration(Duration duration) { + checkNotNull(duration); + checkArgument(!duration.isNegative(), "duration must not be negative"); + if (duration.isZero()) { + return reevaluate(); + } + return new AutoValue_CelAsyncDrainAction(false, duration); + } + + public static CelAsyncDrainAction reevaluate() { + return new AutoValue_CelAsyncDrainAction(true, Duration.ZERO); + } + + public static CelAsyncDrainAction waitForMore() { + return new AutoValue_CelAsyncDrainAction(false, Duration.ZERO); + } +} diff --git a/runtime/src/main/java/dev/cel/runtime/CelAsyncDrainStrategy.java b/runtime/src/main/java/dev/cel/runtime/CelAsyncDrainStrategy.java new file mode 100644 index 000000000..6d8349f8e --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/CelAsyncDrainStrategy.java @@ -0,0 +1,101 @@ +// 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; + +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkNotNull; + +import com.google.errorprone.annotations.Immutable; +import java.time.Duration; +import java.util.List; + +/** + * Controls when asynchronous evaluation re-evaluates the AST after async completions. + * + *

The evaluator consults the strategy each time completions are received. + */ +@Immutable +public interface CelAsyncDrainStrategy { + + /** + * Evaluates the current state of asynchronous evaluation and determines the next step. + * + * @param completedBatch The batch of async call completions accumulated so far in this drain + * cycle. + * @param activeCallsCount The number of async calls currently launched but unresolved. + */ + CelAsyncDrainAction nextAction(List completedBatch, int activeCallsCount); + + /** + * Re-evaluates after a debounce window after the first completion, batching completions that + * complete at roughly the same time. + */ + static CelAsyncDrainStrategy drainReady(Duration debounce) { + return new DrainReadyStrategy(debounce); + } + + /** Re-evaluates with the default debounce window of 100 microseconds. */ + static CelAsyncDrainStrategy drainReady() { + return drainReady(Duration.ofNanos(100_000)); + } + + /** Re-evaluates immediately as soon as any single call completes. */ + static CelAsyncDrainStrategy drainNone() { + return (completed, active) -> { + checkNotNull(completed, "completedBatch must not be null"); + checkArgument(active >= 0, "activeCallsCount must be non-negative: %s", active); + return active == 0 || !completed.isEmpty() + ? CelAsyncDrainAction.reevaluate() + : CelAsyncDrainAction.waitForMore(); + }; + } + + /** Waits for all currently pending calls to finish before re-evaluating. */ + static CelAsyncDrainStrategy drainAll() { + return (completed, active) -> { + checkNotNull(completed, "completedBatch must not be null"); + checkArgument(active >= 0, "activeCallsCount must be non-negative: %s", active); + return active == 0 ? CelAsyncDrainAction.reevaluate() : CelAsyncDrainAction.waitForMore(); + }; + } + + /** Internal implementation of the drain ready strategy with configurable debounce duration. */ + @Immutable + final class DrainReadyStrategy implements CelAsyncDrainStrategy { + private final Duration debounce; + + DrainReadyStrategy(Duration debounce) { + this.debounce = checkNotNull(debounce); + checkArgument(!debounce.isNegative(), "debounce duration must not be negative"); + } + + @Override + public CelAsyncDrainAction nextAction(List completedBatch, int activeCallsCount) { + checkNotNull(completedBatch, "completedBatch must not be null"); + checkArgument( + activeCallsCount >= 0, "activeCallsCount must be non-negative: %s", activeCallsCount); + if (activeCallsCount == 0) { + return CelAsyncDrainAction.reevaluate(); + } + if (completedBatch.isEmpty()) { + return CelAsyncDrainAction.waitForMore(); + } + if (debounce.isZero()) { + return CelAsyncDrainAction.reevaluate(); + } + return CelAsyncDrainAction.waitDuration(debounce); + } + } +} diff --git a/runtime/src/main/java/dev/cel/runtime/CelAsyncEvaluationOptions.java b/runtime/src/main/java/dev/cel/runtime/CelAsyncEvaluationOptions.java new file mode 100644 index 000000000..19609e923 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/CelAsyncEvaluationOptions.java @@ -0,0 +1,141 @@ +// 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; + +import com.google.auto.value.AutoValue; +import javax.annotation.concurrent.ThreadSafe; +import java.util.Optional; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicLong; + +/** Options for configuring asynchronous CEL evaluation. */ +@AutoValue +@ThreadSafe +public abstract class CelAsyncEvaluationOptions { + + private static final int DEFAULT_MAX_CONCURRENCY = 100; + private static final int DEFAULT_MAX_ITERATIONS = 1_000; + + /** + * Maximum number of concurrent async function calls in-flight simultaneously. A value <= 0 + * indicates unbounded concurrency. + */ + public abstract int maxConcurrency(); + + /** Strategy governing when to trigger re-evaluation after async call completions. */ + public abstract CelAsyncDrainStrategy drainStrategy(); + + /** Safety cap on the maximum number of AST re-evaluation passes before aborting. */ + public abstract int maxIterations(); + + /** + * Returns the custom configured {@link ScheduledExecutorService}, if present. + * + *

If absent, {@link #resolveScheduledExecutorService()} falls back to an internal, shared + * single-threaded daemon scheduler. + */ + public abstract Optional scheduledExecutorService(); + + /** Returns the configured lifecycle observer, if present. */ + public abstract Optional observer(); + + /** + * Resolves the {@link ScheduledExecutorService} used for debounce timers, falling back to a + * shared, lazily initialized single-threaded daemon scheduler (named {@code + * cel-async-debounce-*}) if not custom-configured. + * + *

The scheduler is used exclusively as an alarm clock to trigger continuation wakeups; it does + * not execute CEL evaluation tasks. + */ + public ScheduledExecutorService resolveScheduledExecutorService() { + return scheduledExecutorService().orElse(DefaultDebounceSchedulerHolder.INSTANCE); + } + + public abstract Builder toBuilder(); + + /** + * Returns a new {@link Builder} initialized with standard default options: + * + *

+ */ + public static Builder newBuilder() { + return new AutoValue_CelAsyncEvaluationOptions.Builder() + .setMaxConcurrency(DEFAULT_MAX_CONCURRENCY) + .setDrainStrategy(CelAsyncDrainStrategy.drainReady()) + .setMaxIterations(DEFAULT_MAX_ITERATIONS); + } + + /** + * Returns a new {@link Builder} initialized with standard default options. + * + *

Equivalent to calling {@link #newBuilder()}. + */ + public static Builder builder() { + return newBuilder(); + } + + /** + * Returns a {@link CelAsyncEvaluationOptions} instance with the {@link #newBuilder() default + * configuration}. + */ + public static CelAsyncEvaluationOptions defaultOptions() { + return newBuilder().build(); + } + + private static final class DefaultDebounceSchedulerHolder { + private static final AtomicLong counter = new AtomicLong(); + private static final ScheduledExecutorService INSTANCE = + Executors.newSingleThreadScheduledExecutor( + r -> { + Thread t = new Thread(r); + t.setName("cel-async-debounce-" + counter.getAndIncrement()); + t.setDaemon(true); + return t; + }); + } + + /** Builder for {@link CelAsyncEvaluationOptions}. */ + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setMaxConcurrency(int maxConcurrency); + + public abstract Builder setDrainStrategy(CelAsyncDrainStrategy drainStrategy); + + public abstract Builder setMaxIterations(int maxIterations); + + /** + * Sets a custom {@link ScheduledExecutorService} for debounce timers. + * + *

If not set, defaults to an internal, shared single-threaded daemon scheduler. + */ + public abstract Builder setScheduledExecutorService( + ScheduledExecutorService scheduledExecutorService); + + public abstract Builder setObserver(CelAsyncObserver observer); + + public abstract CelAsyncEvaluationOptions build(); + } + + // Package-private constructor prevents extension outside package while allowing AutoValue. + CelAsyncEvaluationOptions() {} +} diff --git a/runtime/src/main/java/dev/cel/runtime/CelAsyncFunctionOverload.java b/runtime/src/main/java/dev/cel/runtime/CelAsyncFunctionOverload.java new file mode 100644 index 000000000..685c81573 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/CelAsyncFunctionOverload.java @@ -0,0 +1,57 @@ +// 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; + +import com.google.common.util.concurrent.ListenableFuture; +import com.google.errorprone.annotations.Immutable; + +/** Represents a CEL custom function overload that executes asynchronously. */ +@Immutable +public interface CelAsyncFunctionOverload extends CelFunctionOverload { + + /** Invokes the overload asynchronously with evaluated arguments. */ + ListenableFuture applyAsync(Object[] args) throws CelEvaluationException; + + /** Optimized overload for single-argument async functions to avoid array allocation. */ + default ListenableFuture applyAsync(Object arg) throws CelEvaluationException { + return applyAsync(new Object[] {arg}); + } + + /** Optimized overload for two-argument async functions to avoid array allocation. */ + default ListenableFuture applyAsync(Object arg1, Object arg2) + throws CelEvaluationException { + return applyAsync(new Object[] {arg1, arg2}); + } + + @Override + default Object apply(Object[] args) throws CelEvaluationException { + throw new UnsupportedOperationException( + "Async overload cannot be evaluated synchronously. Use evalAsync instead."); + } + + /** Helper interface for describing unary async functions. */ + @Immutable + @FunctionalInterface + interface Unary { + ListenableFuture apply(T arg) throws CelEvaluationException; + } + + /** Helper interface for describing binary async functions. */ + @Immutable + @FunctionalInterface + interface Binary { + ListenableFuture apply(T1 arg1, T2 arg2) throws CelEvaluationException; + } +} diff --git a/runtime/src/main/java/dev/cel/runtime/CelAsyncObserver.java b/runtime/src/main/java/dev/cel/runtime/CelAsyncObserver.java new file mode 100644 index 000000000..31b001115 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/CelAsyncObserver.java @@ -0,0 +1,46 @@ +// 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; + +import com.google.common.collect.ImmutableList; +import javax.annotation.concurrent.ThreadSafe; +import org.jspecify.annotations.Nullable; + +/** + * Provides callbacks for monitoring the lifecycle of asynchronous function calls. + * + *

Implementations must be thread-safe: {@code onCallStarted} is invoked from the thread + * dispatching the call, while {@code onCallFinished} is invoked from the call's completion thread. + */ +@ThreadSafe +public interface CelAsyncObserver { + + /** + * Invoked when an asynchronous function call is first dispatched. + * + * @param call The call description. + * @param args The evaluated arguments passed to the function call. + */ + void onCallStarted(CelAsyncCall call, ImmutableList args); + + /** + * Invoked when an asynchronous function call completes with either a result or an exception. + * + * @param call The call description. + * @param result The result of the call if successful, or null if failed. + * @param error The failure cause if the call failed, or null if successful. + */ + void onCallFinished(CelAsyncCall call, @Nullable Object result, @Nullable Throwable error); +} diff --git a/runtime/src/main/java/dev/cel/runtime/CelFunctionBinding.java b/runtime/src/main/java/dev/cel/runtime/CelFunctionBinding.java index 98991d383..3b0084394 100644 --- a/runtime/src/main/java/dev/cel/runtime/CelFunctionBinding.java +++ b/runtime/src/main/java/dev/cel/runtime/CelFunctionBinding.java @@ -15,10 +15,12 @@ package dev.cel.runtime; import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkNotNull; import com.google.common.base.Strings; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; +import com.google.common.util.concurrent.ListenableFuture; import com.google.errorprone.annotations.Immutable; import java.util.Collection; @@ -100,6 +102,74 @@ static CelFunctionBinding from( overloadId, ImmutableList.copyOf(argTypes), impl, /* isStrict= */ true); } + /** + * Create an asynchronous unary function binding from the {@code overloadId}, {@code arg}, and + * {@code impl}. + */ + @SuppressWarnings("unchecked") // Safe from CelFunctionOverload.canHandle check before invocation + static CelFunctionBinding fromAsync( + String overloadId, Class arg, CelAsyncFunctionOverload.Unary impl) { + checkNotNull(overloadId); + checkNotNull(arg); + checkNotNull(impl); + return from( + overloadId, + ImmutableList.of(arg), + new CelAsyncFunctionOverload() { + @Override + public ListenableFuture applyAsync(Object[] args) throws CelEvaluationException { + return impl.apply((T) args[0]); + } + + @Override + public ListenableFuture applyAsync(Object arg1) throws CelEvaluationException { + return impl.apply((T) arg1); + } + }); + } + + /** + * Create an asynchronous binary function binding from the {@code overloadId}, {@code arg1}, + * {@code arg2}, and {@code impl}. + */ + @SuppressWarnings("unchecked") // Safe from CelFunctionOverload.canHandle check before invocation + static CelFunctionBinding fromAsync( + String overloadId, + Class arg1, + Class arg2, + CelAsyncFunctionOverload.Binary impl) { + checkNotNull(overloadId); + checkNotNull(arg1); + checkNotNull(arg2); + checkNotNull(impl); + return from( + overloadId, + ImmutableList.of(arg1, arg2), + new CelAsyncFunctionOverload() { + @Override + public ListenableFuture applyAsync(Object[] args) throws CelEvaluationException { + return impl.apply((T1) args[0], (T2) args[1]); + } + + @Override + public ListenableFuture applyAsync(Object a1, Object a2) + throws CelEvaluationException { + return impl.apply((T1) a1, (T2) a2); + } + }); + } + + /** + * Create an asynchronous function binding from the {@code overloadId}, {@code argTypes}, and + * {@code impl}. + */ + static CelFunctionBinding fromAsync( + String overloadId, Iterable> argTypes, CelAsyncFunctionOverload impl) { + checkNotNull(overloadId); + checkNotNull(argTypes); + checkNotNull(impl); + return from(overloadId, argTypes, impl); + } /** See {@link #fromOverloads(String, Collection)}. */ static ImmutableSet fromOverloads( @@ -110,11 +180,22 @@ static ImmutableSet fromOverloads( /** * Creates a set of bindings for a function, enabling dynamic dispatch logic to select the correct * overload at runtime based on argument types. + * + *

Note: Overloaded functions with {@link CelAsyncFunctionOverload} are not currently + * supported. */ static ImmutableSet fromOverloads( String functionName, Collection overloadBindings) { checkArgument(!Strings.isNullOrEmpty(functionName), "Function name cannot be null or empty"); checkArgument(!overloadBindings.isEmpty(), "You must provide at least one binding."); + // TODO: Dynamic dispatch grouping does not currently support asynchronous + // function overloads. In parsed-only mode, overloaded async functions must be resolved + // at runtime via CelFunctionResolver. + for (CelFunctionBinding binding : overloadBindings) { + checkArgument( + !(binding.getDefinition() instanceof CelAsyncFunctionOverload), + "Asynchronous function overloads cannot be grouped using fromOverloads."); + } return FunctionBindingImpl.groupOverloadsToFunction( functionName, ImmutableSet.copyOf(overloadBindings)); diff --git a/runtime/src/main/java/dev/cel/runtime/CelFunctionResolver.java b/runtime/src/main/java/dev/cel/runtime/CelFunctionResolver.java index 2fb136a1a..836c48d8b 100644 --- a/runtime/src/main/java/dev/cel/runtime/CelFunctionResolver.java +++ b/runtime/src/main/java/dev/cel/runtime/CelFunctionResolver.java @@ -25,6 +25,22 @@ @ThreadSafe public interface CelFunctionResolver { + /** An empty function resolver that resolves no overloads. */ + CelFunctionResolver EMPTY = + 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(); + } + }; + /** * Finds a specific function overload to invoke based on given parameters. * @@ -33,7 +49,7 @@ public interface CelFunctionResolver { * from this list with matching arguments. * @param args The arguments to pass to the function. * @return an optional value of the resolved overload. - * @throws CelEvaluationException if the overload resolution is ambiguous, + * @throws CelEvaluationException if the overload resolution is ambiguous. */ Optional findOverloadMatchingArgs( String functionName, Collection overloadIds, Object[] args) diff --git a/runtime/src/main/java/dev/cel/runtime/CelRuntime.java b/runtime/src/main/java/dev/cel/runtime/CelRuntime.java index 1e7fdcac8..e9c6ca20a 100644 --- a/runtime/src/main/java/dev/cel/runtime/CelRuntime.java +++ b/runtime/src/main/java/dev/cel/runtime/CelRuntime.java @@ -14,6 +14,7 @@ package dev.cel.runtime; +import com.google.common.util.concurrent.ListenableFuture; import com.google.errorprone.annotations.CanIgnoreReturnValue; import com.google.errorprone.annotations.Immutable; import javax.annotation.concurrent.ThreadSafe; @@ -42,6 +43,12 @@ interface Program extends dev.cel.runtime.Program { /** Evaluate the expression using {@code message} fields as the source of input variables. */ Object eval(Message message) throws CelEvaluationException; + /** + * Evaluate the expression asynchronously using {@code message} fields as the source of input + * variables. + */ + ListenableFuture evalAsync(Message message); + /** * Trace evaluates a compiled program without any variables and invokes the listener as * evaluation progresses through the AST. diff --git a/runtime/src/main/java/dev/cel/runtime/CelRuntimeBuilder.java b/runtime/src/main/java/dev/cel/runtime/CelRuntimeBuilder.java index 00f6e3bf7..feacf5e37 100644 --- a/runtime/src/main/java/dev/cel/runtime/CelRuntimeBuilder.java +++ b/runtime/src/main/java/dev/cel/runtime/CelRuntimeBuilder.java @@ -14,6 +14,7 @@ package dev.cel.runtime; +import com.google.common.util.concurrent.ListeningExecutorService; import com.google.errorprone.annotations.CanIgnoreReturnValue; import com.google.errorprone.annotations.CheckReturnValue; import com.google.protobuf.DescriptorProtos.FileDescriptorSet; @@ -214,6 +215,22 @@ public interface CelRuntimeBuilder { @CanIgnoreReturnValue CelRuntimeBuilder setContainer(CelContainer container); + /** + * Sets options to use for asynchronous evaluation. + * + *

If not configured, defaults to {@link CelAsyncEvaluationOptions#defaultOptions()}. + */ + @CanIgnoreReturnValue + CelRuntimeBuilder setAsyncEvaluationOptions(CelAsyncEvaluationOptions asyncEvaluationOptions); + + /** + * Sets the executor to use for asynchronous evaluation. + * + *

This executor is required when evaluating expressions asynchronously via {@link + * Program#evalAsync}. + */ + @CanIgnoreReturnValue + CelRuntimeBuilder setAsyncExecutor(ListeningExecutorService asyncExecutor); /** Build a new instance of the {@code CelRuntime}. */ @CheckReturnValue diff --git a/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java b/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java index 5cda25800..11e7b7ee2 100644 --- a/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java +++ b/runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java @@ -20,6 +20,8 @@ import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.ListeningExecutorService; import com.google.errorprone.annotations.CanIgnoreReturnValue; import com.google.errorprone.annotations.Immutable; import com.google.protobuf.DescriptorProtos; @@ -95,27 +97,22 @@ public abstract class CelRuntimeImpl implements CelRuntime { @AutoValue.CopyAnnotations abstract @Nullable ExtensionRegistry extensionRegistry(); + // CelAsyncEvaluationOptions is an immutable value object configuring asynchronous evaluation. + @SuppressWarnings("Immutable") + @AutoValue.CopyAnnotations + abstract CelAsyncEvaluationOptions asyncEvaluationOptions(); + + // The executor service is an externally managed, thread-safe asynchronous execution pool. + @SuppressWarnings("Immutable") + @AutoValue.CopyAnnotations + abstract Optional asyncExecutor(); + @Override public Program createProgram(CelAbstractSyntaxTree ast) throws CelEvaluationException { return toRuntimeProgram(planner().plan(ast)); } - private static final CelFunctionResolver EMPTY_FUNCTION_RESOLVER = - new CelFunctionResolver() { - @Override - public Optional findOverloadMatchingArgs( - String functionName, Collection overloadIds, Object[] args) { - return Optional.empty(); - } - - @Override - public Optional findOverloadMatchingArgs( - String functionName, Object[] args) { - return Optional.empty(); - } - }; - - public Program toRuntimeProgram(dev.cel.runtime.Program program) { + private Program toRuntimeProgram(dev.cel.runtime.Program program) { return new Program() { @Override @@ -140,7 +137,7 @@ public Object eval(Message message) throws CelEvaluationException { return plannedProgram.evalOrThrow( plannedProgram.interpretable(), ProtoMessageActivationFactory.fromProto(message, plannedProgram.options()), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, /* partialVars= */ null, /* listener= */ null); } @@ -162,17 +159,55 @@ public Object eval(PartialVars partialVars) throws CelEvaluationException { return program.eval(partialVars); } + @Override + public ListenableFuture evalAsync() { + return program.evalAsync(); + } + + @Override + public ListenableFuture evalAsync(Map mapValue) { + return program.evalAsync(mapValue); + } + + @Override + public ListenableFuture evalAsync( + Map mapValue, CelFunctionResolver lateBoundFunctionResolver) { + 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); + } + + @Override + public ListenableFuture evalAsync( + CelVariableResolver resolver, CelFunctionResolver lateBoundFunctionResolver) { + return program.evalAsync(resolver, lateBoundFunctionResolver); + } + + @Override + public ListenableFuture evalAsync(PartialVars partialVars) { + return program.evalAsync(partialVars); + } + @Override public Object trace(CelEvaluationListener listener) throws CelEvaluationException { return ((PlannedProgram) program) - .trace(GlobalResolver.EMPTY, EMPTY_FUNCTION_RESOLVER, null, listener); + .trace(GlobalResolver.EMPTY, CelFunctionResolver.EMPTY, null, listener); } @Override public Object trace(Map mapValue, CelEvaluationListener listener) throws CelEvaluationException { return ((PlannedProgram) program) - .trace(Activation.copyOf(mapValue), EMPTY_FUNCTION_RESOLVER, null, listener); + .trace(Activation.copyOf(mapValue), CelFunctionResolver.EMPTY, null, listener); } @Override @@ -182,7 +217,7 @@ public Object trace(Message message, CelEvaluationListener listener) return plannedProgram.evalOrThrow( plannedProgram.interpretable(), ProtoMessageActivationFactory.fromProto(message, plannedProgram.options()), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, /* partialVars= */ null, listener); } @@ -193,7 +228,7 @@ public Object trace(CelVariableResolver resolver, CelEvaluationListener listener return ((PlannedProgram) program) .trace( (name) -> resolver.find(name).orElse(null), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, null, listener); } @@ -228,13 +263,13 @@ public Object trace(PartialVars partialVars, CelEvaluationListener listener) return ((PlannedProgram) program) .trace( (name) -> partialVars.resolver().find(name).orElse(null), - EMPTY_FUNCTION_RESOLVER, + CelFunctionResolver.EMPTY, partialVars, listener); } @Override - public Object advanceEvaluation(UnknownContext context) throws CelEvaluationException { + public Object advanceEvaluation(UnknownContext context) { throw new UnsupportedOperationException("Unsupported operation."); } }; @@ -253,7 +288,8 @@ public static Builder newBuilder() { .setFunctionBindings(ImmutableMap.of()) .setStandardFunctions(CelStandardFunctions.newBuilder().build()) .setContainer(CelContainer.newBuilder().build()) - .setExtensionRegistry(ExtensionRegistry.getEmptyRegistry()); + .setExtensionRegistry(ExtensionRegistry.getEmptyRegistry()) + .setAsyncEvaluationOptions(CelAsyncEvaluationOptions.defaultOptions()); } /** Builder for {@link CelRuntimeImpl}. */ @@ -280,6 +316,13 @@ public abstract static class Builder implements CelRuntimeBuilder { @Override public abstract Builder setContainer(CelContainer container); + @Override + public abstract Builder setAsyncEvaluationOptions( + CelAsyncEvaluationOptions asyncEvaluationOptions); + + @Override + public abstract Builder setAsyncExecutor(ListeningExecutorService asyncExecutor); + abstract CelOptions options(); abstract CelContainer container(); diff --git a/runtime/src/main/java/dev/cel/runtime/CelRuntimeLegacyImpl.java b/runtime/src/main/java/dev/cel/runtime/CelRuntimeLegacyImpl.java index 428c6dba5..68cba4bdd 100644 --- a/runtime/src/main/java/dev/cel/runtime/CelRuntimeLegacyImpl.java +++ b/runtime/src/main/java/dev/cel/runtime/CelRuntimeLegacyImpl.java @@ -20,6 +20,7 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; +import com.google.common.util.concurrent.ListeningExecutorService; import com.google.errorprone.annotations.CanIgnoreReturnValue; import javax.annotation.concurrent.ThreadSafe; import com.google.protobuf.DescriptorProtos.FileDescriptorSet; @@ -84,6 +85,8 @@ public final class CelRuntimeLegacyImpl implements CelRuntime { private final ImmutableSet celRuntimeLibraries; private final ImmutableList celFunctionBindings; + private final CelAsyncEvaluationOptions asyncEvaluationOptions; + private final @Nullable ListeningExecutorService asyncExecutor; @Override public CelRuntime.Program createProgram(CelAbstractSyntaxTree ast) { @@ -92,6 +95,7 @@ public CelRuntime.Program createProgram(CelAbstractSyntaxTree ast) { } @Override + @SuppressWarnings("deprecation") // Legacy runtime builder preserves deprecated standard env flag public CelRuntimeBuilder toRuntimeBuilder() { CelRuntimeBuilder builder = new Builder() @@ -101,7 +105,8 @@ public CelRuntimeBuilder toRuntimeBuilder() { .setExtensionRegistry(extensionRegistry) .addFileTypes(fileDescriptors) .addLibraries(celRuntimeLibraries) - .addFunctionBindings(celFunctionBindings); + .addFunctionBindings(celFunctionBindings) + .setAsyncEvaluationOptions(asyncEvaluationOptions); if (customTypeFactory != null) { builder.setTypeFactory(customTypeFactory); @@ -111,6 +116,9 @@ public CelRuntimeBuilder toRuntimeBuilder() { builder.setStandardFunctions(overriddenStandardFunctions); } + if (asyncExecutor != null) { + builder.setAsyncExecutor(asyncExecutor); + } return builder; } @@ -132,6 +140,8 @@ public static final class Builder implements CelRuntimeBuilder { @VisibleForTesting Function customTypeFactory; @VisibleForTesting CelStandardFunctions overriddenStandardFunctions; + @VisibleForTesting CelAsyncEvaluationOptions asyncEvaluationOptions; + @VisibleForTesting @Nullable ListeningExecutorService asyncExecutor; private CelOptions options; @@ -257,8 +267,22 @@ public CelRuntimeBuilder setContainer(CelContainer container) { "This method is not supported for the legacy runtime"); } + @Override + public CelRuntimeBuilder setAsyncEvaluationOptions( + CelAsyncEvaluationOptions asyncEvaluationOptions) { + this.asyncEvaluationOptions = checkNotNull(asyncEvaluationOptions); + return this; + } + + @Override + public CelRuntimeBuilder setAsyncExecutor(ListeningExecutorService asyncExecutor) { + this.asyncExecutor = checkNotNull(asyncExecutor); + return this; + } + /** Build a new {@code CelRuntimeLegacyImpl} instance from the builder config. */ @Override + @SuppressWarnings("deprecation") // Legacy runtime uses deprecated DescriptorTypeResolver public CelRuntimeLegacyImpl build() { if (standardEnvironmentEnabled && overriddenStandardFunctions != null) { throw new IllegalArgumentException( @@ -357,7 +381,9 @@ public CelRuntimeLegacyImpl build() { overriddenStandardFunctions, fileDescriptors, runtimeLibraries, - ImmutableList.copyOf(customFunctionBindings.values())); + ImmutableList.copyOf(customFunctionBindings.values()), + asyncEvaluationOptions, + asyncExecutor); } private ImmutableSet newStandardFunctionBindings( @@ -432,6 +458,8 @@ private Builder() { this.celRuntimeLibraries = ImmutableSet.builder(); this.extensionRegistry = ExtensionRegistry.getEmptyRegistry(); this.customTypeFactory = null; + this.asyncEvaluationOptions = CelAsyncEvaluationOptions.defaultOptions(); + this.asyncExecutor = null; } } @@ -444,7 +472,9 @@ private CelRuntimeLegacyImpl( @Nullable CelStandardFunctions overriddenStandardFunctions, ImmutableSet fileDescriptors, ImmutableSet celRuntimeLibraries, - ImmutableList celFunctionBindings) { + ImmutableList celFunctionBindings, + CelAsyncEvaluationOptions asyncEvaluationOptions, + @Nullable ListeningExecutorService asyncExecutor) { this.interpreter = interpreter; this.options = options; this.standardEnvironmentEnabled = standardEnvironmentEnabled; @@ -454,5 +484,7 @@ private CelRuntimeLegacyImpl( this.fileDescriptors = fileDescriptors; this.celRuntimeLibraries = celRuntimeLibraries; this.celFunctionBindings = celFunctionBindings; + this.asyncEvaluationOptions = asyncEvaluationOptions; + this.asyncExecutor = asyncExecutor; } } diff --git a/runtime/src/main/java/dev/cel/runtime/Program.java b/runtime/src/main/java/dev/cel/runtime/Program.java index e808a373c..c9df239eb 100644 --- a/runtime/src/main/java/dev/cel/runtime/Program.java +++ b/runtime/src/main/java/dev/cel/runtime/Program.java @@ -14,6 +14,7 @@ package dev.cel.runtime; +import com.google.common.util.concurrent.ListenableFuture; import com.google.errorprone.annotations.Immutable; import java.util.Map; @@ -46,4 +47,33 @@ Object eval(CelVariableResolver resolver, CelFunctionResolver lateBoundFunctionR /** Evaluate a compiled program with unknown attribute patterns {@code partialVars}. */ Object eval(PartialVars partialVars) throws CelEvaluationException; + + /** Evaluate the expression asynchronously without any variables. */ + ListenableFuture evalAsync(); + + /** + * Evaluate the expression asynchronously using a {@code mapValue} as the source of input + * variables. + */ + ListenableFuture evalAsync(Map mapValue); + + /** + * Evaluate the expression asynchronously using a {@code mapValue} as the source of input + * variables and late-bound functions {@code lateBoundFunctionResolver}. + */ + ListenableFuture evalAsync( + Map mapValue, CelFunctionResolver lateBoundFunctionResolver); + + /** Evaluate the expression asynchronously with a custom variable {@code resolver}. */ + ListenableFuture evalAsync(CelVariableResolver resolver); + + /** + * Evaluate the expression asynchronously with a custom variable {@code resolver} and late-bound + * functions {@code lateBoundFunctionResolver}. + */ + ListenableFuture evalAsync( + CelVariableResolver resolver, CelFunctionResolver lateBoundFunctionResolver); + + /** Evaluate the expression asynchronously with unknown attribute patterns {@code partialVars}. */ + ListenableFuture evalAsync(PartialVars partialVars); } diff --git a/runtime/src/main/java/dev/cel/runtime/ProgramImpl.java b/runtime/src/main/java/dev/cel/runtime/ProgramImpl.java index 2543a9525..cc6795561 100644 --- a/runtime/src/main/java/dev/cel/runtime/ProgramImpl.java +++ b/runtime/src/main/java/dev/cel/runtime/ProgramImpl.java @@ -16,6 +16,7 @@ import com.google.auto.value.AutoValue; import com.google.common.base.Preconditions; +import com.google.common.util.concurrent.ListenableFuture; import com.google.errorprone.annotations.Immutable; import com.google.protobuf.Message; import dev.cel.common.CelOptions; @@ -68,6 +69,50 @@ public Object eval(PartialVars partialVars) throws CelEvaluationException { /* listener= */ Optional.empty()); } + @Override + public ListenableFuture evalAsync() { + throw new UnsupportedOperationException( + "evalAsync is not supported by the legacy interpreter."); + } + + @Override + public ListenableFuture evalAsync(Map mapValue) { + throw new UnsupportedOperationException( + "evalAsync is not supported by the legacy interpreter."); + } + + @Override + public ListenableFuture evalAsync( + Map mapValue, CelFunctionResolver lateBoundFunctionResolver) { + throw new UnsupportedOperationException( + "evalAsync is not supported by the legacy interpreter."); + } + + @Override + public ListenableFuture evalAsync(Message message) { + throw new UnsupportedOperationException( + "evalAsync is not supported by the legacy interpreter."); + } + + @Override + public ListenableFuture evalAsync(CelVariableResolver resolver) { + throw new UnsupportedOperationException( + "evalAsync is not supported by the legacy interpreter."); + } + + @Override + public ListenableFuture evalAsync( + CelVariableResolver resolver, CelFunctionResolver lateBoundFunctionResolver) { + throw new UnsupportedOperationException( + "evalAsync is not supported by the legacy interpreter."); + } + + @Override + public ListenableFuture evalAsync(PartialVars partialVars) { + throw new UnsupportedOperationException( + "evalAsync is not supported by the legacy interpreter."); + } + @Override public Object trace(CelEvaluationListener listener) throws CelEvaluationException { return evalInternal(Activation.EMPTY, listener); diff --git a/runtime/src/main/java/dev/cel/runtime/planner/AsyncCompletionCoordinator.java b/runtime/src/main/java/dev/cel/runtime/planner/AsyncCompletionCoordinator.java new file mode 100644 index 000000000..716608ea8 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/planner/AsyncCompletionCoordinator.java @@ -0,0 +1,267 @@ +// 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 java.util.Objects.requireNonNull; +import static java.util.concurrent.TimeUnit.NANOSECONDS; + +import com.google.common.collect.ImmutableList; +import com.google.errorprone.annotations.concurrent.GuardedBy; +import dev.cel.runtime.CelAsyncCall; +import dev.cel.runtime.CelAsyncDrainAction; +import dev.cel.runtime.CelAsyncEvaluationOptions; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import org.jspecify.annotations.Nullable; + +/** + * Coordinates asynchronous call completion notifications, debouncing, and re-evaluation dispatch. + */ +final class AsyncCompletionCoordinator { + private final Object lock; + private final CelAsyncEvaluationOptions options; + private final AsyncGate gate; + private final Executor executor; + + @GuardedBy("lock") + private final List completedBatch; + + @GuardedBy("lock") + private @Nullable Runnable continuation; + + @GuardedBy("lock") + private @Nullable ScheduledFuture debounceTimer; + + @GuardedBy("lock") + private boolean isWaiting; + + @GuardedBy("lock") + private boolean isCancelled; + + @GuardedBy("lock") + private long cycleId; + + AsyncCompletionCoordinator(CelAsyncEvaluationOptions options, AsyncGate gate, Executor executor) { + this.lock = new Object(); + this.options = requireNonNull(options, "options must not be null"); + this.gate = requireNonNull(gate, "gate must not be null"); + this.executor = requireNonNull(executor, "executor must not be null"); + this.completedBatch = new ArrayList<>(); + this.isWaiting = false; + this.isCancelled = false; + this.cycleId = 0; + } + + boolean hasPendingBatch() { + synchronized (lock) { + return !completedBatch.isEmpty(); + } + } + + void notifyCallCompleted(CelAsyncCall call) { + requireNonNull(call, "call must not be null"); + ImmutableList batchSnapshot; + int activeCount; + long currentCycleId; + + synchronized (lock) { + if (isCancelled) { + return; + } + completedBatch.add(call); + if (!isWaiting) { + return; + } + batchSnapshot = ImmutableList.copyOf(completedBatch); + activeCount = gate.activeCount(); + currentCycleId = cycleId; + } + + // Alien code: evaluate drain strategy outside monitor lock + CelAsyncDrainAction action = options.drainStrategy().nextAction(batchSnapshot, activeCount); + applyDrainAction(action, currentCycleId); + } + + void waitForCompletions(Runnable continuationCallback) { + requireNonNull(continuationCallback, "continuationCallback must not be null"); + ImmutableList batchSnapshot; + int activeCount; + long currentCycleId; + + synchronized (lock) { + if (isCancelled) { + throw new IllegalStateException("Coordinator has been cancelled"); + } + if (isWaiting) { + throw new IllegalStateException("Coordinator is already waiting for completions"); + } + this.continuation = continuationCallback; + this.isWaiting = true; + this.cycleId++; + batchSnapshot = ImmutableList.copyOf(completedBatch); + activeCount = gate.activeCount(); + currentCycleId = this.cycleId; + if (batchSnapshot.isEmpty() && activeCount > 0) { + return; + } + } + + // Alien code: evaluate drain strategy outside monitor lock + CelAsyncDrainAction action = options.drainStrategy().nextAction(batchSnapshot, activeCount); + applyDrainAction(action, currentCycleId); + } + + private void applyDrainAction(CelAsyncDrainAction action, long currentCycleId) { + Runnable toRun = null; + ScheduledFuture timerToCancel = null; + DebounceRequest debounceRequest = null; + + synchronized (lock) { + if (!isCancelled && isWaiting && this.cycleId == currentCycleId) { + if (action.shouldReevaluate()) { + timerToCancel = cancelDebounceTimerUnderLock(); + toRun = drainAndResetUnderLock(); + } else if (action.waitDuration().isZero()) { + // Indefinite wait for next completion: cancel any pending timer + timerToCancel = cancelDebounceTimerUnderLock(); + } else { + // Sliding window debounce: reset existing timer and reschedule for new wait duration + timerToCancel = cancelDebounceTimerUnderLock(); + debounceRequest = new DebounceRequest(action.waitDuration(), this.cycleId); + } + } + } + + if (timerToCancel != null) { + timerToCancel.cancel(false); + } + if (toRun != null) { + executor.execute(toRun); + } else if (debounceRequest != null) { + scheduleDebounce(debounceRequest.waitDuration().toNanos(), debounceRequest.cycleId()); + } + } + + void cancel() { + ScheduledFuture timerToCancel; + synchronized (lock) { + isCancelled = true; + cycleId++; + timerToCancel = cancelDebounceTimerUnderLock(); + isWaiting = false; + continuation = null; + completedBatch.clear(); + } + if (timerToCancel != null) { + timerToCancel.cancel(false); + } + } + + private void scheduleDebounce(long nanos, long scheduledCycleId) { + ScheduledExecutorService scheduler = options.resolveScheduledExecutorService(); + ScheduledFuture future = + scheduler.schedule(() -> onDebounceFired(scheduledCycleId), nanos, NANOSECONDS); + ScheduledFuture redundantFuture = null; + synchronized (lock) { + if (!isCancelled && isWaiting && this.cycleId == scheduledCycleId) { + if (debounceTimer != null) { + redundantFuture = debounceTimer; + } + debounceTimer = future; + } else { + redundantFuture = future; + } + } + if (redundantFuture != null) { + redundantFuture.cancel(false); + } + } + + void onDebounceFired(long firedCycleId) { + Runnable toRun = null; + synchronized (lock) { + if (!isCancelled && isWaiting && this.cycleId == firedCycleId) { + debounceTimer = null; + toRun = drainAndResetUnderLock(); + } + } + if (toRun != null) { + executor.execute(toRun); + } + } + + @GuardedBy("lock") + private @Nullable Runnable drainAndResetUnderLock() { + cycleId++; + isWaiting = false; + completedBatch.clear(); + Runnable run = continuation; + continuation = null; + return run; + } + + @GuardedBy("lock") + private @Nullable ScheduledFuture cancelDebounceTimerUnderLock() { + ScheduledFuture timer = debounceTimer; + debounceTimer = null; + return timer; + } + + boolean isWaiting() { + synchronized (lock) { + return isWaiting; + } + } + + boolean hasContinuation() { + synchronized (lock) { + return continuation != null; + } + } + + boolean hasScheduledDebounceTimer() { + synchronized (lock) { + return debounceTimer != null; + } + } + + long cycleId() { + synchronized (lock) { + return cycleId; + } + } + + private static final class DebounceRequest { + private final Duration waitDuration; + private final long cycleId; + + DebounceRequest(Duration waitDuration, long cycleId) { + this.waitDuration = requireNonNull(waitDuration, "waitDuration must not be null"); + this.cycleId = cycleId; + } + + Duration waitDuration() { + return waitDuration; + } + + long cycleId() { + return cycleId; + } + } +} diff --git a/runtime/src/main/java/dev/cel/runtime/planner/AsyncGate.java b/runtime/src/main/java/dev/cel/runtime/planner/AsyncGate.java new file mode 100644 index 000000000..206669c66 --- /dev/null +++ b/runtime/src/main/java/dev/cel/runtime/planner/AsyncGate.java @@ -0,0 +1,140 @@ +// 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 java.util.Objects.requireNonNull; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.jspecify.annotations.Nullable; + +/** Regulates the number of concurrent asynchronous function executions based on maxConcurrency. */ +final class AsyncGate { + private final @Nullable Semaphore semaphore; + private final Queue pendingTasks; + private final AtomicInteger activeCount; + private final AtomicBoolean cancelled; + + AsyncGate(int maxConcurrency) { + this(maxConcurrency, new ConcurrentLinkedQueue<>()); + } + + AsyncGate(int maxConcurrency, Queue pendingTasks) { + this(maxConcurrency > 0 ? new Semaphore(maxConcurrency) : null, pendingTasks); + } + + AsyncGate(@Nullable Semaphore semaphore, Queue pendingTasks) { + this.semaphore = semaphore; + this.pendingTasks = requireNonNull(pendingTasks, "pendingTasks must not be null"); + this.activeCount = new AtomicInteger(); + this.cancelled = new AtomicBoolean(false); + } + + void cancel() { + cancelled.set(true); + pendingTasks.clear(); + } + + void dispatch(Executor executor, Runnable task) { + requireNonNull(executor, "executor must not be null"); + requireNonNull(task, "task must not be null"); + if (cancelled.get()) { + return; + } + + if (semaphore == null) { + activeCount.incrementAndGet(); + try { + task.run(); + } catch (Throwable e) { + activeCount.decrementAndGet(); + throw e; + } + return; + } + + if (pendingTasks.isEmpty() && semaphore.tryAcquire()) { + activeCount.incrementAndGet(); + try { + task.run(); + } catch (Throwable e) { + activeCount.decrementAndGet(); + semaphore.release(); + throw e; + } + return; + } + + pendingTasks.add(task); + drainPending(executor); + } + + void releasePermit(Executor executor) { + requireNonNull(executor, "executor must not be null"); + activeCount.decrementAndGet(); + if (semaphore != null) { + semaphore.release(); + drainPending(executor); + } + } + + private void drainPending(Executor executor) { + if (semaphore == null || cancelled.get()) { + pendingTasks.clear(); + return; + } + while (!pendingTasks.isEmpty()) { + if (!semaphore.tryAcquire()) { + break; + } + Runnable task = pendingTasks.poll(); + if (task != null) { + activeCount.incrementAndGet(); + AtomicBoolean permitCleanedUp = new AtomicBoolean(false); + try { + executor.execute( + () -> { + try { + task.run(); + } catch (Throwable t) { + permitCleanedUp.set(true); + activeCount.decrementAndGet(); + semaphore.release(); + drainPending(executor); + throw t; + } + }); + } catch (Throwable e) { + if (!permitCleanedUp.get()) { + activeCount.decrementAndGet(); + semaphore.release(); + } + throw e; + } + } else { + semaphore.release(); + break; + } + } + } + + int activeCount() { + return activeCount.get(); + } +} 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 ca7665953..4f0331fe4 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel +++ b/runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel @@ -92,6 +92,7 @@ java_library( "//runtime:resolved_overload", "//runtime:variable_resolver", "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", "@maven//:org_jspecify_jspecify", ], ) @@ -186,6 +187,32 @@ java_library( ], ) +java_library( + name = "async_gate", + srcs = ["AsyncGate.java"], + tags = [ + ], + deps = [ + "@maven//:org_jspecify_jspecify", + ], +) + +java_library( + name = "async_completion_coordinator", + srcs = ["AsyncCompletionCoordinator.java"], + tags = [ + ], + deps = [ + ":async_gate", + "//runtime:async_call", + "//runtime:async_drain_strategy", + "//runtime:async_options", + "@maven//:com_google_errorprone_error_prone_annotations", + "@maven//:com_google_guava_guava", + "@maven//:org_jspecify_jspecify", + ], +) + java_library( name = "activation_wrapper", srcs = ["ActivationWrapper.java"], @@ -622,6 +649,7 @@ cel_android_library( "//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", ], ) @@ -715,6 +743,32 @@ cel_android_library( ], ) +cel_android_library( + name = "async_gate_android", + srcs = ["AsyncGate.java"], + tags = [ + ], + deps = [ + "@maven//:org_jspecify_jspecify", + ], +) + +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_errorprone_error_prone_annotations", + "@maven//:org_jspecify_jspecify", + "@maven_android//:com_google_guava_guava", + ], +) + cel_android_library( name = "activation_wrapper_android", srcs = ["ActivationWrapper.java"], diff --git a/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java b/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java index 1470e4909..f7f3d7f01 100644 --- a/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java +++ b/runtime/src/main/java/dev/cel/runtime/planner/PlannedProgram.java @@ -15,6 +15,7 @@ package dev.cel.runtime.planner; import com.google.auto.value.AutoValue; +import com.google.common.util.concurrent.ListenableFuture; import com.google.errorprone.annotations.Immutable; import dev.cel.common.CelOptions; import dev.cel.common.annotations.Internal; @@ -129,6 +130,38 @@ public Object eval(PartialVars partialVars) throws CelEvaluationException { /* listener= */ null); } + @Override + public ListenableFuture evalAsync() { + throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + } + + @Override + public ListenableFuture evalAsync(Map mapValue) { + throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + } + + @Override + public ListenableFuture evalAsync( + Map mapValue, CelFunctionResolver lateBoundFunctionResolver) { + throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + } + + @Override + public ListenableFuture evalAsync(CelVariableResolver resolver) { + throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + } + + @Override + public ListenableFuture evalAsync( + CelVariableResolver resolver, CelFunctionResolver lateBoundFunctionResolver) { + throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + } + + @Override + public ListenableFuture evalAsync(PartialVars partialVars) { + throw new UnsupportedOperationException("evalAsync is not supported by PlannedProgram."); + } + public Object evalOrThrow( PlannedInterpretable interpretable, GlobalResolver resolver, diff --git a/runtime/src/test/java/dev/cel/runtime/BUILD.bazel b/runtime/src/test/java/dev/cel/runtime/BUILD.bazel index a2e44223a..f898b66fe 100644 --- a/runtime/src/test/java/dev/cel/runtime/BUILD.bazel +++ b/runtime/src/test/java/dev/cel/runtime/BUILD.bazel @@ -43,6 +43,7 @@ java_library( "//common/exceptions:bad_format", "//common/exceptions:divide_by_zero", "//common/exceptions:numeric_overflow", + "//common/exceptions:overload_not_found", "//common/exceptions:runtime_exception", "//common/internal:cel_descriptor_pools", "//common/internal:converter", @@ -69,9 +70,9 @@ java_library( "//runtime:evaluation_exception_builder", "//runtime:evaluation_listener", "//runtime:function_binding", + "//runtime:function_resolver", "//runtime:interpretable", "//runtime:interpreter", - "//runtime:interpreter_util", "//runtime:late_function_binding", "//runtime:lite_runtime", "//runtime:lite_runtime_factory", diff --git a/runtime/src/test/java/dev/cel/runtime/CelAsyncDrainStrategyTest.java b/runtime/src/test/java/dev/cel/runtime/CelAsyncDrainStrategyTest.java new file mode 100644 index 000000000..e93b8688c --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/CelAsyncDrainStrategyTest.java @@ -0,0 +1,299 @@ +// 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; + +import static com.google.common.truth.Truth.assertThat; +import static org.junit.Assert.assertThrows; + +import com.google.common.collect.ImmutableList; +import java.time.Duration; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public final class CelAsyncDrainStrategyTest { + + private static final CelAsyncCall DUMMY_CALL = + new CelAsyncCall() { + @Override + public long callId() { + return 1L; + } + + @Override + public long exprId() { + return 10L; + } + + @Override + public String functionName() { + return "fn"; + } + + @Override + public String overloadId() { + return "fn_overload"; + } + }; + + @Test + public void drainReady_defaultDebounce_noActiveCalls_reevaluates() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(), 0); + + assertThat(action.shouldReevaluate()).isTrue(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainReady_defaultDebounce_emptyBatchWithActiveCalls_waitsForMore() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(), 3); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainReady_defaultDebounce_hasBatchWithActiveCalls_waitsDefaultDuration() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(DUMMY_CALL), 2); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ofNanos(100_000)); + } + + @Test + public void drainReady_zeroDebounce_hasCompletedBatch_reevaluates() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(Duration.ZERO); + + CelAsyncDrainAction actionWithActive = strategy.nextAction(ImmutableList.of(DUMMY_CALL), 5); + + assertThat(actionWithActive.shouldReevaluate()).isTrue(); + assertThat(actionWithActive.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainReady_zeroDebounce_emptyBatchWithActiveCalls_waitsForMore() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(Duration.ZERO); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(), 5); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainReady_withDebounce_activeZero_reevaluates() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(Duration.ofMillis(50)); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(), 0); + + assertThat(action.shouldReevaluate()).isTrue(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainReady_withDebounce_emptyBatchAndActiveCalls_waitsForMore() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(Duration.ofMillis(50)); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(), 3); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainReady_withDebounce_hasCompletedBatchAndActiveCalls_waitsDuration() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(Duration.ofMillis(50)); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(DUMMY_CALL), 2); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ofMillis(50)); + } + + @Test + public void drainNone_noActiveCalls_reevaluates() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainNone(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(), 0); + + assertThat(action.shouldReevaluate()).isTrue(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainNone_withCompletedBatchAndActiveCalls_reevaluates() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainNone(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(DUMMY_CALL), 2); + + assertThat(action.shouldReevaluate()).isTrue(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainNone_emptyBatchWithActiveCalls_waitsForMore() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainNone(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(), 2); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainAll_noActiveCalls_reevaluates() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainAll(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(), 0); + + assertThat(action.shouldReevaluate()).isTrue(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainAll_withCompletedBatchAndNoActiveCalls_reevaluates() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainAll(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(DUMMY_CALL), 0); + + assertThat(action.shouldReevaluate()).isTrue(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainAll_withCompletedBatchAndActiveCalls_waitsForMore() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainAll(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(DUMMY_CALL), 1); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainAll_emptyBatchWithActiveCalls_waitsForMore() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainAll(); + + CelAsyncDrainAction action = strategy.nextAction(ImmutableList.of(), 1); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainReady_nullDebounce_throwsException() { + assertThrows(NullPointerException.class, () -> CelAsyncDrainStrategy.drainReady(null)); + } + + @Test + public void drainReady_nullCompletedBatch_throwsException() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(); + + assertThrows(NullPointerException.class, () -> strategy.nextAction(null, 1)); + } + + @Test + public void drainReady_negativeActiveCalls_throwsException() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainReady(); + + assertThrows(IllegalArgumentException.class, () -> strategy.nextAction(ImmutableList.of(), -1)); + } + + @Test + public void drainNone_nullCompletedBatch_throwsException() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainNone(); + + assertThrows(NullPointerException.class, () -> strategy.nextAction(null, 1)); + } + + @Test + public void drainNone_negativeActiveCalls_throwsException() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainNone(); + + assertThrows(IllegalArgumentException.class, () -> strategy.nextAction(ImmutableList.of(), -1)); + } + + @Test + public void drainAll_nullCompletedBatch_throwsException() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainAll(); + + assertThrows(NullPointerException.class, () -> strategy.nextAction(null, 1)); + } + + @Test + public void drainAll_negativeActiveCalls_throwsException() { + CelAsyncDrainStrategy strategy = CelAsyncDrainStrategy.drainAll(); + + assertThrows(IllegalArgumentException.class, () -> strategy.nextAction(ImmutableList.of(), -1)); + } + + @Test + public void drainReady_negativeDebounce_throwsException() { + assertThrows( + IllegalArgumentException.class, + () -> CelAsyncDrainStrategy.drainReady(Duration.ofMillis(-1))); + } + + @Test + public void drainAction_reevaluate() { + CelAsyncDrainAction action = CelAsyncDrainAction.reevaluate(); + + assertThat(action.shouldReevaluate()).isTrue(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainAction_waitForMore() { + CelAsyncDrainAction action = CelAsyncDrainAction.waitForMore(); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainAction_waitDuration_success() { + CelAsyncDrainAction action = CelAsyncDrainAction.waitDuration(Duration.ofSeconds(2)); + + assertThat(action.shouldReevaluate()).isFalse(); + assertThat(action.waitDuration()).isEqualTo(Duration.ofSeconds(2)); + } + + @Test + public void drainAction_waitZero_reevaluates() { + CelAsyncDrainAction action = CelAsyncDrainAction.waitDuration(Duration.ZERO); + + assertThat(action.shouldReevaluate()).isTrue(); + assertThat(action.waitDuration()).isEqualTo(Duration.ZERO); + } + + @Test + public void drainAction_waitDuration_negative_throwsException() { + assertThrows( + IllegalArgumentException.class, + () -> CelAsyncDrainAction.waitDuration(Duration.ofMillis(-5))); + } + + @Test + public void drainAction_waitDuration_null_throwsException() { + assertThrows(NullPointerException.class, () -> CelAsyncDrainAction.waitDuration(null)); + } +} diff --git a/runtime/src/test/java/dev/cel/runtime/CelAsyncEvaluationOptionsTest.java b/runtime/src/test/java/dev/cel/runtime/CelAsyncEvaluationOptionsTest.java new file mode 100644 index 000000000..fc26513e7 --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/CelAsyncEvaluationOptionsTest.java @@ -0,0 +1,120 @@ +// 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; + +import static com.google.common.truth.Truth.assertThat; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertThrows; + +import com.google.common.collect.ImmutableList; +import com.google.testing.junit.testparameterinjector.TestParameterInjector; +import java.time.Duration; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import org.jspecify.annotations.Nullable; +import org.junit.After; +import org.junit.Test; +import org.junit.runner.RunWith; + +@RunWith(TestParameterInjector.class) +public final class CelAsyncEvaluationOptionsTest { + + private ScheduledExecutorService customScheduler; + + @After + public void tearDown() { + if (customScheduler != null) { + customScheduler.shutdown(); + } + } + + @Test + public void defaultOptions_returnsDefaultValues() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.defaultOptions(); + + assertThat(options.maxConcurrency()).isEqualTo(100); + assertThat(options.maxIterations()).isEqualTo(1_000); + assertThat(options.drainStrategy()).isNotNull(); + assertThat(options.observer()).isEmpty(); + assertThat(options.scheduledExecutorService()).isEmpty(); + assertThat(options.resolveScheduledExecutorService()).isNotNull(); + } + + @Test + public void resolveScheduledExecutorService_defaultScheduler_runsAsDaemonThread() + throws Exception { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.defaultOptions(); + + Future isDaemonFuture = + options.resolveScheduledExecutorService().submit(() -> Thread.currentThread().isDaemon()); + + assertThat(isDaemonFuture.get(5, SECONDS)).isTrue(); + } + + @Test + public void builder_validations() { + CelAsyncEvaluationOptions.Builder builder = CelAsyncEvaluationOptions.builder(); + + assertThrows(NullPointerException.class, () -> builder.setDrainStrategy(null)); + assertThrows(NullPointerException.class, () -> builder.setObserver(null)); + assertThrows(NullPointerException.class, () -> builder.setScheduledExecutorService(null)); + } + + @Test + public void builder_nonPositiveMaxConcurrency_roundTripsCleanly() { + CelAsyncEvaluationOptions unboundedZero = + CelAsyncEvaluationOptions.builder().setMaxConcurrency(0).build(); + CelAsyncEvaluationOptions unboundedNegative = + CelAsyncEvaluationOptions.builder().setMaxConcurrency(-1).build(); + + assertThat(unboundedZero.maxConcurrency()).isEqualTo(0); + assertThat(unboundedNegative.maxConcurrency()).isEqualTo(-1); + } + + @Test + public void builder_customValuesAndRoundTrip() { + CelAsyncDrainStrategy drainStrategy = CelAsyncDrainStrategy.drainReady(Duration.ofMillis(25)); + CelAsyncObserver observer = + new CelAsyncObserver() { + @Override + public void onCallStarted(CelAsyncCall call, ImmutableList args) {} + + @Override + public void onCallFinished( + CelAsyncCall call, @Nullable Object result, @Nullable Throwable error) {} + }; + customScheduler = Executors.newSingleThreadScheduledExecutor(); + + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setMaxConcurrency(8) + .setMaxIterations(50) + .setDrainStrategy(drainStrategy) + .setObserver(observer) + .setScheduledExecutorService(customScheduler) + .build(); + + assertThat(options.maxConcurrency()).isEqualTo(8); + assertThat(options.maxIterations()).isEqualTo(50); + assertThat(options.drainStrategy()).isSameInstanceAs(drainStrategy); + assertThat(options.observer()).hasValue(observer); + assertThat(options.scheduledExecutorService()).hasValue(customScheduler); + assertThat(options.resolveScheduledExecutorService()).isSameInstanceAs(customScheduler); + + CelAsyncEvaluationOptions copy = options.toBuilder().build(); + assertThat(copy).isEqualTo(options); + } +} diff --git a/runtime/src/test/java/dev/cel/runtime/CelRuntimeLegacyImplTest.java b/runtime/src/test/java/dev/cel/runtime/CelRuntimeLegacyImplTest.java index fec5fab41..972b9b1d2 100644 --- a/runtime/src/test/java/dev/cel/runtime/CelRuntimeLegacyImplTest.java +++ b/runtime/src/test/java/dev/cel/runtime/CelRuntimeLegacyImplTest.java @@ -15,7 +15,11 @@ package dev.cel.runtime; import static com.google.common.truth.Truth.assertThat; +import static com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService; +import static org.junit.Assert.assertThrows; +import com.google.common.collect.ImmutableMap; +import com.google.common.util.concurrent.ListeningExecutorService; import com.google.protobuf.Message; import dev.cel.common.CelException; import dev.cel.common.exceptions.CelDivideByZeroException; @@ -23,8 +27,8 @@ import dev.cel.compiler.CelCompilerFactory; import dev.cel.expr.conformance.proto3.TestAllTypes; import dev.cel.runtime.CelStandardFunctions.StandardFunction; +import java.util.Optional; import java.util.function.Function; -import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -37,7 +41,7 @@ public void evalException() throws CelException { CelCompiler compiler = CelCompilerFactory.standardCelCompilerBuilder().build(); CelRuntime runtime = CelRuntimeFactory.standardCelRuntimeBuilder().build(); CelRuntime.Program program = runtime.createProgram(compiler.compile("1/0").getAst()); - CelEvaluationException e = Assert.assertThrows(CelEvaluationException.class, program::eval); + CelEvaluationException e = assertThrows(CelEvaluationException.class, program::eval); assertThat(e).hasCauseThat().isInstanceOf(CelDivideByZeroException.class); } @@ -102,6 +106,8 @@ public void toRuntimeBuilder_collectionProperties_areImmutable() { } @Test + @SuppressWarnings( + "deprecation") // Tests deprecated setStandardEnvironmentEnabled on legacy builder public void toRuntimeBuilder_optionalProperties() { Function customTypeFactory = (typeName) -> TestAllTypes.newBuilder(); CelStandardFunctions overriddenStandardFunctions = @@ -120,4 +126,44 @@ public void toRuntimeBuilder_optionalProperties() { assertThat(newRuntimeBuilder.overriddenStandardFunctions) .isEqualTo(overriddenStandardFunctions); } + + @Test + public void toRuntimeBuilder_asyncProperties_copied() { + ListeningExecutorService executor = newDirectExecutorService(); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.newBuilder().setMaxConcurrency(5).build(); + CelRuntimeBuilder celRuntimeBuilder = + CelRuntimeFactory.standardCelRuntimeBuilder() + .setAsyncEvaluationOptions(options) + .setAsyncExecutor(executor); + CelRuntime celRuntime = celRuntimeBuilder.build(); + + CelRuntimeLegacyImpl.Builder newRuntimeBuilder = + (CelRuntimeLegacyImpl.Builder) celRuntime.toRuntimeBuilder(); + + assertThat(newRuntimeBuilder.asyncEvaluationOptions).isEqualTo(options); + assertThat(newRuntimeBuilder.asyncExecutor).isEqualTo(executor); + } + + @Test + public void evalAsync_legacyInterpreter_throwsUnsupportedOperationException() throws Exception { + CelCompiler compiler = CelCompilerFactory.standardCelCompilerBuilder().build(); + CelRuntime runtime = CelRuntimeFactory.standardCelRuntimeBuilder().build(); + CelRuntime.Program program = runtime.createProgram(compiler.compile("1 + 1").getAst()); + CelVariableResolver resolver = name -> Optional.of(1L); + + assertThrows(UnsupportedOperationException.class, program::evalAsync); + assertThrows(UnsupportedOperationException.class, () -> program.evalAsync(ImmutableMap.of())); + assertThrows( + UnsupportedOperationException.class, + () -> program.evalAsync(ImmutableMap.of(), CelFunctionResolver.EMPTY)); + assertThrows( + UnsupportedOperationException.class, + () -> program.evalAsync(TestAllTypes.getDefaultInstance())); + assertThrows(UnsupportedOperationException.class, () -> program.evalAsync(resolver)); + assertThrows( + UnsupportedOperationException.class, + () -> program.evalAsync(resolver, CelFunctionResolver.EMPTY)); + assertThrows(UnsupportedOperationException.class, () -> program.evalAsync((PartialVars) null)); + } } diff --git a/runtime/src/test/java/dev/cel/runtime/FunctionBindingImplTest.java b/runtime/src/test/java/dev/cel/runtime/FunctionBindingImplTest.java new file mode 100644 index 000000000..0395cf560 --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/FunctionBindingImplTest.java @@ -0,0 +1,365 @@ +// 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; + +import static com.google.common.truth.Truth.assertThat; +import static com.google.common.util.concurrent.Futures.immediateFuture; +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.collect.Iterables; +import com.google.testing.junit.testparameterinjector.TestParameterInjector; +import dev.cel.common.exceptions.CelOverloadNotFoundException; +import org.junit.Test; +import org.junit.runner.RunWith; + +@RunWith(TestParameterInjector.class) +public final class FunctionBindingImplTest { + + @Test + public void dynamicDispatch_unaryOptimizedOverload_invokesOptimizedUnaryApply() throws Exception { + OptimizedFunctionOverload mockOverload = + new OptimizedFunctionOverload() { + @Override + public Object apply(Object arg) { + return (Long) arg * 10L; + } + + @Override + public Object apply(Object[] args) { + throw new AssertionError("Should not invoke array apply for unary optimized overload!"); + } + }; + CelFunctionBinding b1 = + CelFunctionBinding.from( + "custom_opt_unary_long", ImmutableList.of(Long.class), mockOverload); + CelFunctionBinding b2 = + CelFunctionBinding.from("custom_opt_unary_str", String.class, (String arg) -> arg + "!"); + ImmutableSet bindings = + FunctionBindingImpl.groupOverloadsToFunction("custom_opt_unary", ImmutableSet.of(b1, b2)); + OptimizedFunctionOverload overload = + (OptimizedFunctionOverload) + Iterables.find(bindings, b -> b.getOverloadId().equals("custom_opt_unary")) + .getDefinition(); + + assertThat(overload.apply(5L)).isEqualTo(50L); + assertThat(overload.apply("test")).isEqualTo("test!"); + } + + @Test + public void dynamicDispatch_binaryOptimizedOverload_invokesOptimizedBinaryApply() + throws Exception { + OptimizedFunctionOverload mockOverload = + new OptimizedFunctionOverload() { + @Override + public Object apply(Object arg1, Object arg2) { + return (Long) arg1 + (Long) arg2 + 100L; + } + + @Override + public Object apply(Object[] args) { + throw new AssertionError( + "Should not invoke array apply for binary optimized overload!"); + } + }; + CelFunctionBinding b1 = + CelFunctionBinding.from( + "custom_opt_binary_long", ImmutableList.of(Long.class, Long.class), mockOverload); + CelFunctionBinding b2 = + CelFunctionBinding.from( + "custom_opt_bin_str", String.class, String.class, (String a, String b) -> a + b); + ImmutableSet bindings = + FunctionBindingImpl.groupOverloadsToFunction("custom_opt_binary", ImmutableSet.of(b1, b2)); + OptimizedFunctionOverload overload = + (OptimizedFunctionOverload) + Iterables.find(bindings, b -> b.getOverloadId().equals("custom_opt_binary")) + .getDefinition(); + + assertThat(overload.apply(10L, 20L)).isEqualTo(130L); + assertThat(overload.apply("foo", "bar")).isEqualTo("foobar"); + } + + @Test + public void fromAsync_unary_invokesAsyncAndAdapts() throws Exception { + CelFunctionBinding binding = + CelFunctionBinding.fromAsync( + "unary_async", Long.class, (Long arg) -> immediateFuture(arg * 3L)); + CelAsyncFunctionOverload overload = (CelAsyncFunctionOverload) binding.getDefinition(); + + assertThat(binding.getOverloadId()).isEqualTo("unary_async"); + assertThat(binding.getArgTypes()).containsExactly(Long.class); + assertThat(binding.isStrict()).isTrue(); + assertThat(binding.getDefinition()).isInstanceOf(CelAsyncFunctionOverload.class); + assertThat(overload.applyAsync(10L).get(5, SECONDS)).isEqualTo(30L); + assertThat(overload.applyAsync(new Object[] {10L}).get(5, SECONDS)).isEqualTo(30L); + } + + @Test + public void fromAsync_binary_invokesAsyncAndAdapts() throws Exception { + CelFunctionBinding binding = + CelFunctionBinding.fromAsync( + "binary_async", Long.class, Long.class, (Long a, Long b) -> immediateFuture(a + b)); + CelAsyncFunctionOverload overload = (CelAsyncFunctionOverload) binding.getDefinition(); + + assertThat(binding.getOverloadId()).isEqualTo("binary_async"); + assertThat(binding.getArgTypes()).containsExactly(Long.class, Long.class).inOrder(); + assertThat(binding.isStrict()).isTrue(); + assertThat(binding.getDefinition()).isInstanceOf(CelAsyncFunctionOverload.class); + assertThat(overload.applyAsync(15L, 25L).get(5, SECONDS)).isEqualTo(40L); + assertThat(overload.applyAsync(new Object[] {15L, 25L}).get(5, SECONDS)).isEqualTo(40L); + } + + @Test + public void fromAsync_nullArguments_throwsNullPointerException() { + assertThrows( + NullPointerException.class, + () -> CelFunctionBinding.fromAsync(null, Long.class, (Long x) -> immediateFuture(x))); + assertThrows( + NullPointerException.class, + () -> CelFunctionBinding.fromAsync("id", null, (Long x) -> immediateFuture(x))); + assertThrows( + NullPointerException.class, + () -> + CelFunctionBinding.fromAsync( + "id", Long.class, (CelAsyncFunctionOverload.Unary) null)); + + assertThrows( + NullPointerException.class, + () -> + CelFunctionBinding.fromAsync( + null, Long.class, String.class, (a, b) -> immediateFuture(a))); + assertThrows( + NullPointerException.class, + () -> CelFunctionBinding.fromAsync("id", null, String.class, (a, b) -> immediateFuture(a))); + assertThrows( + NullPointerException.class, + () -> CelFunctionBinding.fromAsync("id", Long.class, null, (a, b) -> immediateFuture(a))); + assertThrows( + NullPointerException.class, + () -> + CelFunctionBinding.fromAsync( + "id", + Long.class, + String.class, + (CelAsyncFunctionOverload.Binary) null)); + + assertThrows( + NullPointerException.class, + () -> + CelFunctionBinding.fromAsync( + null, ImmutableList.of(Long.class), args -> immediateFuture(1L))); + assertThrows( + NullPointerException.class, + () -> + CelFunctionBinding.fromAsync( + "id", (Iterable>) null, args -> immediateFuture(1L))); + assertThrows( + NullPointerException.class, + () -> + CelFunctionBinding.fromAsync( + "id", ImmutableList.of(Long.class), (CelAsyncFunctionOverload) null)); + } + + @Test + public void fromAsync_unary_lambdaReturnsNull_returnsNullFuture() throws Exception { + CelFunctionBinding unaryBinding = + CelFunctionBinding.fromAsync( + "null_async", Long.class, (CelAsyncFunctionOverload.Unary) (Long arg) -> null); + CelAsyncFunctionOverload unaryOverload = + (CelAsyncFunctionOverload) unaryBinding.getDefinition(); + + assertThat(unaryOverload.applyAsync(1L)).isNull(); + assertThat(unaryOverload.applyAsync(new Object[] {1L})).isNull(); + } + + @Test + public void fromAsync_binary_lambdaReturnsNull_returnsNullFuture() throws Exception { + CelFunctionBinding binaryBinding = + CelFunctionBinding.fromAsync( + "null_async_bin", + Long.class, + String.class, + (CelAsyncFunctionOverload.Binary) (Long a, String b) -> null); + CelAsyncFunctionOverload binaryOverload = + (CelAsyncFunctionOverload) binaryBinding.getDefinition(); + + assertThat(binaryOverload.applyAsync(1L, "a")).isNull(); + assertThat(binaryOverload.applyAsync(new Object[] {1L, "a"})).isNull(); + } + + @Test + public void fromAsync_unary_synchronousApplyThrowsUnsupportedOperationException() { + CelFunctionBinding binding = + CelFunctionBinding.fromAsync("unary_async", Long.class, (Long arg) -> immediateFuture(arg)); + CelFunctionOverload overload = binding.getDefinition(); + + UnsupportedOperationException e = + assertThrows(UnsupportedOperationException.class, () -> overload.apply(new Object[] {10L})); + + assertThat(e) + .hasMessageThat() + .contains("Async overload cannot be evaluated synchronously. Use evalAsync instead."); + } + + @Test + public void fromAsync_varargs_invokesAsyncAndAdapts() throws Exception { + CelFunctionBinding binding = + CelFunctionBinding.fromAsync( + "custom_async_varargs", + ImmutableList.of(Long.class, String.class), + (Object[] args) -> immediateFuture((Long) args[0] + (String) args[1])); + CelAsyncFunctionOverload overload = (CelAsyncFunctionOverload) binding.getDefinition(); + + assertThat(binding.getOverloadId()).isEqualTo("custom_async_varargs"); + assertThat(binding.getArgTypes()).containsExactly(Long.class, String.class).inOrder(); + assertThat(binding.isStrict()).isTrue(); + assertThat(binding.getDefinition()).isInstanceOf(CelAsyncFunctionOverload.class); + assertThat(overload.applyAsync(new Object[] {10L, "test"}).get(5, SECONDS)).isEqualTo("10test"); + } + + @Test + public void fromOverloads_asyncBinding_throwsIllegalArgumentException() { + CelFunctionBinding asyncBinding = + CelFunctionBinding.fromAsync("async_fn", Long.class, (Long arg) -> immediateFuture(arg)); + + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> CelFunctionBinding.fromOverloads("async_fn", asyncBinding)); + + assertThat(e) + .hasMessageThat() + .contains("Asynchronous function overloads cannot be grouped using fromOverloads."); + } + + @Test + public void fromAsync_withIterableArgTypes_success() throws Exception { + CelAsyncFunctionOverload overload = + args -> immediateFuture((Long) args[0] + (Long) args[1] + (Long) args[2]); + + CelFunctionBinding binding = + CelFunctionBinding.fromAsync( + "ternary_async", ImmutableList.of(Long.class, Long.class, Long.class), overload); + + assertThat(binding.getOverloadId()).isEqualTo("ternary_async"); + assertThat(binding.getArgTypes()).containsExactly(Long.class, Long.class, Long.class).inOrder(); + assertThat(binding.isStrict()).isTrue(); + assertThat(binding.getDefinition()).isSameInstanceAs(overload); + CelAsyncFunctionOverload bindingDef = (CelAsyncFunctionOverload) binding.getDefinition(); + assertThat(bindingDef.applyAsync(new Object[] {10L, 20L, 30L}).get(5, SECONDS)).isEqualTo(60L); + } + + @Test + public void asyncFunctionOverload_defaultMethods_delegatesToVarargsAndThrowsOnSync() + throws Exception { + CelAsyncFunctionOverload overload = + args -> { + long sum = 0L; + for (Object arg : args) { + sum += (Long) arg; + } + return immediateFuture(sum); + }; + + assertThat(overload.applyAsync(42L).get(5, SECONDS)).isEqualTo(42L); + assertThat(overload.applyAsync(10L, 20L).get(5, SECONDS)).isEqualTo(30L); + + UnsupportedOperationException thrown = + assertThrows(UnsupportedOperationException.class, () -> overload.apply(new Object[] {1L})); + assertThat(thrown) + .hasMessageThat() + .contains("Async overload cannot be evaluated synchronously. Use evalAsync instead."); + } + + @Test + public void celFunctionResolver_empty_alwaysReturnsEmpty() throws Exception { + CelFunctionResolver resolver = CelFunctionResolver.EMPTY; + + assertThat(resolver.findOverloadMatchingArgs("fn", new Object[] {1L})).isEmpty(); + assertThat( + resolver.findOverloadMatchingArgs( + "fn", ImmutableList.of("fn_overload"), new Object[] {1L})) + .isEmpty(); + } + + @Test + public void dynamicDispatch_applyVarargs_matchesCorrectOverload() throws Exception { + CelFunctionBinding binding1 = + CelFunctionBinding.from( + "sum_three_longs", + ImmutableList.of(Long.class, Long.class, Long.class), + args -> (Long) args[0] + (Long) args[1] + (Long) args[2]); + CelFunctionBinding binding2 = + CelFunctionBinding.from( + "concat_three_strings", + ImmutableList.of(String.class, String.class, String.class), + args -> (String) args[0] + (String) args[1] + (String) args[2]); + + ImmutableSet overloads = + CelFunctionBinding.fromOverloads("add3", binding1, binding2); + OptimizedFunctionOverload dispatchOverload = + (OptimizedFunctionOverload) + Iterables.find(overloads, b -> b.getOverloadId().equals("add3")).getDefinition(); + + assertThat(dispatchOverload.apply(new Object[] {1L, 2L, 3L})).isEqualTo(6L); + assertThat(dispatchOverload.apply(new Object[] {"a", "b", "c"})).isEqualTo("abc"); + + CelOverloadNotFoundException thrown = + assertThrows( + CelOverloadNotFoundException.class, + () -> dispatchOverload.apply(new Object[] {1L, "b", 3L})); + assertThat(thrown) + .hasMessageThat() + .contains( + "No matching overload for function 'add3'. Overload candidates: sum_three_longs," + + " concat_three_strings"); + } + + @Test + public void fromOverloads_nullOrEmptyFunctionName_throwsIllegalArgumentException() { + CelFunctionBinding binding = CelFunctionBinding.from("fn_1", Long.class, (Long arg) -> 1L); + + assertThrows( + IllegalArgumentException.class, () -> CelFunctionBinding.fromOverloads(null, binding)); + assertThrows( + IllegalArgumentException.class, () -> CelFunctionBinding.fromOverloads("", binding)); + } + + @Test + public void fromOverloads_emptyOverloadBindings_throwsIllegalArgumentException() { + assertThrows( + IllegalArgumentException.class, + () -> CelFunctionBinding.fromOverloads("fn", ImmutableList.of())); + assertThrows(IllegalArgumentException.class, () -> CelFunctionBinding.fromOverloads("fn")); + } + + @Test + public void fromOverloads_varargsAndCollection_success() { + CelFunctionBinding binding1 = + CelFunctionBinding.from("unary_fn", Long.class, (Long arg) -> arg + 1L); + CelFunctionBinding binding2 = + CelFunctionBinding.from("binary_fn", Long.class, Long.class, (Long a, Long b) -> a + b); + + ImmutableSet varargsBindings = + CelFunctionBinding.fromOverloads("poly_fn", binding1, binding2); + ImmutableSet collectionBindings = + CelFunctionBinding.fromOverloads("poly_fn", ImmutableList.of(binding1, binding2)); + + assertThat(varargsBindings).isNotEmpty(); + assertThat(collectionBindings).isNotEmpty(); + } +} diff --git a/runtime/src/test/java/dev/cel/runtime/planner/AsyncCompletionCoordinatorTest.java b/runtime/src/test/java/dev/cel/runtime/planner/AsyncCompletionCoordinatorTest.java new file mode 100644 index 000000000..e2226695f --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/planner/AsyncCompletionCoordinatorTest.java @@ -0,0 +1,649 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev.cel.runtime.planner; + +import static com.google.common.truth.Truth.assertThat; +import static java.util.Objects.requireNonNull; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertThrows; + +import com.google.testing.junit.testparameterinjector.TestParameterInjector; +import dev.cel.runtime.CelAsyncCall; +import dev.cel.runtime.CelAsyncDrainStrategy; +import dev.cel.runtime.CelAsyncEvaluationOptions; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Delayed; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +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 AsyncCompletionCoordinatorTest { + + private static final CelAsyncCall DUMMY_CALL = + new CelAsyncCall() { + @Override + public long callId() { + return 1L; + } + + @Override + public long exprId() { + return 10L; + } + + @Override + public String functionName() { + return "testFn"; + } + + @Override + public String overloadId() { + return "testFn_overload"; + } + }; + + @Test + public void notifyCallCompleted_whenWaitingWithPendingActiveCalls_schedulesDebounceTimer() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(continuationRan.get()).isFalse(); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isTrue(); + assertThat(scheduler.getQueue()).isNotEmpty(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void + notifyCallCompleted_whenWaitingWithDrainAllStrategy_waitsIndefinitelyWithoutSchedulingTimer() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainAll()) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(2); + gate.dispatch(Runnable::run, () -> {}); + gate.dispatch(Runnable::run, () -> {}); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + gate.releasePermit(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(continuationRan.get()).isFalse(); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + assertThat(scheduler.getQueue()).isEmpty(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void notifyCallCompleted_whenDebounceTimerPending_resetsDebounceTimerForSlidingWindow() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(2); + gate.dispatch(Runnable::run, () -> {}); + gate.dispatch(Runnable::run, () -> {}); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + // First completion schedules timer 1 + coordinator.notifyCallCompleted(DUMMY_CALL); + ScheduledFuture firstTimer = (ScheduledFuture) scheduler.getQueue().peek(); + assertThat(firstTimer).isNotNull(); + assertThat(firstTimer.isCancelled()).isFalse(); + + // Second completion arrives while timer 1 is pending: sliding window cancels timer 1 and + // reschedules + coordinator.notifyCallCompleted(DUMMY_CALL); + assertThat(firstTimer.isCancelled()).isTrue(); + assertThat(coordinator.hasScheduledDebounceTimer()).isTrue(); + assertThat(continuationRan.get()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void onDebounceFired_whenWaiting_triggersContinuation() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + ScheduledFuture scheduledTask = + requireNonNull((ScheduledFuture) scheduler.getQueue().peek()); + + ((Runnable) scheduledTask).run(); + + assertThat(continuationRan.get()).isTrue(); + assertThat(coordinator.isWaiting()).isFalse(); + assertThat(coordinator.hasPendingBatch()).isFalse(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + assertThat(coordinator.hasContinuation()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void cancel_cancelsDebounceTimerAndPreventsContinuation() { + AtomicBoolean mayInterruptArg = new AtomicBoolean(true); + ScheduledThreadPoolExecutor scheduler = + new ScheduledThreadPoolExecutor(1) { + @Override + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + ScheduledFuture task = super.schedule(command, delay, unit); + return new CapturingScheduledFuture<>(task, mayInterruptArg); + } + }; + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + + AtomicBoolean continuationRan = new AtomicBoolean(false); + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + ScheduledFuture scheduledTask = (ScheduledFuture) scheduler.getQueue().peek(); + assertThat(scheduledTask).isNotNull(); + assertThat(scheduledTask.isCancelled()).isFalse(); + + coordinator.cancel(); + + assertThat(coordinator.hasPendingBatch()).isFalse(); + assertThat(coordinator.isWaiting()).isFalse(); + assertThat(coordinator.hasContinuation()).isFalse(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + assertThat(scheduledTask.isCancelled()).isTrue(); + assertThat(mayInterruptArg.get()).isFalse(); + assertThat(continuationRan.get()).isFalse(); + + // Executing cancelled scheduled task must not trigger the continuation + ((Runnable) scheduledTask).run(); + assertThat(continuationRan.get()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void staleTimerFromPreviousPass_doesNotTriggerContinuationOnSubsequentPass() { + ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(1); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + // Pass 1: start waiting and schedule a debounce timer + coordinator.notifyCallCompleted(DUMMY_CALL); + AtomicInteger pass1Count = new AtomicInteger(); + coordinator.waitForCompletions(pass1Count::incrementAndGet); + + ScheduledFuture pass1Timer = (ScheduledFuture) scheduler.getQueue().peek(); + assertThat(pass1Timer).isNotNull(); + + // In-flight call completes causing immediate re-evaluation: pass 1 continuation runs + gate.releasePermit(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + assertThat(pass1Count.get()).isEqualTo(1); + assertThat(coordinator.isWaiting()).isFalse(); + + // Pass 2 begins + gate.dispatch(Runnable::run, () -> {}); + AtomicInteger pass2Count = new AtomicInteger(); + coordinator.waitForCompletions(pass2Count::incrementAndGet); + assertThat(coordinator.isWaiting()).isTrue(); + + // Stale timer from Pass 1 executes now + ((Runnable) pass1Timer).run(); + + // Pass 2 must NOT have been triggered by the stale timer + assertThat(pass2Count.get()).isEqualTo(0); + assertThat(coordinator.isWaiting()).isTrue(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void waitForCompletions_whenCoordinatorCancelled_throwsIllegalStateException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + coordinator.cancel(); + + assertThrows(IllegalStateException.class, () -> coordinator.waitForCompletions(() -> {})); + } + + @Test + public void multiThreadedConcurrentCompletions_retainsSingleContinuationDispatch() + throws Exception { + int workerCount = 10; + ExecutorService workers = Executors.newFixedThreadPool(workerCount); + ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainAll()) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(workerCount); + for (int i = 0; i < workerCount; i++) { + gate.dispatch(workers, () -> {}); + } + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, workers); + AtomicInteger continuationDispatches = new AtomicInteger(); + CountDownLatch continuationLatch = new CountDownLatch(1); + CountDownLatch readyLatch = new CountDownLatch(workerCount); + CountDownLatch startLatch = new CountDownLatch(1); + + coordinator.waitForCompletions( + () -> { + continuationDispatches.incrementAndGet(); + continuationLatch.countDown(); + }); + + for (int i = 0; i < workerCount; i++) { + workers.execute( + () -> { + readyLatch.countDown(); + try { + startLatch.await(); + gate.releasePermit(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + } + + readyLatch.await(5, SECONDS); + startLatch.countDown(); + + assertThat(continuationLatch.await(5, SECONDS)).isTrue(); + + workers.shutdown(); + assertThat(workers.awaitTermination(5, SECONDS)).isTrue(); + + assertThat(continuationDispatches.get()).isEqualTo(1); + assertThat(coordinator.isWaiting()).isFalse(); + } finally { + workers.shutdownNow(); + scheduler.shutdownNow(); + } + } + + @Test + public void waitForCompletions_whenAlreadyWaiting_throwsIllegalStateException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + coordinator.waitForCompletions(() -> {}); + + assertThrows(IllegalStateException.class, () -> coordinator.waitForCompletions(() -> {})); + } + + @Test + public void notifyCallCompleted_whenCancelled_ignoresCompletion() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + coordinator.cancel(); + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(coordinator.hasPendingBatch()).isFalse(); + } + + @Test + public void constructorAndMethods_nullArguments_throwsNullPointerException() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + + assertThrows( + NullPointerException.class, + () -> new AsyncCompletionCoordinator(null, gate, Runnable::run)); + assertThrows( + NullPointerException.class, + () -> new AsyncCompletionCoordinator(options, null, Runnable::run)); + assertThrows( + NullPointerException.class, () -> new AsyncCompletionCoordinator(options, gate, null)); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + assertThrows(NullPointerException.class, () -> coordinator.notifyCallCompleted(null)); + assertThrows(NullPointerException.class, () -> coordinator.waitForCompletions(null)); + } + + @Test + public void + waitForCompletions_whenDrainStrategySatisfiedImmediately_dispatchesContinuationWithoutTimer() { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainAll()) + .build(); + AsyncGate gate = new AsyncGate(1); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + // Call completed while not waiting and activeCount is 0 + coordinator.notifyCallCompleted(DUMMY_CALL); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(continuationRan.get()).isTrue(); + assertThat(coordinator.isWaiting()).isFalse(); + assertThat(coordinator.hasPendingBatch()).isFalse(); + assertThat(coordinator.hasScheduledDebounceTimer()).isFalse(); + } + + @Test + public void constructor_initializesCycleIdToZero() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + assertThat(coordinator.cycleId()).isEqualTo(0); + } + + @Test + public void notifyCallCompleted_whenDebounceTimerPending_cancelsExistingTimerWithoutInterrupt() { + AtomicBoolean mayInterruptArg = new AtomicBoolean(true); + ScheduledThreadPoolExecutor scheduler = + new ScheduledThreadPoolExecutor(1) { + @Override + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + ScheduledFuture task = super.schedule(command, delay, unit); + return new CapturingScheduledFuture<>(task, mayInterruptArg); + } + }; + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + coordinator.waitForCompletions(() -> {}); + + // Second completion extends debounce window and cancels previous timer + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(mayInterruptArg.get()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void + scheduleDebounce_whenCoordinatorCancelledConcurrently_cancelsScheduledFutureWithoutInterrupt() { + AtomicBoolean mayInterruptArg = new AtomicBoolean(true); + AtomicBoolean cancelledInsideScheduler = new AtomicBoolean(false); + AsyncCompletionCoordinator[] coordinatorHolder = new AsyncCompletionCoordinator[1]; + + ScheduledThreadPoolExecutor scheduler = + new ScheduledThreadPoolExecutor(1) { + @Override + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + ScheduledFuture task = super.schedule(command, delay, unit); + if (coordinatorHolder[0] != null && !cancelledInsideScheduler.get()) { + cancelledInsideScheduler.set(true); + coordinatorHolder[0].cancel(); + } + return new CapturingScheduledFuture<>(task, mayInterruptArg); + } + }; + + try { + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMinutes(10))) + .setScheduledExecutorService(scheduler) + .build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + coordinatorHolder[0] = coordinator; + + coordinator.notifyCallCompleted(DUMMY_CALL); + coordinator.waitForCompletions(() -> {}); + + ScheduledFuture scheduledTask = (ScheduledFuture) scheduler.getQueue().peek(); + assertThat(scheduledTask).isNotNull(); + assertThat(scheduledTask.isCancelled()).isTrue(); + assertThat(mayInterruptArg.get()).isFalse(); + } finally { + scheduler.shutdownNow(); + } + } + + @Test + public void drainAndReset_incrementsCycleId() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + coordinator.waitForCompletions(() -> {}); + long initialCycleId = coordinator.cycleId(); + + gate.releasePermit(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(coordinator.cycleId()).isGreaterThan(initialCycleId); + } + + @Test + public void drainAndReset_clearsContinuation() { + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + + coordinator.waitForCompletions(() -> {}); + + assertThat(coordinator.hasContinuation()).isTrue(); + + gate.releasePermit(Runnable::run); + coordinator.notifyCallCompleted(DUMMY_CALL); + + assertThat(coordinator.hasContinuation()).isFalse(); + } + + @Test + public void onDebounceFired_whenCycleMismatch_doesNotExecuteContinuation() { + AtomicInteger executedCount = new AtomicInteger(); + Executor rejectingNullExecutor = + task -> { + requireNonNull(task, "task must not be null"); + executedCount.incrementAndGet(); + task.run(); + }; + CelAsyncEvaluationOptions options = CelAsyncEvaluationOptions.builder().build(); + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, rejectingNullExecutor); + + coordinator.waitForCompletions(() -> {}); + + // Trigger debounce callback with a mismatched cycle ID (e.g. from an earlier pass) + coordinator.onDebounceFired(coordinator.cycleId() - 1); + + assertThat(executedCount.get()).isEqualTo(0); + assertThat(coordinator.isWaiting()).isTrue(); + assertThat(coordinator.hasContinuation()).isTrue(); + } + + @Test + public void waitForCompletions_whenNoCallsInFlightAndEmptyBatch_evaluatesDrainStrategyAndInvokesContinuation() { + AsyncGate gate = new AsyncGate(1); + CelAsyncEvaluationOptions options = + CelAsyncEvaluationOptions.builder() + .setDrainStrategy(CelAsyncDrainStrategy.drainReady(Duration.ofMillis(50))) + .build(); + AsyncCompletionCoordinator coordinator = + new AsyncCompletionCoordinator(options, gate, Runnable::run); + AtomicBoolean continuationRan = new AtomicBoolean(false); + + coordinator.waitForCompletions(() -> continuationRan.set(true)); + + assertThat(continuationRan.get()).isTrue(); + assertThat(coordinator.isWaiting()).isFalse(); + } + + private static final class CapturingScheduledFuture implements ScheduledFuture { + private final ScheduledFuture delegate; + private final AtomicBoolean capturedMayInterrupt; + + CapturingScheduledFuture(ScheduledFuture delegate, AtomicBoolean capturedMayInterrupt) { + this.delegate = delegate; + this.capturedMayInterrupt = capturedMayInterrupt; + } + + @Override + public long getDelay(TimeUnit unit) { + return delegate.getDelay(unit); + } + + @Override + public int compareTo(Delayed o) { + return delegate.compareTo(o); + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + capturedMayInterrupt.set(mayInterruptIfRunning); + return delegate.cancel(mayInterruptIfRunning); + } + + @Override + public boolean isCancelled() { + return delegate.isCancelled(); + } + + @Override + public boolean isDone() { + return delegate.isDone(); + } + + @Override + public V get() throws InterruptedException, ExecutionException { + return delegate.get(); + } + + @Override + public V get(long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + return delegate.get(timeout, unit); + } + } +} diff --git a/runtime/src/test/java/dev/cel/runtime/planner/AsyncGateTest.java b/runtime/src/test/java/dev/cel/runtime/planner/AsyncGateTest.java new file mode 100644 index 000000000..3815efb6b --- /dev/null +++ b/runtime/src/test/java/dev/cel/runtime/planner/AsyncGateTest.java @@ -0,0 +1,419 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package dev.cel.runtime.planner; + +import static com.google.common.truth.Truth.assertThat; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.Assert.assertThrows; + +import com.google.common.collect.ForwardingQueue; +import com.google.testing.junit.testparameterinjector.TestParameterInjector; +import java.util.AbstractQueue; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; +import org.junit.runner.RunWith; + +@RunWith(TestParameterInjector.class) +public final class AsyncGateTest { + + @Test + public void unboundedConcurrency_runsImmediatelyAndTracksActive() { + AsyncGate gate = new AsyncGate(0); + AtomicBoolean ran = new AtomicBoolean(false); + + gate.dispatch(Runnable::run, () -> ran.set(true)); + + assertThat(ran.get()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + + gate.releasePermit(Runnable::run); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void boundedConcurrency_queuesAndDrainsOnRelease() { + AsyncGate gate = new AsyncGate(1); + AtomicInteger tasksExecuted = new AtomicInteger(); + + gate.dispatch(Runnable::run, tasksExecuted::incrementAndGet); + + assertThat(tasksExecuted.get()).isEqualTo(1); + assertThat(gate.activeCount()).isEqualTo(1); + + // Second task should be queued in pendingTasks because concurrency limit is 1 + gate.dispatch(Runnable::run, tasksExecuted::incrementAndGet); + + assertThat(tasksExecuted.get()).isEqualTo(1); + + // Releasing permit drains the pending task + gate.releasePermit(Runnable::run); + + assertThat(tasksExecuted.get()).isEqualTo(2); + assertThat(gate.activeCount()).isEqualTo(1); + + gate.releasePermit(Runnable::run); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void cancel_clearsPendingTasksAndIgnoresNewTasks() { + AsyncGate gate = new AsyncGate(1); + AtomicInteger tasksExecuted = new AtomicInteger(); + + gate.dispatch(Runnable::run, tasksExecuted::incrementAndGet); + + assertThat(tasksExecuted.get()).isEqualTo(1); + + gate.dispatch(Runnable::run, tasksExecuted::incrementAndGet); + + assertThat(tasksExecuted.get()).isEqualTo(1); + + gate.cancel(); + + // Releasing permit should not run the queued task because it was cleared + gate.releasePermit(Runnable::run); + + assertThat(tasksExecuted.get()).isEqualTo(1); + + // New dispatch after cancel is ignored + gate.dispatch(Runnable::run, tasksExecuted::incrementAndGet); + + assertThat(tasksExecuted.get()).isEqualTo(1); + } + + @Test + public void cancel_withQueuedPendingTasks_clearsPendingTasksImmediately() { + ConcurrentLinkedQueue pendingTasks = new ConcurrentLinkedQueue<>(); + AsyncGate gate = new AsyncGate(1, pendingTasks); + + gate.dispatch(Runnable::run, () -> {}); + gate.dispatch(Runnable::run, () -> {}); + + assertThat(pendingTasks).hasSize(1); + + gate.cancel(); + + assertThat(pendingTasks).isEmpty(); + } + + @Test + public void drainPending_whenCancelled_clearsPendingTasksAndDoesNotRun() { + ConcurrentLinkedQueue pendingTasks = new ConcurrentLinkedQueue<>(); + AsyncGate gate = new AsyncGate(1, pendingTasks); + AtomicBoolean taskRan = new AtomicBoolean(false); + + gate.dispatch(Runnable::run, () -> {}); + gate.cancel(); + + pendingTasks.add(() -> taskRan.set(true)); + + gate.releasePermit(Runnable::run); + + assertThat(taskRan.get()).isFalse(); + assertThat(pendingTasks).isEmpty(); + } + + @Test + public void dispatch_whenPermitAvailableAfterQueueing_drainPendingRunsTask() { + AtomicBoolean taskRan = new AtomicBoolean(false); + Semaphore semaphore = new Semaphore(1); + ConcurrentLinkedQueue delegate = new ConcurrentLinkedQueue<>(); + Queue queue = + new ForwardingQueue() { + @Override + protected Queue delegate() { + return delegate; + } + + @Override + public boolean add(Runnable r) { + boolean res = super.add(r); + // Release permit directly on semaphore without calling drainPending. + // This verifies that dispatch drains the queue if a permit became available after + // queueing. + semaphore.release(); + return res; + } + }; + AsyncGate gate = new AsyncGate(semaphore, queue); + + gate.dispatch(Runnable::run, () -> {}); + gate.dispatch(Runnable::run, () -> taskRan.set(true)); + + assertThat(taskRan.get()).isTrue(); + } + + @Test + public void concurrentDrainAndCancel_retainsPermitBalance() throws Exception { + ExecutorService executor = Executors.newFixedThreadPool(4); + try { + for (int i = 0; i < 50; i++) { + AsyncGate gate = new AsyncGate(1); + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(2); + + gate.dispatch(executor, () -> {}); + gate.dispatch(executor, () -> {}); + + executor.execute( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + gate.releasePermit(executor); + doneLatch.countDown(); + } + }); + + executor.execute( + () -> { + try { + startLatch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + gate.cancel(); + doneLatch.countDown(); + } + }); + + startLatch.countDown(); + assertThat(doneLatch.await(5, SECONDS)).isTrue(); + assertThat(gate.activeCount()).isAtLeast(0); + } + } finally { + executor.shutdown(); + executor.awaitTermination(5, SECONDS); + } + } + + @Test + public void drainPending_whenTaskPolledIsNull_releasesPermitAndBreaks() { + Semaphore semaphore = new Semaphore(1); + AtomicInteger pollCount = new AtomicInteger(); + Queue queue = + new AbstractQueue() { + @Override + public boolean offer(Runnable e) { + return true; + } + + @Override + public Runnable peek() { + return null; + } + + @Override + public Iterator iterator() { + return Collections.emptyIterator(); + } + + @Override + public int size() { + return 1; + } + + @Override + public boolean isEmpty() { + // Always returns false: if break; is missing, loop will spin and poll again + return false; + } + + @Override + public Runnable poll() { + pollCount.incrementAndGet(); + return null; + } + }; + AsyncGate gate = new AsyncGate(semaphore, queue); + + gate.releasePermit(Runnable::run); + + // Verify loop terminates immediately on null task without polling again + assertThat(pollCount.get()).isEqualTo(1); + // Verify acquired permit is released back to semaphore when queue poll returns null + assertThat(semaphore.availablePermits()).isEqualTo(2); + } + + @Test + public void dispatch_taskThrowsRuntimeException_releasesPermitAndDecrementsActiveCount() { + AsyncGate gate = new AsyncGate(1); + assertThrows( + RuntimeException.class, + () -> + gate.dispatch( + Runnable::run, + () -> { + throw new RuntimeException("fail"); + })); + + assertThat(gate.activeCount()).isEqualTo(0); + // Next task should be able to acquire permit immediately + AtomicBoolean secondRan = new AtomicBoolean(false); + gate.dispatch(Runnable::run, () -> secondRan.set(true)); + assertThat(secondRan.get()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + } + + @Test + public void dispatch_unboundedConcurrency_taskThrows_decrementsActiveCountAndRethrows() { + AsyncGate gate = new AsyncGate(0); + + assertThrows( + RuntimeException.class, + () -> + gate.dispatch( + Runnable::run, + () -> { + throw new RuntimeException("fail"); + })); + + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void drainPending_executorThrows_releasesPermitDecrementsActiveCountAndRethrows() { + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + gate.dispatch(Runnable::run, () -> {}); + + Executor rejectingExecutor = + r -> { + throw new RejectedExecutionException("rejected"); + }; + + assertThrows(RejectedExecutionException.class, () -> gate.releasePermit(rejectingExecutor)); + + assertThat(gate.activeCount()).isEqualTo(0); + AtomicBoolean secondRan = new AtomicBoolean(false); + gate.dispatch(Runnable::run, () -> secondRan.set(true)); + assertThat(secondRan.get()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + } + + @Test + public void dispatch_nullArguments_throwsNullPointerException() { + AsyncGate gate = new AsyncGate(1); + + assertThrows(NullPointerException.class, () -> gate.dispatch(null, () -> {})); + assertThrows(NullPointerException.class, () -> gate.dispatch(Runnable::run, null)); + assertThrows(NullPointerException.class, () -> gate.releasePermit(null)); + } + + @Test + public void negativeConcurrency_treatedAsUnbounded() { + AsyncGate gate = new AsyncGate(-1); + AtomicBoolean ran = new AtomicBoolean(false); + + gate.dispatch(Runnable::run, () -> ran.set(true)); + + assertThat(ran.get()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + + gate.releasePermit(Runnable::run); + assertThat(gate.activeCount()).isEqualTo(0); + } + + @Test + public void releasePermit_boundedConcurrency_releasesSemaphorePermit() { + AsyncGate gate = new AsyncGate(1); + gate.dispatch(Runnable::run, () -> {}); + + gate.releasePermit(Runnable::run); + + assertThat(gate.activeCount()).isEqualTo(0); + + // Verifies semaphore permit was actually restored so a new task can be immediately admitted + AtomicBoolean secondRan = new AtomicBoolean(false); + gate.dispatch(Runnable::run, () -> secondRan.set(true)); + + assertThat(secondRan.get()).isTrue(); + assertThat(gate.activeCount()).isEqualTo(1); + } + + @Test + public void drainPending_taskThrowsInWorkerThread_decrementsActiveCountAndReleasesPermit() { + AsyncGate gate = new AsyncGate(1); + // Occupy the only permit + gate.dispatch(Runnable::run, () -> {}); + + // Enqueue a task that will fail when drained + gate.dispatch( + Runnable::run, + () -> { + throw new RuntimeException("worker thread boom"); + }); + + // Release first permit, which triggers drainPending on the queued failing task + assertThrows(RuntimeException.class, () -> gate.releasePermit(Runnable::run)); + + // Active count must be decremented back to 0 despite the exception + assertThat(gate.activeCount()).isEqualTo(0); + + // Permit must be released so subsequent tasks can execute + AtomicBoolean subsequentRan = new AtomicBoolean(false); + gate.dispatch(Runnable::run, () -> subsequentRan.set(true)); + assertThat(subsequentRan.get()).isTrue(); + } + + @Test + public void dispatch_whenTasksPending_enqueuesWithoutQueueBarging() { + Semaphore semaphore = new Semaphore(1); + Queue pendingTasks = new ConcurrentLinkedQueue<>(); + AsyncGate gate = new AsyncGate(semaphore, pendingTasks); + + // Saturate permit + gate.dispatch(Runnable::run, () -> {}); + + List order = new ArrayList<>(); + // Enqueue first task using a paused executor + List queued = new ArrayList<>(); + gate.dispatch(queued::add, () -> order.add("first")); + + // Release permit directly on semaphore without draining (simulated concurrent permit release) + // Now semaphore has a permit available, but pendingTasks has "first" + semaphore.release(); + + // An incoming dispatch must NOT steal the permit ahead of "first" + gate.dispatch(queued::add, () -> order.add("second")); + + // "first" was drained by the incoming dispatch into queued; run it and release permit + assertThat(queued).hasSize(1); + queued.remove(0).run(); + gate.releasePermit(queued::add); + + // "second" was drained by releasePermit into queued; run it + assertThat(queued).hasSize(1); + queued.remove(0).run(); + + assertThat(order).containsExactly("first", "second").inOrder(); + } +} diff --git a/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel b/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel index 9116818dc..38d1d0d70 100644 --- a/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel +++ b/runtime/src/test/java/dev/cel/runtime/planner/BUILD.bazel @@ -40,6 +40,9 @@ java_library( "//extensions", "//parser:macro", "//runtime", + "//runtime:async_call", + "//runtime:async_drain_strategy", + "//runtime:async_options", "//runtime:descriptor_type_resolver", "//runtime:dispatcher", "//runtime:function_binding", @@ -49,6 +52,8 @@ java_library( "//runtime:runtime_helpers", "//runtime:standard_functions", "//runtime:unknown_attributes", + "//runtime/planner:async_completion_coordinator", + "//runtime/planner:async_gate", "//runtime/planner:program_planner", "//runtime/standard:type", "@cel_spec//proto/cel/expr/conformance/proto3:test_all_types_java_proto",