Fix GIL-deadlocks when doing async iteration by Qqwy · Pull Request #6 · channable/opsqueue

Qqwy · 2025-08-11T16:24:15Z

In rare situations we saw deadlocks happening in production, specifically when doing async iteration of the submission-output iterator (returning that Rust stream back to Python element-by-element).

This PR resolves this issue, by relinquishing the GIL more often.

Ensures the test suite fails after at most 30 seconds rather than hanging for many minutes if there is a deadlock or other hanging situation.
Acquire the GIL only per chunk when reading the Python input submission iterator
Release the GIL inside the implementation of __next__ and __anext__ of the (sync/async) output result iterator.
- Ensure that locking the Rust-async-aware internal stream mutex is never done on the Python thread but only on the Tokio runtime thread.
Document why we run Tokio on a separate thread (e.g. use the multi threaded runtime with 1 thread) rather than using the current_thread runtime: That causes deadlocks when a Python future depends on a Rust future or vice-versa since the two async runtimes do not know of each other's existence.
Ensure that for the pyo3_async_runtimes we do not use its tokio module that secretly spawns its own runtime but rather we depend on the one that we've explicitly constructed ourselves.

SemMulder

Awesome! This is great work!

The choices made here look like common sense to me, which then makes it confusing to me why upstream made different choices. 😕

This journey, and the solution in this PR, do make me wary of the use of async on the Rust side when interfacing with Python. Debugging using GDB was also not a fun experience with how Rust compiles Futures to state machines, very hard to make sense out of that, at scale. Something to warn others about who need to write Python interop stuff in the future :).

How do you see the use of async Rust in this Python library? Would it make sense to evaluate whether it's use is warranted, long-term?

Code looks good to me! I think we can improve the documentation a little bit in places though :).

SemMulder · 2025-08-12T08:56:57Z

libs/opsqueue_python/src/async_util.rs

+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let waker = cx.waker();
+        Python::with_gil(|py| {
+            py.allow_threads(|| pin!(&mut self.0).poll(&mut Context::from_waker(waker)))


pin!(&mut self.0)

The pinning is needed because of the way Rust implements Futures I guess?

&mut Context::from_waker(waker))

Can't we just forward cx directly here?

SemMulder · 2025-08-12T09:02:54Z

libs/opsqueue_python/src/async_util.rs

+/// An alternative runtime for `pyo3_async_runtimes`
+/// since `pyo3_async_runtimes::tokio` runs its _own_ Tokio runtime
+/// rather than using whatever is in scope or allowing the user to pass the specific runtime.
+///
+/// How this runtime works, is to use whatever Tokio runtime was entered using `runtime.enter()` beforehand


Oof, that pyo3_async_runtimes::tokio::future_into_py feels like it does more than what it says on the tin 😅.

Maybe add a link to this: https://docs.rs/pyo3-async-runtimes/latest/src/pyo3_async_runtimes/tokio.rs.html#188-198

libs/opsqueue_python/src/async_util.rs

SemMulder · 2025-08-12T09:14:20Z

libs/opsqueue_python/src/async_util.rs

+impl pyo3_async_runtimes::generic::ContextExt for TokioRuntimeThatIsInScope {
+    fn scope<F, R>(
+        locals: TaskLocals,
+        fut: F,
+    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = R> + Send>>
+    where
+        F: std::future::Future<Output = R> + Send + 'static,
+    {
+        let cell = UnsyncOnceCell::new();
+        cell.set(locals).unwrap();
+
+        Box::pin(TASK_LOCALS.scope(cell, fut))
+    }
+
+    fn get_task_locals() -> Option<TaskLocals> {
+        TASK_LOCALS
+            .try_with(|c| {
+                c.get()
+                    .map(|locals| Python::with_gil(|py| locals.clone_ref(py)))
+            })
+            .unwrap_or_default()
+    }
+}


Maybe add something like:

// Copied from https://docs.rs/pyo3-async-runtimes/latest/src/pyo3_async_runtimes/tokio.rs.html#100-119

SemMulder · 2025-08-12T09:18:33Z

libs/opsqueue_python/src/common.rs

+/// Note that we very intentionally use the multi-threaded scheduler
+/// but with a single thread.
+///
+/// We **cannot** use the current_thread scheduler,
+/// since that would result in the Python task scheduler (e.g. `asyncio`)
+/// and the Rust task scheduler (`Tokio`) to run on the same thread.
+/// This seems to work fine, until you end up with a Python future
+/// that depends on a Rust future completing or vice-versa:
+/// Since the task schedulers each have their own task queues,
+/// work co-operatively, and know nothing of each-other,
+/// they will not (nor can they) yield to the other.
+/// The result: deadlocks!
+///
+/// Therefore, we run the Tokio scheduler on a separate thread.
+/// Since switching between the 'Python scheduler thread' and the
+/// 'Tokio scheduler thread' is preemptive,
+/// the same problem now no longer occurs:
+/// Both schedulers are able to make forward progress (even on a 1-CPU machine).


Sanity check: is the fact that block_on executes the passed Future on the current thread problematic?

If block_on is called from a thread that also has a Python asyncio eventloop, that loop will not make progress while the thread is blocking on the Future. This might not matter for sync function calls, but it might for async calls? I think the only Python to Rust async calls we have are around PyChunksAsyncIter? Regardless, it might be worth mentioning/thinking about?

Is the fact that block_on executes the passed Future on the current thread problematic?

It is not, as long as:

We can be sure that code does not try to wait for a Python future

We won't attempt to lock a Tokio mutex in this thread

I agree that it should be added to the documentation of (our) block_on/block_until_interrupted function.

SemMulder · 2025-08-12T09:21:15Z

libs/opsqueue_python/src/producer.rs

+            async_util::async_allow_threads(Box::pin(async move {
+                match me.stream_completed_submission_chunks(submission_id).await {
+                    Ok(iter) => {
+                        let async_iter = PyChunksAsyncIter::from(iter);
+                        Ok(async_iter)
+                    }
+                    Err(e) => PyResult::Err(e.into()),
                }
-                Err(e) => PyResult::Err(e.into()),
-            }
-        })
+            })),


It's curious to me that upstream doesn't have an out-of-the box way to deal with this. Same with the GIL semantics around Iterator we implement ourselves, above.

For the sync iterator they now do, but for async iteration unfortunately not yet. There is an experimental-async feature flag that adds some more async support in pyo3 (without pyo3_async_runtimes), but __anext__ is still missing there.

For the sync iterator they now do

Is it possible for us to use that here instead of implementing it manually using std::iter::from_fn?

Or is upstream's thing for consuming a Rust iterator in Python, rather than consuming a Python iterator in Rust?

SemMulder · 2025-08-12T09:25:42Z

libs/opsqueue_python/src/producer.rs

+            runtime.block_on(async {
+                // We lock the stream in a separate Tokio task
+                // that explicitly runs on the runtime thread rather than on the main Python thread.
+                // This reduces the possibility for deadlocks even further.


This reduces the possibility for deadlocks even further.

Maybe add how this reduces the possibility of a deadlock? It costs quite a lot of brainpower for a reader to come up with that, so just putting it here explicitly saves us a bunch of brain cells ;)

Same for __anext__ below.

Qqwy · 2025-08-13T10:30:09Z

The plan: Merge this PR as-is so the deadlock-fixes are online early.
Improving the documentation further will be part of a follow-up PR.

Qqwy · 2025-08-13T10:30:14Z

@OpsBotPrime merge

…dlocks (rather than 45min later)

…ures, and ensure we can unlock the GIL in coroutines as well

…t__ methods. They should now never use the GIL inside their bodies (except maybe for momentary logging somewhere deeply inside), and locking the main Tokio mutex explicitly happens on a separate on-runtime Tokio task.

…e documentation of its construction

Approved-by: Qqwy Priority: Normal Auto-deploy: false

OpsBotPrime · 2025-08-13T10:30:21Z

Rebased as 963eaa8, waiting for CI …

OpsBotPrime · 2025-08-13T10:30:25Z

CI job 🟡 started.

Qqwy requested a review from SemMulder August 11, 2025 16:24

Qqwy mentioned this pull request Aug 11, 2025

Fix GIL deadlock #5

Closed

Qqwy requested a review from rlycx August 11, 2025 16:24

SemMulder reviewed Aug 12, 2025

View reviewed changes

SemMulder and others added 7 commits August 13, 2025 12:30

Fix paths for building and starting opsqueue during tests

6a615e0

Acquire the GIL only per chunk

f6c415e

Ensure test suite times out after 30 seconds, so we fail early on dea…

a2cd971

…dlocks (rather than 45min later)

Add new helper functions to cleanly turn Rust futures into Python fut…

b7e957a

…ures, and ensure we can unlock the GIL in coroutines as well

Clarify why we're using a 1-thread multi_threaded Tokio runtime in th…

3e0a1e2

…e documentation of its construction

Merge #6: Fix GIL-deadlocks when doing async iteration

963eaa8

Approved-by: Qqwy Priority: Normal Auto-deploy: false

Qqwy mentioned this pull request Aug 13, 2025

Improve documentation of design decisions regarding the async runtime #7

Open

OpsBotPrime force-pushed the marten/fix-gil-deadlock3 branch from b873e3e to 963eaa8 Compare August 13, 2025 10:35

OpsBotPrime merged commit 963eaa8 into master Aug 13, 2025
6 of 7 checks passed

rlycx mentioned this pull request Oct 17, 2025

CI keeps failing due to timeout #20

Open

Comments

Conversation

Qqwy commented Aug 11, 2025

Uh oh!

SemMulder left a comment

Choose a reason for hiding this comment

Uh oh!

SemMulder Aug 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SemMulder Aug 12, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

SemMulder Aug 12, 2025

Choose a reason for hiding this comment

Uh oh!

SemMulder Aug 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Qqwy Aug 12, 2025

Choose a reason for hiding this comment

Uh oh!

SemMulder Aug 12, 2025

Choose a reason for hiding this comment

Uh oh!

Qqwy Aug 12, 2025

Choose a reason for hiding this comment

Uh oh!

SemMulder Aug 12, 2025

Choose a reason for hiding this comment

Uh oh!

SemMulder Aug 12, 2025

Choose a reason for hiding this comment

Uh oh!

SemMulder Aug 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Qqwy commented Aug 13, 2025

Uh oh!

Qqwy commented Aug 13, 2025

Uh oh!

OpsBotPrime commented Aug 13, 2025

Uh oh!

OpsBotPrime commented Aug 13, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

SemMulder Aug 12, 2025 •

edited

Loading

SemMulder Aug 12, 2025 •

edited

Loading

SemMulder Aug 12, 2025 •

edited

Loading