Skip to content

Commit 52acaa7

Browse files
l46kokcopybara-github
authored andcommitted
Add AsyncCallRecord primitive for async eval
PiperOrigin-RevId: 982639852
1 parent ef2f7aa commit 52acaa7

7 files changed

Lines changed: 963 additions & 5 deletions

File tree

runtime/planner/BUILD.bazel

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,3 +35,9 @@ java_library(
3535
visibility = ["//:internal"],
3636
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator"],
3737
)
38+
39+
java_library(
40+
name = "async_call_state_tracker",
41+
visibility = ["//:internal"],
42+
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_call_state_tracker"],
43+
)

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -237,7 +237,9 @@ public int hashCode(Object object) {
237237

238238
object = runtimeHelpers.adaptValue(object);
239239
if (object instanceof Number) {
240-
return Double.hashCode(((Number) object).doubleValue());
240+
double value = ((Number) object).doubleValue();
241+
// Normalize -0.0 to 0.0. objectEquals reports the two as equal, so they must hash alike.
242+
return Double.hashCode(value == 0.0d ? 0.0d : value);
241243
}
242244
if (object instanceof Iterable) {
243245
int h = 1;
Lines changed: 301 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,301 @@
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.planner;
16+
17+
import static com.google.common.base.Preconditions.checkNotNull;
18+
19+
import com.google.common.util.concurrent.ListenableFuture;
20+
import javax.annotation.concurrent.ThreadSafe;
21+
import dev.cel.runtime.CelAsyncCall;
22+
import dev.cel.runtime.CelAsyncFunctionOverload;
23+
import dev.cel.runtime.RuntimeEquality;
24+
import java.util.Optional;
25+
import java.util.concurrent.atomic.AtomicBoolean;
26+
import org.jspecify.annotations.Nullable;
27+
28+
/** Tracks the execution state and result of a single asynchronous function call. */
29+
@ThreadSafe
30+
// CEL-Internal-4
31+
final class AsyncCallRecord implements CelAsyncCall {
32+
33+
enum State {
34+
NOT_STARTED,
35+
RUNNING,
36+
SUCCESS,
37+
FAILURE,
38+
CANCELLED
39+
}
40+
41+
// Type markers keep values of different kinds from colliding in the bucket hash, e.g. the
42+
// string "NaN" and the double NaN. Collisions remain harmless because matches() disambiguates the
43+
// bucket.
44+
private static final int STRING_HASH_MARKER = 's';
45+
private static final int BOOL_HASH_MARKER = 'b';
46+
private static final int NUMBER_HASH_MARKER = 'n';
47+
private static final int COMPLEX_HASH_MARKER = 'x';
48+
49+
private final long callId;
50+
private final long exprId;
51+
private final String functionName;
52+
private final String overloadId;
53+
54+
@SuppressWarnings("Immutable") // Array not mutated after construction
55+
private final Object[] args;
56+
57+
private final CelAsyncFunctionOverload overload;
58+
59+
private final Object lock = new Object();
60+
private final AtomicBoolean completionReported = new AtomicBoolean(false);
61+
private volatile State state = State.NOT_STARTED;
62+
private volatile @Nullable Object result;
63+
private volatile @Nullable Throwable error;
64+
private volatile @Nullable ListenableFuture<?> inFlightFuture;
65+
66+
static AsyncCallRecord create(
67+
long callId,
68+
long exprId,
69+
String functionName,
70+
String overloadId,
71+
Object[] args,
72+
CelAsyncFunctionOverload overload) {
73+
return new AsyncCallRecord(callId, exprId, functionName, overloadId, args, overload);
74+
}
75+
76+
/**
77+
* Computes the bucket hash under which a call is tracked.
78+
*
79+
* <p>This is a bucketing hint, not an identity: calls that {@link #matches} considers identical
80+
* hash alike, but distinct calls may share a bucket. Resolve the exact call via {@link #matches}.
81+
*/
82+
static int hashCall(long exprId, String overloadId, Object[] args) {
83+
int result = 31 * Long.hashCode(exprId) + overloadId.hashCode();
84+
for (Object arg : args) {
85+
result = result * 31 + hashArg(arg);
86+
}
87+
return result;
88+
}
89+
90+
/**
91+
* Returns whether this record tracks a call to the same expression node, function, overload, and
92+
* arguments.
93+
*
94+
* <p>Arguments are compared under CEL equality, except that NaN compares equal to itself so that
95+
* a node re-evaluated with a NaN argument can find its existing record.
96+
*/
97+
boolean matches(
98+
long exprId,
99+
String functionName,
100+
String overloadId,
101+
Object[] args,
102+
RuntimeEquality runtimeEquality) {
103+
if (this.exprId != exprId
104+
|| !this.functionName.equals(functionName)
105+
|| !this.overloadId.equals(overloadId)
106+
|| this.args.length != args.length) {
107+
return false;
108+
}
109+
for (int i = 0; i < this.args.length; i++) {
110+
Object arg = this.args[i];
111+
Object otherArg = args[i];
112+
if (!celEquals(runtimeEquality, arg, otherArg) && !(isNan(arg) && isNan(otherArg))) {
113+
return false;
114+
}
115+
}
116+
return true;
117+
}
118+
119+
private static boolean celEquals(
120+
RuntimeEquality runtimeEquality, @Nullable Object a, @Nullable Object b) {
121+
try {
122+
return runtimeEquality.objectEquals(a, b);
123+
} catch (UnsupportedOperationException e) {
124+
// Proto equality is unimplemented on the lite runtime (b/381937349). Degrade to a non-match
125+
// rather than failing the call, mirroring cel-go where types.Equal yields an error value.
126+
// Any other exception indicates a genuine defect and is left to propagate.
127+
return false;
128+
}
129+
}
130+
131+
@Override
132+
public long callId() {
133+
return callId;
134+
}
135+
136+
@Override
137+
public long exprId() {
138+
return exprId;
139+
}
140+
141+
@Override
142+
public String functionName() {
143+
return functionName;
144+
}
145+
146+
@Override
147+
public String overloadId() {
148+
return overloadId;
149+
}
150+
151+
/**
152+
* Transitions the call state from {@link State#NOT_STARTED} to {@link State#RUNNING}.
153+
*
154+
* @return true if the transition succeeded, false if the call was already running, completed, or
155+
* cancelled.
156+
*/
157+
boolean markRunning() {
158+
synchronized (lock) {
159+
if (state != State.NOT_STARTED) {
160+
return false;
161+
}
162+
state = State.RUNNING;
163+
return true;
164+
}
165+
}
166+
167+
void setInFlightFuture(ListenableFuture<?> future) {
168+
checkNotNull(future);
169+
boolean shouldCancel;
170+
synchronized (lock) {
171+
inFlightFuture = future;
172+
shouldCancel = (state == State.CANCELLED && !future.isDone());
173+
}
174+
if (shouldCancel) {
175+
future.cancel(/* mayInterruptIfRunning= */ false);
176+
}
177+
}
178+
179+
boolean cancelInFlight() {
180+
ListenableFuture<?> futureToCancel = null;
181+
synchronized (lock) {
182+
if (!isPending()) {
183+
return false;
184+
}
185+
state = State.CANCELLED;
186+
ListenableFuture<?> future = inFlightFuture;
187+
if (future != null && !future.isDone()) {
188+
futureToCancel = future;
189+
}
190+
}
191+
if (futureToCancel != null) {
192+
futureToCancel.cancel(/* mayInterruptIfRunning= */ false);
193+
}
194+
return true;
195+
}
196+
197+
/**
198+
* Claims the right to report this call's completion, returning true for the first caller only.
199+
*
200+
* <p>Tracked separately from {@link State} because a call cancelled after dispatch still holds a
201+
* concurrency permit and must release it exactly once.
202+
*/
203+
boolean markCompletionReported() {
204+
return completionReported.compareAndSet(false, true);
205+
}
206+
207+
boolean isCancelled() {
208+
return state == State.CANCELLED;
209+
}
210+
211+
boolean complete(@Nullable Object result) {
212+
synchronized (lock) {
213+
if (!isPending()) {
214+
return false;
215+
}
216+
this.result = result;
217+
state = State.SUCCESS;
218+
return true;
219+
}
220+
}
221+
222+
boolean fail(Throwable error) {
223+
checkNotNull(error);
224+
synchronized (lock) {
225+
if (!isPending()) {
226+
return false;
227+
}
228+
this.error = error;
229+
state = State.FAILURE;
230+
return true;
231+
}
232+
}
233+
234+
Object[] args() {
235+
return args.clone();
236+
}
237+
238+
CelAsyncFunctionOverload overload() {
239+
return overload;
240+
}
241+
242+
State state() {
243+
return state;
244+
}
245+
246+
/**
247+
* Returns the completed result, if present.
248+
*
249+
* <p>Note: If a call completed successfully with a {@code null} value, this method returns {@code
250+
* Optional.empty()}. Callers should check {@link #state()} to distinguish between a call that has
251+
* not completed and one that succeeded with {@code null}.
252+
*/
253+
Optional<Object> result() {
254+
return Optional.ofNullable(result);
255+
}
256+
257+
Optional<Throwable> error() {
258+
return Optional.ofNullable(error);
259+
}
260+
261+
private static int hashArg(@Nullable Object arg) {
262+
if (arg instanceof String) {
263+
return STRING_HASH_MARKER * 31 + arg.hashCode();
264+
}
265+
if (arg instanceof Boolean) {
266+
return BOOL_HASH_MARKER * 31 + arg.hashCode();
267+
}
268+
if (arg instanceof Number) {
269+
// Hash int, uint, and double through a common double representation so that values CEL
270+
// considers equal (1 == 1u == 1.0) share a bucket. NaN needs no special case because
271+
// Double.hashCode(NaN) is a constant across all double and float NaN representations.
272+
double value = ((Number) arg).doubleValue();
273+
// Normalize -0.0 to 0.0, which CEL considers equal to 0.0.
274+
return NUMBER_HASH_MARKER * 31 + Double.hashCode(value == 0.0d ? 0.0d : value);
275+
}
276+
return COMPLEX_HASH_MARKER;
277+
}
278+
279+
private static boolean isNan(@Nullable Object value) {
280+
return value instanceof Number && Double.isNaN(((Number) value).doubleValue());
281+
}
282+
283+
private boolean isPending() {
284+
return state == State.NOT_STARTED || state == State.RUNNING;
285+
}
286+
287+
private AsyncCallRecord(
288+
long callId,
289+
long exprId,
290+
String functionName,
291+
String overloadId,
292+
Object[] args,
293+
CelAsyncFunctionOverload overload) {
294+
this.callId = callId;
295+
this.exprId = exprId;
296+
this.functionName = checkNotNull(functionName);
297+
this.overloadId = checkNotNull(overloadId);
298+
this.args = checkNotNull(args).clone();
299+
this.overload = checkNotNull(overload);
300+
}
301+
}

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -217,6 +217,22 @@ java_library(
217217
],
218218
)
219219

220+
java_library(
221+
name = "async_call_state_tracker",
222+
srcs = ["AsyncCallRecord.java"],
223+
tags = [
224+
],
225+
deps = [
226+
"//runtime:async_call",
227+
"//runtime:function_overload",
228+
"//runtime:runtime_equality",
229+
"@maven//:com_google_code_findbugs_annotations",
230+
"@maven//:com_google_errorprone_error_prone_annotations",
231+
"@maven//:com_google_guava_guava",
232+
"@maven//:org_jspecify_jspecify",
233+
],
234+
)
235+
220236
java_library(
221237
name = "activation_wrapper",
222238
srcs = ["ActivationWrapper.java"],

runtime/src/test/java/dev/cel/runtime/RuntimeEqualityTest.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ public void objectEquals_and_hashCode() {
3737
assertEqualityAndHashCode(runtimeEquality, 2, 2L);
3838
assertEqualityAndHashCode(runtimeEquality, 3, 3.0);
3939
assertEqualityAndHashCode(runtimeEquality, 4, UnsignedLong.valueOf(4));
40+
assertEqualityAndHashCode(runtimeEquality, 0.0d, -0.0d);
4041
assertEqualityAndHashCode(
4142
runtimeEquality,
4243
ImmutableList.of(1, 2, 3),

0 commit comments

Comments
 (0)