FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Add multi-threaded compilation and evaluation tests and resolve compi… · cel-expr/cel-python@04d2c2f · GitHub

/ cel-python Public

Commit 04d2c2f

Browse files
dmitriplotnikov
authored and
copybara-github
committed
Add multi-threaded compilation and evaluation tests and resolve compilation deadlock in CEL Python.
This change adds cel_parallel_test.py to benchmark and validate compiling and evaluating a diverse set of CEL expressions concurrently across multiple worker threads using concurrent.futures.ThreadPoolExecutor as well as sequentially. In addition, this resolves an AB-BA deadlock between the Python GIL and Protobuf DescriptorPool C++ mutex during concurrent compilation by releasing the GIL in PyCelEnv::Compile and acquiring the GIL via py::gil_scoped_acquire on-demand in PyDescriptorDatabase callbacks, while protecting PyCelEnvInternal with a mutex. Test duration metrics: - Multi-threaded compilation (1,000 iterations): ~248 ms - Sequential compilation (1,000 iterations): ~394 ms - Multi-threaded evaluation (10,000 iterations): ~782 ms - Sequential evaluation (10,000 iterations): ~434 ms PiperOrigin-RevId: 968187521
1 parent ce7b742 commit 04d2c2f

9 files changed

Lines changed: 292 additions & 64 deletions

‎MODULE.bazel‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ bazel_dep(name = "bazel_skylib", version = "1.9.0")
1919
bazel_dep(name = "cel-cpp", version = "0.16.1", repo_name = "com_google_cel_cpp")
2020
git_override(
2121
module_name = "cel-cpp",
22-
commit = "1dcff093b3ca3d575f982e2ca154f9c751e06055",
22+
commit = "98a7da06d3b9492e10e22d624fe55d749ce137c0",
2323
remote = "https://github.com/cel-expr/cel-cpp",
2424
)
2525

‎cel_expr_python/BUILD‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ pybind_library(
5151
":cel_extension",
5252
":status_macros",
5353
"@com_google_absl//absl/base",
54+
"@com_google_absl//absl/base:core_headers",
5455
"@com_google_absl//absl/base:no_destructor",
5556
"@com_google_absl//absl/container:flat_hash_map",
5657
"@com_google_absl//absl/functional:function_ref",
@@ -60,6 +61,7 @@ pybind_library(
6061
"@com_google_absl//absl/status:statusor",
6162
"@com_google_absl//absl/strings",
6263
"@com_google_absl//absl/strings:str_format",
64+
"@com_google_absl//absl/synchronization",
6365
"@com_google_absl//absl/time",
6466
"@com_google_absl//absl/types:optional",
6567
"@com_google_absl//absl/types:span",
@@ -94,7 +96,6 @@ pybind_library(
9496
"@com_google_cel_cpp//runtime:reference_resolver",
9597
"@com_google_cel_cpp//runtime:runtime_builder",
9698
"@com_google_cel_cpp//runtime:runtime_options",
97-
"@com_google_cel_cpp//validator",
9899
"@com_google_cel_spec//proto/cel/expr:checked_cc_proto",
99100
"@com_google_cel_spec//proto/cel/expr:syntax_cc_proto",
100101
"@com_google_protobuf//:protobuf",
@@ -166,6 +167,21 @@ py_test(
166167
}),
167168
)
168169

170+
py_test(
171+
name = "cel_parallel_test",
172+
srcs = ["cel_parallel_test.py"],
173+
data = [
174+
":cel",
175+
],
176+
deps = [
177+
"//testing:proto2_test_all_types_py_pb2",
178+
"@com_google_absl_py//absl/testing:absltest",
179+
] + select({
180+
"@platforms//os:windows": [],
181+
"//conditions:default": [":cel"],
182+
}),
183+
)
184+
169185
py_test(
170186
name = "cel_env_test",
171187
srcs = ["cel_env_test.py"],

‎cel_expr_python/cel_parallel_test.py‎

Lines changed: 204 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,204 @@
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+
"""Multi-threaded tests for cel-python."""
16+
17+
import collections.abc
18+
import concurrent.futures
19+
import dataclasses
20+
import gc
21+
import logging
22+
import time
23+
from typing import Any
24+
25+
from absl.testing import absltest
26+
from cel_expr_python import cel
27+
from cel.expr.conformance.proto2 import test_all_types_pb2 as test_all_types_pb
28+
29+
30+
@dataclasses.dataclass(frozen=True)
31+
class _TestCase:
32+
expr: str
33+
data: collections.abc.Callable[[int], dict[str, Any]]
34+
expected: collections.abc.Callable[[int], Any]
35+
36+
37+
_NUM_EVALUATIONS = 10000
38+
_NUM_COMPILATIONS = 1000
39+
40+
_TEST_MSG = test_all_types_pb.TestAllTypes(single_int64=100)
41+
42+
_TEST_CASES = [
43+
_TestCase(
44+
expr="var_int * var_int",
45+
data=lambda n: {"var_int": n},
46+
expected=lambda n: n * n,
47+
),
48+
_TestCase(
49+
expr="var_str + '_' + string(var_int)",
50+
data=lambda n: {"var_str": "num", "var_int": n},
51+
expected=lambda n: f"num_{n}",
52+
),
53+
_TestCase(
54+
expr="var_int % 2 == 0",
55+
data=lambda n: {"var_int": n},
56+
expected=lambda n: n % 2 == 0,
57+
),
58+
_TestCase(
59+
expr="[var_int, var_int + 1, var_int + 2]",
60+
data=lambda n: {"var_int": n},
61+
expected=lambda n: [n, n + 1, n + 2],
62+
),
63+
_TestCase(
64+
expr="var_int_map[var_int]",
65+
data=lambda n: {"var_int_map": {n: f"val_{n}"}, "var_int": n},
66+
expected=lambda n: f"val_{n}",
67+
),
68+
_TestCase(
69+
expr="var_msg.single_int64 + var_int",
70+
data=lambda n: {"var_msg": _TEST_MSG, "var_int": n},
71+
expected=lambda n: 100 + n,
72+
),
73+
_TestCase(
74+
expr=(
75+
"cel.expr.conformance.proto2.TestAllTypes{"
76+
" single_int64: var_int, single_string: var_str"
77+
"}"
78+
),
79+
data=lambda n: {"var_int": n, "var_str": f"msg_{n}"},
80+
expected=lambda n: test_all_types_pb.TestAllTypes(
81+
single_int64=n, single_string=f"msg_{n}"
82+
),
83+
),
84+
_TestCase(
85+
expr="{'key': var_str, 'value': var_int}",
86+
data=lambda n: {"var_str": f"val_{n}", "var_int": n},
87+
expected=lambda n: {"key": f"val_{n}", "value": n},
88+
),
89+
_TestCase(
90+
expr="[var_int, var_int + 1, var_int + 2].all(x, x >= var_int)",
91+
data=lambda n: {"var_int": n},
92+
expected=lambda n: True,
93+
),
94+
]
95+
96+
97+
class CelParallelTest(absltest.TestCase):
98+
99+
def setUp(self):
100+
super().setUp()
101+
102+
self.env = cel.NewEnv(
103+
variables={
104+
"var_int": cel.Type.INT,
105+
"var_str": cel.Type.STRING,
106+
"var_int_map": cel.Type.Map(cel.Type.INT, cel.Type.STRING),
107+
"var_msg": cel.Type("cel.expr.conformance.proto2.TestAllTypes"),
108+
},
109+
)
110+
self.object_counts_before_test = self._grab_object_counts()
111+
112+
def tearDown(self):
113+
"""Tears down the test environment."""
114+
super().tearDown()
115+
116+
gc.collect()
117+
# Assert that all Arenas have been garbage-collected
118+
self.assertEqual(cel._InternalArena._get_instance_count(), 0)
119+
self._check_for_leaks()
120+
121+
def _grab_object_counts(self) -> dict[str, int]:
122+
gc.collect()
123+
all_objects = gc.get_objects()
124+
type_counts = {}
125+
for obj in all_objects:
126+
obj_type = type(obj)
127+
type_counts[obj_type.__name__] = type_counts.get(obj_type, 0) + 1
128+
return type_counts
129+
130+
def _check_for_leaks(self):
131+
type_counts = self._grab_object_counts()
132+
for key, count in type_counts.items():
133+
if count != self.object_counts_before_test.get(key, 0):
134+
self.fail(
135+
f"Object count for {key} did not match expected count. "
136+
f"Expected: {self.object_counts_before_test.get(key, 0)}, "
137+
f"Actual: {count}",
138+
)
139+
140+
def _test_eval(self, multi_threaded: bool):
141+
compiled_exprs = [self.env.compile(tc.expr) for tc in _TEST_CASES]
142+
143+
def eval_expr(n: int) -> Any:
144+
idx = n % len(_TEST_CASES)
145+
test_case = _TEST_CASES[idx]
146+
expr = compiled_exprs[idx]
147+
data = test_case.data(n)
148+
return expr.eval(data=data).plain_value()
149+
150+
start_time = time.perf_counter()
151+
if multi_threaded:
152+
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
153+
results = list(executor.map(eval_expr, range(_NUM_EVALUATIONS)))
154+
else:
155+
results = [eval_expr(n) for n in range(_NUM_EVALUATIONS)]
156+
duration_ms = (time.perf_counter() - start_time) * 1000
157+
158+
mode = "Multi-threaded" if multi_threaded else "Sequential"
159+
logging.info("%s evaluation duration: %.2f ms", mode, duration_ms)
160+
161+
self.assertLen(results, _NUM_EVALUATIONS)
162+
for i, res in enumerate(results):
163+
test_case = _TEST_CASES[i % len(_TEST_CASES)]
164+
self.assertEqual(res, test_case.expected(i))
165+
166+
def testMultiThreadedEval(self):
167+
self._test_eval(multi_threaded=True)
168+
169+
def testSequentialEval(self):
170+
self._test_eval(multi_threaded=False)
171+
172+
def _test_compile(self, multi_threaded: bool):
173+
def compile_expr(n: int) -> cel.Expression:
174+
test_case = _TEST_CASES[n % len(_TEST_CASES)]
175+
return self.env.compile(test_case.expr)
176+
177+
start_time = time.perf_counter()
178+
if multi_threaded:
179+
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
180+
results = list(executor.map(compile_expr, range(_NUM_COMPILATIONS)))
181+
else:
182+
results = [compile_expr(n) for n in range(_NUM_COMPILATIONS)]
183+
duration_ms = (time.perf_counter() - start_time) * 1000
184+
185+
mode = "Multi-threaded" if multi_threaded else "Sequential"
186+
logging.info("%s compilation duration: %.2f ms", mode, duration_ms)
187+
188+
self.assertLen(results, _NUM_COMPILATIONS)
189+
for i, expr in enumerate(results):
190+
test_case = _TEST_CASES[i % len(_TEST_CASES)]
191+
data = test_case.data(i)
192+
self.assertEqual(
193+
expr.eval(data=data).plain_value(), test_case.expected(i)
194+
)
195+
196+
def testMultiThreadedCompilation(self):
197+
self._test_compile(multi_threaded=True)
198+
199+
def testSequentialCompilation(self):
200+
self._test_compile(multi_threaded=False)
201+
202+
203+
if __name__ == "__main__":
204+
absltest.main()

‎cel_expr_python/py_cel_env.cc‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -196,6 +196,18 @@ std::shared_ptr<PyCelActivation> PyCelEnv::NewActivation(
196196

197197
PyCelExpression PyCelEnv::Compile(const std::string& cel_expr,
198198
bool disable_check) {
199+
// Release the GIL before entering C++ compilation to prevent lock
200+
// inversion/deadlock with DescriptorPool's internal mutex during concurrent
201+
// multi-threaded compilation.
202+
//
203+
// When DescriptorPool performs a descriptor lookup on a cache miss, it calls
204+
// back into Python via PyDescriptorDatabase (which re-acquires the GIL via
205+
// PyGILState_Ensure). If another thread were to enter Compile() with the GIL
206+
// held, it would block on DescriptorPool's internal C++ mutex while holding
207+
// the GIL, causing an AB-BA deadlock with any thread inside
208+
// PyDescriptorDatabase waiting for the GIL. Releasing the GIL here guarantees
209+
// a strict one-way lock hierarchy (DescriptorPool Mutex -> Python GIL).
210+
py::gil_scoped_release gil_release;
199211
return ThrowIfError(PyCelExpression::Compile(env_, cel_expr, disable_check));
200212
}
201213

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL