Interrupt running code from another thread by raising an exception, similar to how Python signal handlers work

[sbstp]

I'm trying to reproduce the behavior of Python signal handlers when an exception is raised inside of them, from Rust. It will raise the exception as soon as possible in whatever code is currently running. I have a little demo setup here, but it's not working. The Python loop is not interrupted with the exception injected by restore.

use pyo3::{Python, exceptions::PyTypeError, PyErr};

#[tokio::main]
async fn main() -> eyre::Result<()> {
    tokio::task::spawn_blocking(move || {
        Python::with_gil(|py| {
            py.run(r#"
import time;
for i in range(15):
  time.sleep(1)
print('post sleep python')"#, None, None).expect("");
            println!("post sleep rust");
        })
    });

    tokio::signal::ctrl_c().await?;
    Python::with_gil(|py| {
        println!("attempting to throw");
        PyTypeError::new_err("Error").restore(py);
        assert!(PyErr::occurred(py));
        drop(PyErr::fetch(py));
        println!("thrown");
    });

    Ok(())
}

The output looks like this:

^Cattempting to throw
thrown
post sleep python
post sleep rust

No exception is raised, and the Python code completes after the loop is finished. Is it possible to get this running properly?

[birkenfeld]

Why are you doing a PyErr:: fetch which clears the exception from the interpreter state? Does it have differently if you leave that line out?

[sbstp]

I copied this from the example here, https://pyo3.rs/v0.20.0/exception#raising-an-exception

Removing the fetch/drop does not change the behavior.

[birkenfeld]

Like the text says, that example shows how to "write and fetch" exceptions, it's probably not the best example for the chapter heading.

But I have to apologize, I just now looked closer at your example and I didn't consider before that Tokio's spawn_blocking must obviously run its closure in a thread pool, so the raising thread is not the thread you want an exception to be run in.

This can be done but is not encouraged. You'd need to use PyThreadState_SetAsyncExc like shown here but it is not wrapped in PyO3, so you need to work with the raw ffi binding.

A much better solution would be to use a flag, event, queue or other synchronization method to signal the other thread to stop.

[sbstp]

Thanks for the answer. I'm working on a task broker, similar to celery, but specialized for the field I work in. We want to use signals and exceptions to make writing worker tasks easy & avoid having to regularly check if the task has been cancelled in the worker task code. We are using 1 process per task run so using signals won't mess with other tasks.

For now I will probably stick to putting the signal handler in the Python code than in the Rust code, but if needed I'll use PyThreadState_SetAsyncExc. Thanks again!