sunchao commented on code in PR #7181:
URL: https://github.com/apache/datafusion-comet/pull/7181#discussion_r4366518997
##########
spark/src/main/scala/org/apache/comet/CometExecIterator.scala:
##########
@@ -235,6 +235,11 @@ class CometExecIterator(
CometExecIterator.startMemoryUsageLog()
+ // A native Python UDF ignores the task's kill, so watch the task to
interrupt its calls.
+ if (PythonUdfMonitor.supported && nativeLib.hasPythonUdf(plan)) {
Review Comment:
[P1] Compute `hasPythonUdf` before registering the listener that calls
`close()`. Spark can invoke a completion listener immediately when an input
feeder constructs its parent iterator after the consuming task completes.
`close()` then frees `plan` before this line reads it. That can terminate the
executor for any plan in a Python-enabled build, even with native UDF execution
disabled in the session. Cache the Boolean in the cleanup-protected
`createdPlan` block.
Evidence: A bounded Spark 4.1.3 `PipedRDD` fixture recorded `completed=true
closeRanBeforeProbe=true` using an inert handle. At this head, line 232
publishes the listener, line 375 calls `releasePlan`, and `hasPythonUdf`
dereferences the released `ExecutionContext` to read `spark_plan`. The base
constructor made no native plan access after publishing that listener.
##########
spark/src/main/scala/org/apache/comet/PythonUdfMonitor.scala:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.comet
+
+import java.util.concurrent.ConcurrentHashMap
+
+import org.apache.spark.TaskContext
+import org.apache.spark.internal.Logging
+
+/**
+ * Interrupts the native Python UDF calls of killed tasks. Spark's Python
runner has a
+ * `MonitorThread` that destroys the worker of a killed task. A native UDF
runs in the embedded
+ * interpreter on the task thread instead, where neither a Java interrupt nor
a stopped input
+ * reaches it. So one daemon thread per task polls its kill state and, while
the task is killed
+ * but still running, raises an exception in the task's running Python calls,
once per call so
+ * that its cleanup can run. Calls that start after the first interrupt fail
without entering
+ * Python.
+ */
+object PythonUdfMonitor extends Logging {
+
+ // Raising the exception is cheap and lets the call unwind on its own, so
unlike Spark's
+ // monitor this one does not wait `spark.python.task.killTimeout` before
acting. Repeating it
+ // reaches calls of plans that the task registers after an earlier attempt.
+ private val PollIntervalMs = 100L
+
+ private val watchedTasks = ConcurrentHashMap.newKeySet[java.lang.Long]()
+
+ /** Whether the native library can run Python UDFs, and so may need its
calls interrupted. */
+ lazy val supported: Boolean = NativeBase.supportsPythonUdf()
+
+ /** Starts watching the task unless it is watched already. The watch ends
with the task. */
+ def watch(context: TaskContext): Unit = {
+ val taskAttemptId = context.taskAttemptId()
+ if (watchedTasks.add(taskAttemptId)) {
+ val thread = new Thread(s"Comet Python UDF monitor for task
$taskAttemptId") {
+ override def run(): Unit = {
+ try {
+ while (!context.isCompleted()) {
Review Comment:
[P2] Keep the watch alive until the last native UDF plan finishes closing.
Spark sets `isCompleted` before invoking completion listeners. When a consumer
takes the first batch and the producer is evaluating the next one,
`releasePlan` can wait in `BatchProducer.stop()` after the monitor exits. A
kill during that wait cannot interrupt even a pure Python loop, which can
retain the executor slot indefinitely.
Evidence: A Spark 4.1.3 `take(1)` fixture used the head’s monitor, UDF
module, and byte-identical `BatchProducer`. After `cancelJobGroup`, it observed
`completed=true`, `interrupted=true`, zero native interrupts, and both plan
close and task end pending. A direct interrupt returned `1`, released close,
and Spark reported `TaskKilled`. [Spark completion
order](https://github.com/apache/spark/blob/v4.2.0/core/src/main/scala/org/apache/spark/TaskContextImpl.scala#L159-L164)
##########
native/core/src/execution/python_udf.rs:
##########
@@ -29,14 +29,177 @@ use pyo3::ffi::Py_uintptr_t;
use pyo3::prelude::*;
use pyo3::sync::MutexExt;
use pyo3::types::{PyBytes, PyList, PyTuple};
-use std::sync::{Mutex, PoisonError};
+use pyo3::PyTypeInfo;
+use std::collections::HashMap;
+use std::os::raw::{c_long, c_ulong};
+use std::sync::{Arc, LazyLock, Mutex, MutexGuard, PoisonError};
// Tasks create their callables concurrently. Concurrent first imports of a
package
// can hand a thread a partially initialized module (observed with PyArrow on
Python
// 3.10), so callables are unpickled one at a time. Take it with
`lock_py_attached`,
// which waits for the lock without holding the GIL.
static IMPORT_LOCK: Mutex<()> = Mutex::new(());
+pyo3::create_exception!(
+ comet,
+ TaskKilled,
+ pyo3::exceptions::PyBaseException,
+ "Raised in a running Arrow UDF when Spark kills its task."
+);
+
+unsafe extern "C" {
+ // Part of the stable Python C API, but not declared by pyo3-ffi.
+ fn PyThread_get_thread_ident() -> c_ulong;
+}
+
+/// The native plans of one Spark task attempt that hold Arrow UDFs.
+#[derive(Default)]
+struct TaskCalls {
+ plans: usize,
+ killed: bool,
+ /// The calls running now.
+ calls: Vec<Call>,
+}
+
+struct Call {
+ /// Python thread id of the thread running the call.
+ thread: c_ulong,
+ /// Whether [`interrupt_task`] has raised in the call. It raises once, so
that a second
+ /// exception does not abort the cleanup the first one started.
+ interrupted: bool,
+}
+
+// Holders of this lock never wait for the GIL, so a thread attached to Python
may take it.
+static TASKS: LazyLock<Mutex<HashMap<i64, TaskCalls>>> =
LazyLock::new(Default::default);
+
+fn tasks() -> MutexGuard<'static, HashMap<i64, TaskCalls>> {
+ TASKS.lock().unwrap_or_else(PoisonError::into_inner)
+}
+
+/// Registers a native plan's Arrow UDFs under their Spark task attempt, so
that
+/// [`interrupt_task`] reaches their calls. The task's entry is removed when
its last
+/// registration is dropped, which happens when the plan is released.
+pub struct TaskRegistration {
+ task_attempt_id: i64,
+}
+
+impl TaskRegistration {
+ pub fn new(task_attempt_id: i64) -> Self {
+ tasks().entry(task_attempt_id).or_default().plans += 1;
+ Self { task_attempt_id }
+ }
+
+ fn is_killed(&self) -> bool {
+ tasks()
+ .get(&self.task_attempt_id)
+ .is_some_and(|task| task.killed)
+ }
+
+ fn killed_error(&self) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "Arrow UDF stopped because task {} was killed",
+ self.task_attempt_id
+ ))
+ }
+}
+
+impl Drop for TaskRegistration {
+ fn drop(&mut self) {
+ let mut tasks = tasks();
+ if let Some(task) = tasks.get_mut(&self.task_attempt_id) {
+ task.plans -= 1;
+ if task.plans == 0 {
+ tasks.remove(&self.task_attempt_id);
+ }
+ }
+ }
+}
+
+/// A call to the user's callable, recorded so that [`interrupt_task`] can
raise in its thread.
+struct RunningCall<'a> {
+ task: &'a TaskRegistration,
+ thread: c_ulong,
+}
+
+impl<'a> RunningCall<'a> {
+ /// Fails without recording the call when the task has already been killed.
+ fn enter(_py: Python<'_>, task: &'a TaskRegistration) -> Result<Self> {
+ // SAFETY: the caller is attached to the interpreter.
+ let thread = unsafe { PyThread_get_thread_ident() };
+ let mut tasks = tasks();
+ let calls = tasks.entry(task.task_attempt_id).or_default();
+ if calls.killed {
+ return Err(task.killed_error());
+ }
+ calls.calls.push(Call {
+ thread,
+ interrupted: false,
+ });
+ Ok(Self { task, thread })
+ }
+}
+
+impl Drop for RunningCall<'_> {
+ fn drop(&mut self) {
+ let mut tasks = tasks();
+ let Some(calls) = tasks.get_mut(&self.task.task_attempt_id) else {
+ return;
+ };
+ let Some(index) = calls.calls.iter().position(|c| c.thread ==
self.thread) else {
+ return;
+ };
+ if calls.calls.swap_remove(index).interrupted {
+ // An interrupt that arrived as the callable returned is still
pending on this
+ // thread. Nothing can set another one now that the call is
unrecorded, so clear
+ // it rather than let it fire in later Python code on this thread.
+ // SAFETY: the call is made while attached, on the thread it names.
+ unsafe {
+ pyo3::ffi::PyThreadState_SetAsyncExc(self.thread as c_long,
std::ptr::null_mut())
Review Comment:
[P2] Avoid unconditionally clearing an interrupt that Python already
delivered. On CPython 3.11 and 3.12, `PyThreadState_SetAsyncExc(id, NULL)`
re-arms the interpreter-wide pending flag even when no thread has an exception
to consume. Other UDFs then repeatedly enter the pending-event path. After
unregistering the call and releasing the mutex, consume any genuinely late
interrupt on its originating thread before reusing it.
Evidence: For a surviving UDF summing three million integers, median thread
CPU over 13 samples rose from 100.22 to 111.71 ms on 3.11.15 and from 134.66 to
154.77 ms on 3.12.13. Both reported a pending flag of `1` with zero target
threads. A disposable Python checkpoint kept the flag at `0` and also passed
the case where a C callable returns directly to Rust with an interrupt pending.
[CPython
setter](https://github.com/python/cpython/blob/v3.12.13/Python/pystate.c#L1832-L1838)
· [Pending-event
handler](https://github.com/python/cpython/blob/v3.12.13/Python/ceval_gil.c#L1065-L1073)
##########
native/core/src/execution/python_udf.rs:
##########
@@ -29,14 +29,177 @@ use pyo3::ffi::Py_uintptr_t;
use pyo3::prelude::*;
use pyo3::sync::MutexExt;
use pyo3::types::{PyBytes, PyList, PyTuple};
-use std::sync::{Mutex, PoisonError};
+use pyo3::PyTypeInfo;
+use std::collections::HashMap;
+use std::os::raw::{c_long, c_ulong};
+use std::sync::{Arc, LazyLock, Mutex, MutexGuard, PoisonError};
// Tasks create their callables concurrently. Concurrent first imports of a
package
// can hand a thread a partially initialized module (observed with PyArrow on
Python
// 3.10), so callables are unpickled one at a time. Take it with
`lock_py_attached`,
// which waits for the lock without holding the GIL.
static IMPORT_LOCK: Mutex<()> = Mutex::new(());
+pyo3::create_exception!(
+ comet,
+ TaskKilled,
+ pyo3::exceptions::PyBaseException,
+ "Raised in a running Arrow UDF when Spark kills its task."
+);
+
+unsafe extern "C" {
+ // Part of the stable Python C API, but not declared by pyo3-ffi.
+ fn PyThread_get_thread_ident() -> c_ulong;
+}
+
+/// The native plans of one Spark task attempt that hold Arrow UDFs.
+#[derive(Default)]
+struct TaskCalls {
+ plans: usize,
+ killed: bool,
+ /// The calls running now.
+ calls: Vec<Call>,
+}
+
+struct Call {
+ /// Python thread id of the thread running the call.
+ thread: c_ulong,
+ /// Whether [`interrupt_task`] has raised in the call. It raises once, so
that a second
+ /// exception does not abort the cleanup the first one started.
+ interrupted: bool,
+}
+
+// Holders of this lock never wait for the GIL, so a thread attached to Python
may take it.
+static TASKS: LazyLock<Mutex<HashMap<i64, TaskCalls>>> =
LazyLock::new(Default::default);
+
+fn tasks() -> MutexGuard<'static, HashMap<i64, TaskCalls>> {
+ TASKS.lock().unwrap_or_else(PoisonError::into_inner)
+}
+
+/// Registers a native plan's Arrow UDFs under their Spark task attempt, so
that
+/// [`interrupt_task`] reaches their calls. The task's entry is removed when
its last
+/// registration is dropped, which happens when the plan is released.
+pub struct TaskRegistration {
+ task_attempt_id: i64,
+}
+
+impl TaskRegistration {
+ pub fn new(task_attempt_id: i64) -> Self {
+ tasks().entry(task_attempt_id).or_default().plans += 1;
Review Comment:
[P2] Key cancellation by a process-unique token shared by the plans of one
`TaskContext`. A cancelled local task can remain in native code after its
`SparkContext` stops. If the replacement context reuses its attempt ID, this
registration inherits the old `killed` flag and rejects a healthy UDF. Use the
same token for `watchedTasks` and the native registry.
Evidence: Two sequential Spark 4.1.3 `local[1]` contexts both assigned
attempt ID `0`. While the first context’s cancelled call was still waiting, the
replacement’s identity UDF received input `73` and returned `Arrow UDF stopped
because task 0 was killed` with `isInterrupted=false`. Attempt `1` returned
`73` for the same UDF and input. [Spark attempt-ID
contract](https://github.com/apache/spark/blob/v4.1.3/core/src/main/scala/org/apache/spark/TaskContext.scala#L231-L235)
##########
native/core/src/execution/python_udf.rs:
##########
@@ -297,14 +468,18 @@ impl ArrowPythonUdf {
kwargs.set_item(name, arg).map_err(python_error)?;
}
}
- let result = self
- .callable
- .bind(py)
- .call(
- PyTuple::new(py, positional).map_err(python_error)?,
- Some(&kwargs),
- )
- .map_err(|error| python_traceback_error(py, error))?;
+ let positional = PyTuple::new(py,
positional).map_err(python_error)?;
+ let call = self
+ .task
+ .as_deref()
+ .map(|task| RunningCall::enter(py, task))
+ .transpose()?;
+ let result = self.callable.bind(py).call(positional,
Some(&kwargs));
+ drop(call);
Review Comment:
[P2] Keep the call registered through user-defined result conversion and
check cancellation before returning the converted batch. Spark 4.2 accepts
array-like results whose `__arrow_array__` executes Python inside
`RecordBatch.from_arrays`. Dropping `RunningCall` here makes that code
invisible to the monitor. A cancelled conversion can keep running and
ultimately return a successful batch.
Evidence: On CPython 3.11.15 and 3.12.13, two interrupts during a bounded
`__arrow_array__` loop both returned `0`. After the loop was released, native
evaluation returned `[73]`. Stock Spark 4.2 accepted the same result shape and
stopped the worker during conversion on cancellation. [Spark result
materialization](https://github.com/apache/spark/blob/v4.2.0/python/pyspark/worker.py#L2563-L2581)
##########
native/core/src/execution/python_udf.rs:
##########
@@ -297,14 +468,18 @@ impl ArrowPythonUdf {
kwargs.set_item(name, arg).map_err(python_error)?;
}
}
- let result = self
- .callable
- .bind(py)
- .call(
- PyTuple::new(py, positional).map_err(python_error)?,
- Some(&kwargs),
- )
- .map_err(|error| python_traceback_error(py, error))?;
+ let positional = PyTuple::new(py,
positional).map_err(python_error)?;
+ let call = self
+ .task
+ .as_deref()
+ .map(|task| RunningCall::enter(py, task))
Review Comment:
[P2] Check the killed flag before `Python::attach`, and retain this check in
`RunningCall::enter` for races. A batch submitted after the registration is
already marked killed currently waits for the GIL and imports its arguments
before it can fail. An unrelated UDF holding the GIL can therefore retain this
cancelled task’s executor slot even though its callable will never run.
Evidence: A CPython 3.12.13 probe marked task `101` killed with no active
calls, then evaluated its next batch while another UDF held the GIL in
`math.factorial(420000)`. Returning the kill error took 1,481 ms. With a
pre-attach check, the same evaluation returned in under 0.1 ms.
##########
native/core/src/execution/python_udf.rs:
##########
@@ -600,6 +775,122 @@ mod tests {
});
}
+ fn running_calls(task_attempt_id: i64) -> Option<usize> {
+ tasks().get(&task_attempt_id).map(|task| task.calls.len())
+ }
+
+ #[test]
+ fn interrupts_running_call_of_killed_task() {
+ const TASK: i64 = -6811;
+ initialize_python().unwrap();
+ // `except Exception` must not swallow the interrupt.
+ let command = Python::attach(|py| {
+ cloudpickled_lambda(
+ py,
+ c"(lambda ns: (exec('def f(a):\\n n = 0\\n while
True:\\n n += 1\\n try:\\n n %= 7\\n except
Exception:\\n pass\\n', ns), ns['f'])[1])({})",
+ )
+ });
+ let task = Arc::new(TaskRegistration::new(TASK));
+ let udf = Arc::new(
+ ArrowPythonUdf::from_command(&command, DataType::Int64, true,
true, false, "")
+ .unwrap()
+ .with_task(Arc::clone(&task)),
+ );
+ let input: ArrayRef = Arc::new(Int64Array::from(vec![1]));
+ let (sender, receiver) = std::sync::mpsc::channel();
+ {
+ let (udf, input) = (Arc::clone(&udf), Arc::clone(&input));
+ // Not joined: a call that ignores the interrupt would never
return.
+ std::thread::spawn(move || sender.send(udf.evaluate(&[input], 1)));
Review Comment:
[P2] Release the worker’s cloned `Arc<ArrowPythonUdf>` before sending the
result, or join the worker after the timed receive succeeds. The receiver can
wake while the detached worker still owns the registration. The final assertion
then intermittently fails even though the Python call was interrupted and
unregistered correctly.
Evidence: The unmodified `interrupts_running_call_of_killed_task` failed 5
of 24 isolated CPython 3.12.13 runs at line 825 with actual `Some(0)` and
expected `None`. A control changing only the worker to drop its `Arc` before
`send` passed all 24 runs.
##########
native/core/src/execution/python_udf.rs:
##########
@@ -29,14 +29,177 @@ use pyo3::ffi::Py_uintptr_t;
use pyo3::prelude::*;
use pyo3::sync::MutexExt;
use pyo3::types::{PyBytes, PyList, PyTuple};
-use std::sync::{Mutex, PoisonError};
+use pyo3::PyTypeInfo;
+use std::collections::HashMap;
+use std::os::raw::{c_long, c_ulong};
+use std::sync::{Arc, LazyLock, Mutex, MutexGuard, PoisonError};
// Tasks create their callables concurrently. Concurrent first imports of a
package
// can hand a thread a partially initialized module (observed with PyArrow on
Python
// 3.10), so callables are unpickled one at a time. Take it with
`lock_py_attached`,
// which waits for the lock without holding the GIL.
static IMPORT_LOCK: Mutex<()> = Mutex::new(());
+pyo3::create_exception!(
+ comet,
+ TaskKilled,
+ pyo3::exceptions::PyBaseException,
+ "Raised in a running Arrow UDF when Spark kills its task."
+);
+
+unsafe extern "C" {
+ // Part of the stable Python C API, but not declared by pyo3-ffi.
+ fn PyThread_get_thread_ident() -> c_ulong;
+}
+
+/// The native plans of one Spark task attempt that hold Arrow UDFs.
+#[derive(Default)]
+struct TaskCalls {
+ plans: usize,
+ killed: bool,
+ /// The calls running now.
+ calls: Vec<Call>,
+}
+
+struct Call {
+ /// Python thread id of the thread running the call.
+ thread: c_ulong,
+ /// Whether [`interrupt_task`] has raised in the call. It raises once, so
that a second
+ /// exception does not abort the cleanup the first one started.
+ interrupted: bool,
+}
+
+// Holders of this lock never wait for the GIL, so a thread attached to Python
may take it.
+static TASKS: LazyLock<Mutex<HashMap<i64, TaskCalls>>> =
LazyLock::new(Default::default);
+
+fn tasks() -> MutexGuard<'static, HashMap<i64, TaskCalls>> {
+ TASKS.lock().unwrap_or_else(PoisonError::into_inner)
+}
+
+/// Registers a native plan's Arrow UDFs under their Spark task attempt, so
that
+/// [`interrupt_task`] reaches their calls. The task's entry is removed when
its last
+/// registration is dropped, which happens when the plan is released.
+pub struct TaskRegistration {
+ task_attempt_id: i64,
+}
+
+impl TaskRegistration {
+ pub fn new(task_attempt_id: i64) -> Self {
+ tasks().entry(task_attempt_id).or_default().plans += 1;
+ Self { task_attempt_id }
+ }
+
+ fn is_killed(&self) -> bool {
+ tasks()
+ .get(&self.task_attempt_id)
+ .is_some_and(|task| task.killed)
+ }
+
+ fn killed_error(&self) -> ArrowError {
+ ArrowError::ComputeError(format!(
+ "Arrow UDF stopped because task {} was killed",
+ self.task_attempt_id
+ ))
+ }
+}
+
+impl Drop for TaskRegistration {
+ fn drop(&mut self) {
+ let mut tasks = tasks();
+ if let Some(task) = tasks.get_mut(&self.task_attempt_id) {
+ task.plans -= 1;
+ if task.plans == 0 {
+ tasks.remove(&self.task_attempt_id);
+ }
+ }
+ }
+}
+
+/// A call to the user's callable, recorded so that [`interrupt_task`] can
raise in its thread.
+struct RunningCall<'a> {
+ task: &'a TaskRegistration,
+ thread: c_ulong,
+}
+
+impl<'a> RunningCall<'a> {
+ /// Fails without recording the call when the task has already been killed.
+ fn enter(_py: Python<'_>, task: &'a TaskRegistration) -> Result<Self> {
+ // SAFETY: the caller is attached to the interpreter.
+ let thread = unsafe { PyThread_get_thread_ident() };
+ let mut tasks = tasks();
+ let calls = tasks.entry(task.task_attempt_id).or_default();
+ if calls.killed {
+ return Err(task.killed_error());
+ }
+ calls.calls.push(Call {
+ thread,
+ interrupted: false,
+ });
+ Ok(Self { task, thread })
+ }
+}
+
+impl Drop for RunningCall<'_> {
+ fn drop(&mut self) {
+ let mut tasks = tasks();
+ let Some(calls) = tasks.get_mut(&self.task.task_attempt_id) else {
+ return;
+ };
+ let Some(index) = calls.calls.iter().position(|c| c.thread ==
self.thread) else {
+ return;
+ };
+ if calls.calls.swap_remove(index).interrupted {
+ // An interrupt that arrived as the callable returned is still
pending on this
+ // thread. Nothing can set another one now that the call is
unrecorded, so clear
+ // it rather than let it fire in later Python code on this thread.
+ // SAFETY: the call is made while attached, on the thread it names.
+ unsafe {
+ pyo3::ffi::PyThreadState_SetAsyncExc(self.thread as c_long,
std::ptr::null_mut())
+ };
+ }
+ }
+}
+
+/// Marks a Spark task attempt as killed and raises [`TaskKilled`] in its
running Arrow UDF
+/// calls that it has not interrupted yet. Later calls of the task fail
without entering
+/// Python. Returns the number of calls interrupted. A task with no registered
plan is not
+/// recorded, so the caller repeats this until the task completes.
+///
+/// The exception takes effect when the thread next runs Python bytecode: a
call blocked in
+/// native code that released the GIL is interrupted only when that code
returns.
+pub fn interrupt_task(task_attempt_id: i64) -> usize {
+ let running = match tasks().get_mut(&task_attempt_id) {
+ Some(task) => {
+ task.killed = true;
+ task.calls.iter().any(|call| !call.interrupted)
+ }
+ None => false,
+ };
+ if !running {
+ return 0;
+ }
+ // Calls cannot start after the flag is set, so only calls recorded before
it remain.
+ Python::attach(|py| {
+ let exception = TaskKilled::type_object(py);
+ let mut tasks = tasks();
+ let Some(task) = tasks.get_mut(&task_attempt_id) else {
+ return 0;
+ };
+ let mut interrupted = 0;
+ for call in task.calls.iter_mut().filter(|call| !call.interrupted) {
+ // SAFETY: attached to the interpreter, and `exception` is a live
type object.
+ let set = unsafe {
+ pyo3::ffi::PyThreadState_SetAsyncExc(call.thread as c_long,
exception.as_ptr())
Review Comment:
[P2] A first `TaskKilled` can strand a shared Python lock. If a UDF is
waiting in `logging.Handler.acquire()`, delivery can occur after it acquires
the `RLock` and before `Handler.handle()` enters its `try/finally`. The call
exits owning the lock, and healthy tasks using the same handler block. Use
explicit cooperative cancellation points in the shared interpreter, or isolate
calls that require forced interruption.
Evidence: On CPython 3.11.15 and 3.12.13, one interrupt made the cancelled
UDF return `TaskKilled` while still owning the handler lock. A peer’s
`logger.warning()` timed out after 300 ms and completed when the victim thread
released the lock. With zero interrupts, all three UDFs completed. This happens
during acquisition, before the cleanup covered by the earlier review. [CPython
handler](https://github.com/python/cpython/blob/v3.12.13/Lib/logging/__init__.py#L1025-L1030)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
