Skip to content

Move Python::attach calls outside of tokio event loop in future_into_py #60

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

Merged
merged 1 commit into from
Sep 18, 2025
Merged

Move Python::attach calls outside of tokio event loop in future_into_py #60

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
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,10 @@ To see unreleased changes, please see the CHANGELOG on the main branch.
<!-- towncrier release notes start -->

- Avoid attaching to the runtime when cloning TaskLocals by using std::sync::Arc. [#62](https://github.com/PyO3/pyo3-async-runtimes/pull/62)
- **Breaking**: Finalize the future without holding GIL inside async-std/tokio runtime.
Trait `Runtime` now requires `spawn_blocking` function,
`future_into_py` functions now require future return type to be `Send`.
[#60](https://github.com/PyO3/pyo3-async-runtimes/pull/60)

## [0.26.0] - 2025-09-02

Expand Down
15 changes: 12 additions & 3 deletions src/async_std.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
use async_std::task;
use futures::FutureExt;
use pyo3::prelude::*;
use std::{any::Any, cell::RefCell, future::Future, panic::AssertUnwindSafe, pin::Pin};
use std::{any::Any, cell::RefCell, future::Future, panic, panic::AssertUnwindSafe, pin::Pin};

use crate::{
generic::{self, ContextExt, JoinError, LocalContextExt, Runtime, SpawnLocalExt},
Expand Down Expand Up @@ -74,6 +74,15 @@ impl Runtime for AsyncStdRuntime {
.map_err(AsyncStdJoinErr)
})
}

fn spawn_blocking<F>(f: F) -> Self::JoinHandle
where
F: FnOnce() + Send + 'static,
{
task::spawn_blocking(move || {
panic::catch_unwind(AssertUnwindSafe(f)).map_err(|e| AsyncStdJoinErr(Box::new(e)))
})
}
}

impl ContextExt for AsyncStdRuntime {
Expand Down Expand Up @@ -276,7 +285,7 @@ pub fn future_into_py_with_locals<F, T>(
) -> PyResult<Bound<PyAny>>
where
F: Future<Output = PyResult<T>> + Send + 'static,
T: for<'py> IntoPyObject<'py>,
T: for<'py> IntoPyObject<'py> + Send + 'static,
{
generic::future_into_py_with_locals::<AsyncStdRuntime, F, T>(py, locals, fut)
}
Expand Down Expand Up @@ -322,7 +331,7 @@ where
pub fn future_into_py<F, T>(py: Python, fut: F) -> PyResult<Bound<PyAny>>
where
F: Future<Output = PyResult<T>> + Send + 'static,
T: for<'py> IntoPyObject<'py>,
T: for<'py> IntoPyObject<'py> + Send + 'static,
{
generic::future_into_py::<AsyncStdRuntime, _, T>(py, fut)
}
Expand Down
113 changes: 84 additions & 29 deletions src/generic.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,11 @@ pub trait Runtime: Send + 'static {
fn spawn<F>(fut: F) -> Self::JoinHandle
where
F: Future<Output = ()> + Send + 'static;

/// Spawn a function onto this runtime's blocking event loop
fn spawn_blocking<F>(f: F) -> Self::JoinHandle
where
F: FnOnce() + Send + 'static;
}

/// Extension trait for async/await runtimes that support spawning local tasks
Expand Down Expand Up @@ -161,6 +166,10 @@ where
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -265,6 +274,10 @@ where
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -415,6 +428,10 @@ fn set_result(
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -540,6 +557,10 @@ where
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -581,7 +602,7 @@ pub fn future_into_py_with_locals<R, F, T>(
where
R: Runtime + ContextExt,
F: Future<Output = PyResult<T>> + Send + 'static,
T: for<'py> IntoPyObject<'py>,
T: for<'py> IntoPyObject<'py> + Send + 'static,
{
let (cancel_tx, cancel_rx) = oneshot::channel();

Expand All @@ -606,44 +627,50 @@ where
)
.await;

Python::attach(move |py| {
if cancelled(future_tx1.bind(py))
.map_err(dump_err(py))
.unwrap_or(false)
{
return;
}

let _ = set_result(
&locals2.event_loop(py),
future_tx1.bind(py),
result.and_then(|val| val.into_py_any(py)),
)
.map_err(dump_err(py));
});
})
.await
{
if e.is_panic() {
// We should not hold GIL inside async-std/tokio event loop,
// because a blocked task may prevent other tasks from progressing.
R::spawn_blocking(|| {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Kinda hard to tell from the diff, but the only change here is the addition of spawn_blocking?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes.

There is "hide whitespace" option in GitHub diff viewer.

Image Image

Python::attach(move |py| {
if cancelled(future_tx2.bind(py))
if cancelled(future_tx1.bind(py))
.map_err(dump_err(py))
.unwrap_or(false)
{
return;
}

let panic_message = format!(
"rust future panicked: {}",
get_panic_message(&e.into_panic())
);
let _ = set_result(
locals.0.event_loop.bind(py),
future_tx2.bind(py),
Err(RustPanic::new_err(panic_message)),
&locals2.event_loop(py),
future_tx1.bind(py),
result.and_then(|val| val.into_py_any(py)),
)
.map_err(dump_err(py));
});
});
})
.await
{
if e.is_panic() {
R::spawn_blocking(|| {
Python::attach(move |py| {
if cancelled(future_tx2.bind(py))
.map_err(dump_err(py))
.unwrap_or(false)
{
return;
}

let panic_message = format!(
"rust future panicked: {}",
get_panic_message(&e.into_panic())
);
let _ = set_result(
locals.0.event_loop.bind(py),
future_tx2.bind(py),
Err(RustPanic::new_err(panic_message)),
)
.map_err(dump_err(py));
});
});
}
}
});
Expand Down Expand Up @@ -812,6 +839,10 @@ impl PyDoneCallback {
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -844,7 +875,7 @@ pub fn future_into_py<R, F, T>(py: Python, fut: F) -> PyResult<Bound<PyAny>>
where
R: Runtime + ContextExt,
F: Future<Output = PyResult<T>> + Send + 'static,
T: for<'py> IntoPyObject<'py>,
T: for<'py> IntoPyObject<'py> + Send + 'static,
{
future_into_py_with_locals::<R, F, T>(py, get_current_locals::<R>(py)?, fut)
}
Expand Down Expand Up @@ -921,6 +952,10 @@ where
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -1126,6 +1161,10 @@ where
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -1240,6 +1279,10 @@ where
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -1389,6 +1432,10 @@ where
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -1584,6 +1631,10 @@ async def forward(gen, sender):
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down Expand Up @@ -1737,6 +1788,10 @@ where
/// # {
/// # unreachable!()
/// # }
/// #
/// # fn spawn_blocking<F>(f: F) -> Self::JoinHandle where F: FnOnce() + Send + 'static {
/// # unreachable!()
/// # }
/// # }
/// #
/// # impl ContextExt for MyCustomRuntime {
Expand Down
11 changes: 9 additions & 2 deletions src/tokio.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,13 @@ impl GenericRuntime for TokioRuntime {
fut.await;
})
}

fn spawn_blocking<F>(f: F) -> Self::JoinHandle
where
F: FnOnce() + Send + 'static,
{
get_runtime().spawn_blocking(f)
}
}

impl ContextExt for TokioRuntime {
Expand Down Expand Up @@ -318,7 +325,7 @@ pub fn future_into_py_with_locals<F, T>(
) -> PyResult<Bound<PyAny>>
where
F: Future<Output = PyResult<T>> + Send + 'static,
T: for<'py> IntoPyObject<'py>,
T: for<'py> IntoPyObject<'py> + Send + 'static,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a breaking change, right? Should we highlight that in the Changelog? What if someone wants to use this with a non-Send return type?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Practically it is uncommon to have a future which is Send, but result is not Send.

However, generic code (like generic wrappers of future_into_py) is affected, so it is worth highlighting in changelog. Will do.

{
generic::future_into_py_with_locals::<TokioRuntime, F, T>(py, locals, fut)
}
Expand Down Expand Up @@ -364,7 +371,7 @@ where
pub fn future_into_py<F, T>(py: Python, fut: F) -> PyResult<Bound<PyAny>>
where
F: Future<Output = PyResult<T>> + Send + 'static,
T: for<'py> IntoPyObject<'py>,
T: for<'py> IntoPyObject<'py> + Send + 'static,
{
generic::future_into_py::<TokioRuntime, _, T>(py, fut)
}
Expand Down
Loading