Skip to content

Commit 045e576

Browse files
l46kokcopybara-github
authored andcommitted
Add plumbing for executor, async evaluation option. Define program API contracts
PiperOrigin-RevId: 979522474
1 parent 6a8a056 commit 045e576

14 files changed

Lines changed: 581 additions & 6 deletions

publish/BUILD.bazel

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ RUNTIME_TARGETS = [
3232
"//runtime/src/main/java/dev/cel/runtime:async_call",
3333
"//runtime/src/main/java/dev/cel/runtime:async_drain_strategy",
3434
"//runtime/src/main/java/dev/cel/runtime:async_observer",
35+
"//runtime/src/main/java/dev/cel/runtime:async_options",
3536
"//runtime/src/main/java/dev/cel/runtime:base",
3637
"//runtime/src/main/java/dev/cel/runtime:interpreter",
3738
"//runtime/src/main/java/dev/cel/runtime:late_function_binding",

runtime/BUILD.bazel

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ java_library(
1212
":async_call",
1313
":async_drain_strategy",
1414
":async_observer",
15+
":async_options",
1516
":descriptor_message_provider",
1617
":evaluation_exception",
1718
":function_overload",
@@ -417,3 +418,13 @@ cel_android_library(
417418
name = "async_observer_android",
418419
exports = ["//runtime/src/main/java/dev/cel/runtime:async_observer_android"],
419420
)
421+
422+
java_library(
423+
name = "async_options",
424+
exports = ["//runtime/src/main/java/dev/cel/runtime:async_options"],
425+
)
426+
427+
cel_android_library(
428+
name = "async_options_android",
429+
exports = ["//runtime/src/main/java/dev/cel/runtime:async_options_android"],
430+
)

runtime/src/main/java/dev/cel/runtime/BUILD.bazel

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -821,6 +821,7 @@ java_library(
821821
tags = [
822822
],
823823
deps = [
824+
":async_options",
824825
":descriptor_type_resolver",
825826
":dispatcher",
826827
":evaluation_exception",
@@ -871,6 +872,7 @@ java_library(
871872
tags = [
872873
],
873874
deps = [
875+
":async_options",
874876
":descriptor_message_provider",
875877
":descriptor_type_resolver",
876878
":dispatcher",
@@ -926,6 +928,7 @@ java_library(
926928
],
927929
deps = [
928930
":activation",
931+
":async_options",
929932
":evaluation_exception",
930933
":evaluation_listener",
931934
":function_binding",
@@ -950,6 +953,7 @@ java_library(
950953
"@maven//:com_google_errorprone_error_prone_annotations",
951954
"@maven//:com_google_guava_guava",
952955
"@maven//:com_google_protobuf_protobuf_java",
956+
"@maven//:org_jspecify_jspecify",
953957
],
954958
)
955959

@@ -1363,6 +1367,36 @@ cel_android_library(
13631367
],
13641368
)
13651369

1370+
java_library(
1371+
name = "async_options",
1372+
srcs = ["CelAsyncEvaluationOptions.java"],
1373+
tags = [
1374+
],
1375+
deps = [
1376+
":async_drain_strategy",
1377+
":async_observer",
1378+
"//:auto_value",
1379+
"@maven//:com_google_code_findbugs_annotations",
1380+
"@maven//:com_google_errorprone_error_prone_annotations",
1381+
"@maven//:org_jspecify_jspecify",
1382+
],
1383+
)
1384+
1385+
cel_android_library(
1386+
name = "async_options_android",
1387+
srcs = ["CelAsyncEvaluationOptions.java"],
1388+
tags = [
1389+
],
1390+
deps = [
1391+
":async_drain_strategy_android",
1392+
":async_observer_android",
1393+
"//:auto_value",
1394+
"@maven//:com_google_code_findbugs_annotations",
1395+
"@maven//:com_google_errorprone_error_prone_annotations",
1396+
"@maven//:org_jspecify_jspecify",
1397+
],
1398+
)
1399+
13661400
java_library(
13671401
name = "program",
13681402
srcs = ["Program.java"],
@@ -1374,6 +1408,7 @@ java_library(
13741408
":partial_vars",
13751409
":variable_resolver",
13761410
"@maven//:com_google_errorprone_error_prone_annotations",
1411+
"@maven//:com_google_guava_guava",
13771412
],
13781413
)
13791414

@@ -1388,6 +1423,7 @@ cel_android_library(
13881423
":partial_vars_android",
13891424
":variable_resolver",
13901425
"@maven//:com_google_errorprone_error_prone_annotations",
1426+
"@maven_android//:com_google_guava_guava",
13911427
],
13921428
)
13931429

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
// Copyright 2026 Google LLC
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// https://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package dev.cel.runtime;
16+
17+
import com.google.auto.value.AutoValue;
18+
import javax.annotation.concurrent.ThreadSafe;
19+
import java.util.Optional;
20+
import java.util.concurrent.Executors;
21+
import java.util.concurrent.ScheduledExecutorService;
22+
import java.util.concurrent.atomic.AtomicLong;
23+
24+
/** Options for configuring asynchronous CEL evaluation. */
25+
@AutoValue
26+
@ThreadSafe
27+
public abstract class CelAsyncEvaluationOptions {
28+
29+
private static final int DEFAULT_MAX_CONCURRENCY = 100;
30+
private static final int DEFAULT_MAX_ITERATIONS = 1_000;
31+
32+
/**
33+
* Maximum number of concurrent async function calls in-flight simultaneously. A value <= 0
34+
* indicates unbounded concurrency.
35+
*/
36+
public abstract int maxConcurrency();
37+
38+
/** Strategy governing when to trigger re-evaluation after async call completions. */
39+
public abstract CelAsyncDrainStrategy drainStrategy();
40+
41+
/** Safety cap on the maximum number of AST re-evaluation passes before aborting. */
42+
public abstract int maxIterations();
43+
44+
/**
45+
* Returns the custom configured {@link ScheduledExecutorService}, if present.
46+
*
47+
* <p>If absent, {@link #resolveScheduledExecutorService()} falls back to an internal, shared
48+
* single-threaded daemon scheduler.
49+
*/
50+
public abstract Optional<ScheduledExecutorService> scheduledExecutorService();
51+
52+
/** Returns the configured lifecycle observer, if present. */
53+
public abstract Optional<CelAsyncObserver> observer();
54+
55+
/**
56+
* Resolves the {@link ScheduledExecutorService} used for debounce timers, falling back to a
57+
* shared, lazily initialized single-threaded daemon scheduler (named {@code
58+
* cel-async-debounce-*}) if not custom-configured.
59+
*
60+
* <p>The scheduler is used exclusively as an alarm clock to trigger continuation wakeups; it does
61+
* not execute CEL evaluation tasks.
62+
*/
63+
public ScheduledExecutorService resolveScheduledExecutorService() {
64+
return scheduledExecutorService().orElse(DefaultDebounceSchedulerHolder.INSTANCE);
65+
}
66+
67+
public abstract Builder toBuilder();
68+
69+
/**
70+
* Returns a new {@link Builder} initialized with standard default options:
71+
*
72+
* <ul>
73+
* <li>Maximum concurrency: 100 in-flight calls
74+
* <li>Maximum iterations: 1,000 evaluation passes
75+
* <li>Drain strategy: {@link CelAsyncDrainStrategy#drainReady()} (100-microsecond debounce
76+
* window)
77+
* <li>Scheduled executor service: A shared, lazily initialized single-threaded daemon scheduler
78+
* used exclusively for debounce timer wakeups.
79+
* </ul>
80+
*/
81+
public static Builder newBuilder() {
82+
return new AutoValue_CelAsyncEvaluationOptions.Builder()
83+
.setMaxConcurrency(DEFAULT_MAX_CONCURRENCY)
84+
.setDrainStrategy(CelAsyncDrainStrategy.drainReady())
85+
.setMaxIterations(DEFAULT_MAX_ITERATIONS);
86+
}
87+
88+
/**
89+
* Returns a new {@link Builder} initialized with standard default options.
90+
*
91+
* <p>Equivalent to calling {@link #newBuilder()}.
92+
*/
93+
public static Builder builder() {
94+
return newBuilder();
95+
}
96+
97+
/**
98+
* Returns a {@link CelAsyncEvaluationOptions} instance with the {@link #newBuilder() default
99+
* configuration}.
100+
*/
101+
public static CelAsyncEvaluationOptions defaultOptions() {
102+
return newBuilder().build();
103+
}
104+
105+
private static final class DefaultDebounceSchedulerHolder {
106+
private static final AtomicLong counter = new AtomicLong();
107+
private static final ScheduledExecutorService INSTANCE =
108+
Executors.newSingleThreadScheduledExecutor(
109+
r -> {
110+
Thread t = new Thread(r);
111+
t.setName("cel-async-debounce-" + counter.getAndIncrement());
112+
t.setDaemon(true);
113+
return t;
114+
});
115+
}
116+
117+
/** Builder for {@link CelAsyncEvaluationOptions}. */
118+
@AutoValue.Builder
119+
public abstract static class Builder {
120+
public abstract Builder setMaxConcurrency(int maxConcurrency);
121+
122+
public abstract Builder setDrainStrategy(CelAsyncDrainStrategy drainStrategy);
123+
124+
public abstract Builder setMaxIterations(int maxIterations);
125+
126+
/**
127+
* Sets a custom {@link ScheduledExecutorService} for debounce timers.
128+
*
129+
* <p>If not set, defaults to an internal, shared single-threaded daemon scheduler.
130+
*/
131+
public abstract Builder setScheduledExecutorService(
132+
ScheduledExecutorService scheduledExecutorService);
133+
134+
public abstract Builder setObserver(CelAsyncObserver observer);
135+
136+
public abstract CelAsyncEvaluationOptions build();
137+
}
138+
139+
// Package-private constructor prevents extension outside package while allowing AutoValue.
140+
CelAsyncEvaluationOptions() {}
141+
}

runtime/src/main/java/dev/cel/runtime/CelRuntime.java

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414

1515
package dev.cel.runtime;
1616

17+
import com.google.common.util.concurrent.ListenableFuture;
1718
import com.google.errorprone.annotations.CanIgnoreReturnValue;
1819
import com.google.errorprone.annotations.Immutable;
1920
import javax.annotation.concurrent.ThreadSafe;
@@ -42,6 +43,12 @@ interface Program extends dev.cel.runtime.Program {
4243
/** Evaluate the expression using {@code message} fields as the source of input variables. */
4344
Object eval(Message message) throws CelEvaluationException;
4445

46+
/**
47+
* Evaluate the expression asynchronously using {@code message} fields as the source of input
48+
* variables.
49+
*/
50+
ListenableFuture<Object> evalAsync(Message message);
51+
4552
/**
4653
* Trace evaluates a compiled program without any variables and invokes the listener as
4754
* evaluation progresses through the AST.

runtime/src/main/java/dev/cel/runtime/CelRuntimeBuilder.java

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414

1515
package dev.cel.runtime;
1616

17+
import com.google.common.util.concurrent.ListeningExecutorService;
1718
import com.google.errorprone.annotations.CanIgnoreReturnValue;
1819
import com.google.errorprone.annotations.CheckReturnValue;
1920
import com.google.protobuf.DescriptorProtos.FileDescriptorSet;
@@ -214,6 +215,22 @@ public interface CelRuntimeBuilder {
214215
@CanIgnoreReturnValue
215216
CelRuntimeBuilder setContainer(CelContainer container);
216217

218+
/**
219+
* Sets options to use for asynchronous evaluation.
220+
*
221+
* <p>If not configured, defaults to {@link CelAsyncEvaluationOptions#defaultOptions()}.
222+
*/
223+
@CanIgnoreReturnValue
224+
CelRuntimeBuilder setAsyncEvaluationOptions(CelAsyncEvaluationOptions asyncEvaluationOptions);
225+
226+
/**
227+
* Sets the executor to use for asynchronous evaluation.
228+
*
229+
* <p>This executor is required when evaluating expressions asynchronously via {@link
230+
* Program#evalAsync}.
231+
*/
232+
@CanIgnoreReturnValue
233+
CelRuntimeBuilder setAsyncExecutor(ListeningExecutorService asyncExecutor);
217234

218235
/** Build a new instance of the {@code CelRuntime}. */
219236
@CheckReturnValue

0 commit comments

Comments
 (0)