Skip to content

Add async execution engine and AsyncDriver #1240

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
copybara-service wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add async execution engine and AsyncDriver #1240

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions common/src/main/java/dev/cel/common/CelOverloadDecl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,8 @@
* <p>An overload is declared in either a global function `Ex: f(x, ...)` or a method call style
* `Ex: x.f(...)`.
*/
// TODO: Port https://github.com/cel-expr/cel-go/pull/1483 to support first-class
// properties on declarations and bindings for late-bound and asynchronous functions.
@AutoValue
@Immutable
public abstract class CelOverloadDecl {
Expand Down
26 changes: 26 additions & 0 deletions runtime/planner/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,22 +22,48 @@ java_library(
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:planned_program"],
)

cel_android_library(
name = "planned_program_android",
visibility = ["//:internal"],
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:planned_program_android"],
)

java_library(
name = "async_gate",
testonly = 1,
visibility = ["//:internal"],
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate"],
)

cel_android_library(
name = "async_gate_android",
testonly = 1,
visibility = ["//:internal"],
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate_android"],
)

java_library(
name = "async_completion_coordinator",
testonly = 1,
visibility = ["//:internal"],
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator"],
)

cel_android_library(
name = "async_completion_coordinator_android",
testonly = 1,
visibility = ["//:internal"],
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_completion_coordinator_android"],
)

java_library(
name = "async_call_state_tracker",
visibility = ["//:internal"],
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_call_state_tracker"],
)

cel_android_library(
name = "async_call_state_tracker_android",
visibility = ["//:internal"],
exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_call_state_tracker_android"],
)
4 changes: 2 additions & 2 deletions runtime/src/main/java/dev/cel/runtime/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -829,7 +829,6 @@ java_library(
":function_binding",
":function_resolver",
":partial_vars",
":program",
":proto_message_runtime_equality",
":runtime",
":runtime_equality",
Expand All @@ -855,7 +854,6 @@ java_library(
"//runtime:activation",
"//runtime:interpretable",
"//runtime:proto_message_activation_factory",
"//runtime:resolved_overload",
"//runtime/planner:planned_program",
"//runtime/planner:program_planner",
"//runtime/standard:type",
Expand Down Expand Up @@ -997,6 +995,7 @@ java_library(
"//common/types:type_providers",
"//common/values",
"//common/values:cel_value_provider",
"//runtime:async_options",
"//runtime:evaluation_exception",
"//runtime/planner:program_planner",
"//runtime/standard:standard_function",
Expand Down Expand Up @@ -1025,6 +1024,7 @@ cel_android_library(
"//common/types:type_providers_android",
"//common/values:cel_value_provider_android",
"//common/values:values_android",
"//runtime:async_options_android",
"//runtime:evaluation_exception",
"//runtime/planner:program_planner_android",
"//runtime/standard:standard_function_android",
Expand Down
4 changes: 1 addition & 3 deletions runtime/src/main/java/dev/cel/runtime/CelFunctionBinding.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,9 +188,7 @@ static ImmutableSet<CelFunctionBinding> fromOverloads(
String functionName, Collection<CelFunctionBinding> 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.
// TODO: Support dynamic overload resolution for asynchronous function overloads.
for (CelFunctionBinding binding : overloadBindings) {
checkArgument(
!(binding.getDefinition() instanceof CelAsyncFunctionOverload),
Expand Down
97 changes: 39 additions & 58 deletions runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,22 +112,7 @@ public Program createProgram(CelAbstractSyntaxTree ast) throws CelEvaluationExce
return toRuntimeProgram(planner().plan(ast));
}

private static final CelFunctionResolver EMPTY_FUNCTION_RESOLVER =
new CelFunctionResolver() {
@Override
public Optional<CelResolvedOverload> findOverloadMatchingArgs(
String functionName, Collection<String> overloadIds, Object[] args) {
return Optional.empty();
}

@Override
public Optional<CelResolvedOverload> findOverloadMatchingArgs(
String functionName, Object[] args) {
return Optional.empty();
}
};

public Program toRuntimeProgram(dev.cel.runtime.Program program) {
private Program toRuntimeProgram(PlannedProgram program) {
return new Program() {

@Override
Expand All @@ -148,11 +133,9 @@ public Object eval(Map<String, ?> mapValue, CelFunctionResolver lateBoundFunctio

@Override
public Object eval(Message message) throws CelEvaluationException {
PlannedProgram plannedProgram = (PlannedProgram) program;
return plannedProgram.evalOrThrow(
plannedProgram.interpretable(),
ProtoMessageActivationFactory.fromProto(message, plannedProgram.options()),
EMPTY_FUNCTION_RESOLVER,
return program.evalOrThrow(
ProtoMessageActivationFactory.fromProto(message, program.options()),
CelFunctionResolver.EMPTY,
/* partialVars= */ null,
/* listener= */ null);
}
Expand Down Expand Up @@ -190,12 +173,6 @@ public ListenableFuture<Object> evalAsync(
return program.evalAsync(mapValue, lateBoundFunctionResolver);
}

@Override
public ListenableFuture<Object> evalAsync(Message message) {
throw new UnsupportedOperationException(
"evalAsync is not supported by this Program implementation.");
}

@Override
public ListenableFuture<Object> evalAsync(CelVariableResolver resolver) {
return program.evalAsync(resolver);
Expand All @@ -212,40 +189,42 @@ public ListenableFuture<Object> evalAsync(PartialVars partialVars) {
return program.evalAsync(partialVars);
}

@Override
public ListenableFuture<Object> evalAsync(Message message) {
checkNotNull(message, "message");
return program.evalAsync(
ProtoMessageActivationFactory.fromProto(message, program.options()),
CelFunctionResolver.EMPTY,
/* partialVars= */ null);
}

@Override
public Object trace(CelEvaluationListener listener) throws CelEvaluationException {
return ((PlannedProgram) program)
.trace(GlobalResolver.EMPTY, EMPTY_FUNCTION_RESOLVER, null, listener);
return program.trace(GlobalResolver.EMPTY, CelFunctionResolver.EMPTY, null, listener);
}

@Override
public Object trace(Map<String, ?> mapValue, CelEvaluationListener listener)
throws CelEvaluationException {
return ((PlannedProgram) program)
.trace(Activation.copyOf(mapValue), EMPTY_FUNCTION_RESOLVER, null, listener);
return program.trace(
Activation.copyOf(mapValue), CelFunctionResolver.EMPTY, null, listener);
}

@Override
public Object trace(Message message, CelEvaluationListener listener)
throws CelEvaluationException {
PlannedProgram plannedProgram = (PlannedProgram) program;
return plannedProgram.evalOrThrow(
plannedProgram.interpretable(),
ProtoMessageActivationFactory.fromProto(message, plannedProgram.options()),
EMPTY_FUNCTION_RESOLVER,
return program.evalOrThrow(
ProtoMessageActivationFactory.fromProto(message, program.options()),
CelFunctionResolver.EMPTY,
/* partialVars= */ null,
listener);
}

@Override
public Object trace(CelVariableResolver resolver, CelEvaluationListener listener)
throws CelEvaluationException {
return ((PlannedProgram) program)
.trace(
(name) -> resolver.find(name).orElse(null),
EMPTY_FUNCTION_RESOLVER,
null,
listener);
return program.trace(
(name) -> resolver.find(name).orElse(null), CelFunctionResolver.EMPTY, null, listener);
}

@Override
Expand All @@ -254,12 +233,8 @@ public Object trace(
CelFunctionResolver lateBoundFunctionResolver,
CelEvaluationListener listener)
throws CelEvaluationException {
return ((PlannedProgram) program)
.trace(
(name) -> resolver.find(name).orElse(null),
lateBoundFunctionResolver,
null,
listener);
return program.trace(
(name) -> resolver.find(name).orElse(null), lateBoundFunctionResolver, null, listener);
}

@Override
Expand All @@ -268,23 +243,22 @@ public Object trace(
CelFunctionResolver lateBoundFunctionResolver,
CelEvaluationListener listener)
throws CelEvaluationException {
return ((PlannedProgram) program)
.trace(Activation.copyOf(mapValue), lateBoundFunctionResolver, null, listener);
return program.trace(
Activation.copyOf(mapValue), lateBoundFunctionResolver, null, listener);
}

@Override
public Object trace(PartialVars partialVars, CelEvaluationListener listener)
throws CelEvaluationException {
return ((PlannedProgram) program)
.trace(
(name) -> partialVars.resolver().find(name).orElse(null),
EMPTY_FUNCTION_RESOLVER,
partialVars,
listener);
return program.trace(
(name) -> partialVars.resolver().find(name).orElse(null),
CelFunctionResolver.EMPTY,
partialVars,
listener);
}

@Override
public Object advanceEvaluation(UnknownContext context) throws CelEvaluationException {
public Object advanceEvaluation(UnknownContext context) {
throw new UnsupportedOperationException("Unsupported operation.");
}
};
Expand Down Expand Up @@ -347,6 +321,10 @@ public abstract Builder setAsyncEvaluationOptions(
@Override
public abstract CelValueProvider valueProvider();

abstract CelAsyncEvaluationOptions asyncEvaluationOptions();

abstract Optional<ListeningExecutorService> asyncExecutor();

abstract CelStandardFunctions standardFunctions();

abstract ExtensionRegistry extensionRegistry();
Expand Down Expand Up @@ -604,7 +582,10 @@ public CelRuntime build() {
celValueConverter,
container(),
options(),
lateBoundFunctionNamesBuilder().build());
lateBoundFunctionNamesBuilder().build(),
runtimeEquality,
asyncEvaluationOptions(),
asyncExecutor().orElse(null));
setPlanner(planner);

setFunctionBindings(ImmutableMap.copyOf(mutableFunctionBindings));
Expand Down
5 changes: 4 additions & 1 deletion runtime/src/main/java/dev/cel/runtime/LiteRuntimeImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,7 +229,10 @@ public CelLiteRuntime build() {
celValueProvider.celValueConverter(),
container,
celOptions,
lateBoundFunctionNamesBuilder.build());
lateBoundFunctionNamesBuilder.build(),
// TODO: Support async eval in lite runtime.
CelAsyncEvaluationOptions.defaultOptions(),
/* asyncExecutor= */ null);

return new LiteRuntimeImpl(
planner,
Expand Down
8 changes: 6 additions & 2 deletions runtime/src/main/java/dev/cel/runtime/RuntimeEquality.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,12 +135,12 @@ public Optional<Object> findInMap(Map<?, ?> map, Object index) {
* comparable even if they are not of the same type, where type differences are usually trivially
* false.
*/
@SuppressWarnings({"rawtypes", "unchecked"})
@SuppressWarnings({"rawtypes", "unchecked", "ReferenceEquality"})
public boolean objectEquals(Object x, Object y) {
if (celOptions.disableCelStandardEquality()) {
return Objects.equals(x, y);
}
if (x == y) {
if (x == y && !isNan(x)) {
return true;
}
x = runtimeHelpers.adaptValue(x);
Expand Down Expand Up @@ -278,6 +278,10 @@ private static Optional<Long> unsignedToLongLossless(UnsignedLong v) {
return Optional.empty();
}

private static boolean isNan(Object value) {
return value instanceof Number && Double.isNaN(((Number) value).doubleValue());
}

RuntimeEquality(RuntimeHelpers runtimeHelpers, CelOptions celOptions) {
this.runtimeHelpers = runtimeHelpers;
this.celOptions = celOptions;
Expand Down
Loading
Loading