This is an automated email from the ASF dual-hosted git repository.
Croway pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new dddd56f1dd77 CAMEL-24690: camel-python - evaluate under the
interpreter lock, the globals are shared (#26311)
dddd56f1dd77 is described below
commit dddd56f1dd77f00fe85abb1420bc1f32c76188fc
Author: Federico Mariani <[email protected]>
AuthorDate: Mon Sep 14 12:16:57 2026 +0200
CAMEL-24690: camel-python - evaluate under the interpreter lock, the
globals are shared (#26311)
* CAMEL-24690: camel-python - evaluate a python expression under a lock;
the interpreter globals are shared
PythonExpression keeps one PythonInterpreter per expression and evaluate()
sets exchange, body,
headers and properties into its globals, runs the compiled code and calls
cleanup(), with no
synchronization. Concurrent exchanges interleave those steps, so one
exchange can be evaluated
with another one's body and cleanup() runs while another thread is
mid-eval. Binding, running and
cleaning up now happen under the interpreter's monitor, in PythonExpression
and in the
ScriptingLanguage entry point of PythonLanguage. A new test evaluates one
expression from 8 threads
and checks every result.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* CAMEL-24690: camel-python - compile under the same lock, keep the monitor
private
PythonInterpreter.compile() reads and writes the interpreter's shared
CompilerFlags and installs
its PySystemState as the thread's current one, so compiling outside the
lock raced the
evaluation of another thread; the cache miss now runs inside it. Both
classes synchronize on a
private lock object rather than the interpreter itself.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
---------
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
.../camel/language/python/PythonExpression.java | 9 +++
.../camel/language/python/PythonLanguage.java | 44 +++++++------
.../python/PythonExpressionConcurrentTest.java | 76 ++++++++++++++++++++++
3 files changed, 109 insertions(+), 20 deletions(-)
diff --git
a/components/camel-python/src/main/java/org/apache/camel/language/python/PythonExpression.java
b/components/camel-python/src/main/java/org/apache/camel/language/python/PythonExpression.java
index 7202f12b1594..3de078780711 100644
---
a/components/camel-python/src/main/java/org/apache/camel/language/python/PythonExpression.java
+++
b/components/camel-python/src/main/java/org/apache/camel/language/python/PythonExpression.java
@@ -29,6 +29,7 @@ public class PythonExpression extends ExpressionSupport {
private final Class<?> type;
private final PythonInterpreter compiler;
private final PyCode compiledExpression;
+ private final Object lock = new Object();
public PythonExpression(String expressionString, Class<?> type) {
this.expressionString = expressionString;
@@ -47,6 +48,14 @@ public class PythonExpression extends ExpressionSupport {
@Override
public <T> T evaluate(Exchange exchange, Class<T> type) {
+ // the interpreter's globals are shared by every evaluation of this
expression: bind, run and clean up
+ // under one lock so concurrent exchanges cannot see each other's
bindings
+ synchronized (lock) {
+ return doEvaluate(exchange, type);
+ }
+ }
+
+ private <T> T doEvaluate(Exchange exchange, Class<T> type) {
try {
compiler.set("exchange", exchange);
compiler.set("context", exchange.getContext());
diff --git
a/components/camel-python/src/main/java/org/apache/camel/language/python/PythonLanguage.java
b/components/camel-python/src/main/java/org/apache/camel/language/python/PythonLanguage.java
index fbad8206d2b5..e0aa4a8c6d7c 100644
---
a/components/camel-python/src/main/java/org/apache/camel/language/python/PythonLanguage.java
+++
b/components/camel-python/src/main/java/org/apache/camel/language/python/PythonLanguage.java
@@ -38,6 +38,8 @@ public class PythonLanguage extends TypedLanguageSupport
implements ScriptingLan
private final PythonInterpreter compiler = new PythonInterpreter();
+ private final Object lock = new Object();
+
private PythonLanguage(Map<String, PyCode> compiledScriptsCache) {
this.compiledScriptsCache = compiledScriptsCache;
}
@@ -64,31 +66,33 @@ public class PythonLanguage extends TypedLanguageSupport
implements ScriptingLan
public <T> T evaluate(String script, Map<String, Object> bindings,
Class<T> resultType) {
script = loadResource(script);
- PyCode code = getCompiledScriptFromCache(script);
-
- if (code == null) {
+ // compile, bind, run and clean up under one lock: the interpreter's
compiler flags, system state and
+ // globals are all shared by every caller of this method
+ synchronized (lock) {
+ PyCode code = getCompiledScriptFromCache(script);
+ if (code == null) {
+ try {
+ code = compiler.compile(script);
+ addCompiledScriptToCache(script, code);
+ } catch (Exception e) {
+ throw new ExpressionIllegalSyntaxException(script, e);
+ }
+ }
try {
- code = compiler.compile(script);
- addCompiledScriptToCache(script, code);
+ if (bindings != null) {
+ bindings.forEach(compiler::set);
+ }
+ PyObject out = compiler.eval(code);
+ if (out != null) {
+ String value = out.toString();
+ return
getCamelContext().getTypeConverter().convertTo(resultType, value);
+ }
} catch (Exception e) {
throw new ExpressionIllegalSyntaxException(script, e);
+ } finally {
+ compiler.cleanup();
}
}
-
- try {
- if (bindings != null) {
- bindings.forEach(compiler::set);
- }
- PyObject out = compiler.eval(code);
- if (out != null) {
- String value = out.toString();
- return
getCamelContext().getTypeConverter().convertTo(resultType, value);
- }
- } catch (Exception e) {
- throw new ExpressionIllegalSyntaxException(script, e);
- } finally {
- compiler.cleanup();
- }
return null;
}
diff --git
a/components/camel-python/src/test/java/org/apache/camel/language/python/PythonExpressionConcurrentTest.java
b/components/camel-python/src/test/java/org/apache/camel/language/python/PythonExpressionConcurrentTest.java
new file mode 100644
index 000000000000..584222a993d8
--- /dev/null
+++
b/components/camel-python/src/test/java/org/apache/camel/language/python/PythonExpressionConcurrentTest.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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 org.apache.camel.language.python;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Expression;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Concurrent evaluations of one python expression must not see each other's
bindings.
+ */
+public class PythonExpressionConcurrentTest extends CamelTestSupport {
+
+ @Test
+ public void testConcurrentEvaluationsDoNotShareBindings() throws Exception
{
+ Expression expression =
context.resolveLanguage("python").createExpression("body + '-' +
str(headers['n'])");
+ int threads = 8;
+ int rounds = 200;
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ CountDownLatch start = new CountDownLatch(1);
+ List<String> failures = new CopyOnWriteArrayList<>();
+ List<Runnable> tasks = new ArrayList<>();
+ for (int t = 0; t < threads; t++) {
+ final int id = t;
+ tasks.add(() -> {
+ try {
+ start.await();
+ for (int i = 0; i < rounds; i++) {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getMessage().setBody("t" + id);
+ exchange.getMessage().setHeader("n", i);
+ String expected = "t" + id + "-" + i;
+ String actual = expression.evaluate(exchange,
String.class);
+ if (!expected.equals(actual)) {
+ failures.add(expected + " != " + actual);
+ }
+ }
+ } catch (Exception e) {
+ failures.add(e.toString());
+ }
+ });
+ }
+ tasks.forEach(pool::submit);
+ start.countDown();
+ pool.shutdown();
+ assertTrue(pool.awaitTermination(2, TimeUnit.MINUTES));
+ assertEquals(List.of(), failures);
+ }
+}
