From f83a894c276ab34a95c4b27d4e018afa93ca046f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jind=C5=99ich=20B=C3=A4r?= Date: Thu, 5 Nov 2026 11:54:40 +0100 Subject: [PATCH 1/4] fix(python): don't crash when the interpreter exits during a result delivery pyo3-async-runtimes delivers results from a tokio thread via `call_soon_threadsafe`, which releases the GIL in `socket.send`. If the awaiting coroutine was the last thing the program did, the interpreter can start finalizing before that thread gets the GIL back, and CPython < 3.13.8 kills it with `pthread_exit`. The forced unwind through Rust frames aborts the process. Deliveries now go through a custom pyo3-async-runtimes runtime that counts them. An atexit hook (which runs before the interpreter is marked as finalizing) stops new deliveries and waits for the in-flight ones to leave Python. Closes #606 --- impit-python/src/async_client.rs | 6 +- impit-python/src/exit_gate.rs | 131 +++++++++++++++++++++++++++++++ impit-python/src/lib.rs | 3 + impit-python/src/response.rs | 77 +++++++++--------- impit-python/test/exit_test.py | 52 ++++++++++++ 5 files changed, 226 insertions(+), 43 deletions(-) create mode 100644 impit-python/src/exit_gate.rs create mode 100644 impit-python/test/exit_test.py diff --git a/impit-python/src/async_client.rs b/impit-python/src/async_client.rs index 9a5a5004..990936b2 100644 --- a/impit-python/src/async_client.rs +++ b/impit-python/src/async_client.rs @@ -27,7 +27,7 @@ impl AsyncClient { slf: Py, py: Python<'_>, ) -> Result, pyo3::PyErr> { - pyo3_async_runtimes::tokio::future_into_py::<_, Py>(py, async { Ok(slf) }) + crate::exit_gate::future_into_py::<_, Py>(py, async { Ok(slf) }) } pub fn __aexit__<'python>( @@ -37,7 +37,7 @@ impl AsyncClient { _traceback: &crate::Bound<'_, crate::PyAny>, py: Python<'python>, ) -> Result, pyo3::PyErr> { - pyo3_async_runtimes::tokio::future_into_py::<_, ()>(py, async { Ok(()) }) + crate::exit_gate::future_into_py::<_, ()>(py, async { Ok(()) }) } #[new] @@ -400,7 +400,7 @@ impl AsyncClient { let stream_value = stream.unwrap_or(false); let impit = Arc::clone(&self.impit); - pyo3_async_runtimes::tokio::future_into_py::<_, ImpitPyResponse>(py, async move { + crate::exit_gate::future_into_py::<_, ImpitPyResponse>(py, async move { let response = match method_str.to_lowercase().as_str() { "get" => impit.get(url, Some(body), Some(options)).await, "post" => impit.post(url, Some(body), Some(options)).await, diff --git a/impit-python/src/exit_gate.rs b/impit-python/src/exit_gate.rs new file mode 100644 index 00000000..caad1e6b --- /dev/null +++ b/impit-python/src/exit_gate.rs @@ -0,0 +1,131 @@ +//! Keeps tokio threads out of the interpreter once it starts shutting down. +//! +//! On CPython < 3.13.8, a non-Python thread that re-takes the GIL during finalization is killed +//! with `pthread_exit`, which aborts the process when there are Rust frames on its stack. A result +//! delivery from `future_into_py` releases the GIL inside `call_soon_threadsafe` (`socket.send`), +//! so if the awaiting coroutine was the last thing the program did, the interpreter can start +//! finalizing before the delivering thread gets the GIL back. `atexit` hooks run before the +//! interpreter is marked as finalizing, so we close the gate there and wait for in-flight +//! deliveries to leave Python. + +use std::cell::OnceCell; +use std::future::Future; +use std::pin::Pin; +use std::sync::{Condvar, Mutex}; +use std::time::Duration; + +use pyo3::prelude::*; +use pyo3_async_runtimes::generic::{ContextExt, Runtime}; +use pyo3_async_runtimes::TaskLocals; + +const DRAIN_TIMEOUT: Duration = Duration::from_secs(5); + +struct State { + closed: bool, + in_flight: usize, +} + +static STATE: Mutex = Mutex::new(State { + closed: false, + in_flight: 0, +}); +static DRAINED: Condvar = Condvar::new(); + +struct InFlight; + +impl InFlight { + fn enter() -> Option { + let mut state = STATE.lock().unwrap(); + if state.closed { + return None; + } + state.in_flight += 1; + Some(InFlight) + } +} + +impl Drop for InFlight { + fn drop(&mut self) { + let mut state = STATE.lock().unwrap(); + state.in_flight -= 1; + if state.in_flight == 0 { + DRAINED.notify_all(); + } + } +} + +#[pyfunction] +fn close(py: Python<'_>) { + py.detach(|| { + let mut state = STATE.lock().unwrap(); + state.closed = true; + let _ = DRAINED.wait_timeout_while(state, DRAIN_TIMEOUT, |s| s.in_flight > 0); + }); +} + +pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { + let close = wrap_pyfunction!(close, m)?; + m.py() + .import("atexit")? + .call_method1("register", (close,))?; + Ok(()) +} + +struct GatedTokio; + +tokio::task_local! { + static TASK_LOCALS: OnceCell; +} + +impl Runtime for GatedTokio { + type JoinError = tokio::task::JoinError; + type JoinHandle = tokio::task::JoinHandle<()>; + + fn spawn(fut: F) -> Self::JoinHandle + where + F: Future + Send + 'static, + { + pyo3_async_runtimes::tokio::get_runtime().spawn(fut) + } + + // pyo3-async-runtimes delivers every result (and panic) to Python from a closure passed here. + // Once the gate is closed, the closure is dropped without attaching, and pyo3 defers the + // decrefs of the Python objects it holds. + fn spawn_blocking(f: F) -> Self::JoinHandle + where + F: FnOnce() + Send + 'static, + { + pyo3_async_runtimes::tokio::get_runtime().spawn_blocking(move || { + if let Some(_in_flight) = InFlight::enter() { + f(); + } + }) + } +} + +impl ContextExt for GatedTokio { + fn scope(locals: TaskLocals, fut: F) -> Pin + Send>> + where + F: Future + Send + 'static, + { + let cell = OnceCell::new(); + cell.set(locals).unwrap(); + Box::pin(TASK_LOCALS.scope(cell, fut)) + } + + fn get_task_locals() -> Option { + TASK_LOCALS + .try_with(|c| c.get().cloned()) + .unwrap_or_default() + } +} + +/// Drop-in replacement for `pyo3_async_runtimes::tokio::future_into_py` that goes through the +/// exit gate. +pub fn future_into_py(py: Python<'_>, fut: F) -> PyResult> +where + F: Future> + Send + 'static, + T: for<'py> IntoPyObject<'py> + Send + 'static, +{ + pyo3_async_runtimes::generic::future_into_py::(py, fut) +} diff --git a/impit-python/src/lib.rs b/impit-python/src/lib.rs index e6b89936..2617c6ba 100644 --- a/impit-python/src/lib.rs +++ b/impit-python/src/lib.rs @@ -5,6 +5,7 @@ mod async_client; mod client; mod cookies; mod errors; +mod exit_gate; mod fingerprint; mod request; mod response; @@ -20,6 +21,8 @@ fn impit(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; + exit_gate::register(m)?; + m.add("HTTPError", m.py().get_type::())?; m.add("RequestError", m.py().get_type::())?; m.add( diff --git a/impit-python/src/response.rs b/impit-python/src/response.rs index 49e833dc..911b3d69 100644 --- a/impit-python/src/response.rs +++ b/impit-python/src/response.rs @@ -109,9 +109,7 @@ impl PyResponseAsyncBytesIterator { self.content_returned = true; let content = PyBytes::new(py, &content).unbind(); let future = - pyo3_async_runtimes::tokio::future_into_py::<_, Py>(py, async move { - Ok(content) - })?; + crate::exit_gate::future_into_py::<_, Py>(py, async move { Ok(content) })?; return Ok(future); } @@ -141,44 +139,43 @@ impl PyResponseAsyncBytesIterator { let parent_response = self.parent_response.as_ref().map(|p| p.clone_ref(py)); - let future = - pyo3_async_runtimes::tokio::future_into_py::<_, Py>(py, async move { - let chunk_result = { - let mut stream_guard = stream_arc.lock().await; - if let Some(stream) = stream_guard.as_mut() { - stream.next().await - } else { - None - } - }; - match chunk_result { - Some(Ok(chunk)) => Ok(Python::attach(|py| PyBytes::new(py, &chunk).unbind())), - Some(Err(e)) => { - if let Some(parent) = parent_response { - Python::attach(|py| { - if let Ok(mut parent_ref) = parent.try_borrow_mut(py) { - parent_ref.inner_state = InnerResponseState::StreamingClosed; - parent_ref.is_stream_consumed = true; - parent_ref.is_closed = true; - } - }); - } - Err(ImpitPyError(ImpitError::from(e, None)).into()) + let future = crate::exit_gate::future_into_py::<_, Py>(py, async move { + let chunk_result = { + let mut stream_guard = stream_arc.lock().await; + if let Some(stream) = stream_guard.as_mut() { + stream.next().await + } else { + None + } + }; + match chunk_result { + Some(Ok(chunk)) => Ok(Python::attach(|py| PyBytes::new(py, &chunk).unbind())), + Some(Err(e)) => { + if let Some(parent) = parent_response { + Python::attach(|py| { + if let Ok(mut parent_ref) = parent.try_borrow_mut(py) { + parent_ref.inner_state = InnerResponseState::StreamingClosed; + parent_ref.is_stream_consumed = true; + parent_ref.is_closed = true; + } + }); } - None => { - if let Some(parent) = parent_response { - Python::attach(|py| { - if let Ok(mut parent_ref) = parent.try_borrow_mut(py) { - parent_ref.inner_state = InnerResponseState::StreamingClosed; - parent_ref.is_stream_consumed = true; - parent_ref.is_closed = true; - } - }); - } - Err(pyo3::exceptions::PyStopAsyncIteration::new_err("")) + Err(ImpitPyError(ImpitError::from(e, None)).into()) + } + None => { + if let Some(parent) = parent_response { + Python::attach(|py| { + if let Ok(mut parent_ref) = parent.try_borrow_mut(py) { + parent_ref.inner_state = InnerResponseState::StreamingClosed; + parent_ref.is_stream_consumed = true; + parent_ref.is_closed = true; + } + }); } + Err(pyo3::exceptions::PyStopAsyncIteration::new_err("")) } - })?; + } + })?; Ok(future) } @@ -295,7 +292,7 @@ impl ImpitPyResponse { } fn aclose(slf: Py, py: Python<'_>) -> PyResult> { - pyo3_async_runtimes::tokio::future_into_py(py, async move { + crate::exit_gate::future_into_py(py, async move { Self::close_async_impl(slf)?; Ok(()) }) @@ -390,7 +387,7 @@ impl ImpitPyResponse { } fn aread(slf: Py, py: Python<'_>) -> PyResult> { - pyo3_async_runtimes::tokio::future_into_py(py, async move { + crate::exit_gate::future_into_py(py, async move { let (state, response_option, content_option, is_stream_consumed) = Python::attach(|py| { let mut slf_ref = slf.borrow_mut(py); diff --git a/impit-python/test/exit_test.py b/impit-python/test/exit_test.py new file mode 100644 index 00000000..5289695a --- /dev/null +++ b/impit-python/test/exit_test.py @@ -0,0 +1,52 @@ +import subprocess +import sys +import textwrap + +# The result delivery of the last awaited call releases the GIL inside `call_soon_threadsafe` +# (`socket.send`). If the interpreter starts finalizing before the delivering tokio thread gets the +# GIL back, CPython < 3.13.8 kills that thread with `pthread_exit`, which crashes the process. +# The script below widens both windows so the race is hit on every run. +SCRIPT = textwrap.dedent(""" + import asyncio + import asyncio.selector_events as selector_events + import sys + import time + import types + + import impit + + + class SlowTeardown: + def __del__(self, _now=time.perf_counter): + end = _now() + 0.1 + while _now() < end: + pass + + + # Torn down after the interpreter is marked as finalizing. + module = types.ModuleType('_slow_teardown') + module.obj = SlowTeardown() + sys.modules[module.__name__] = module + del module + + + def _write_to_self(self, _orig=selector_events.BaseSelectorEventLoop._write_to_self, _sleep=time.sleep): + _orig(self) + _sleep(0.02) + + + selector_events.BaseSelectorEventLoop._write_to_self = _write_to_self + + + async def main(): + async with impit.AsyncClient(): + pass + + + asyncio.run(main()) +""") + + +def test_exit_right_after_await_does_not_crash() -> None: + result = subprocess.run([sys.executable, '-c', SCRIPT], capture_output=True, timeout=30, check=False) # noqa: S603 + assert result.returncode == 0, result.stderr.decode() From 6edd43a24d464e790965c8c2838aa39d28a909bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jind=C5=99ich=20B=C3=A4r?= Date: Thu, 5 Nov 2026 14:14:07 +0100 Subject: [PATCH 2/4] fix(python): don't hang async calls in atexit hooks that run after impit's The exit gate stayed closed after impit's atexit hook, so results of calls made from hooks registered before `import impit` were dropped and never arrived. Creating a future now reopens the gate, and the hook is only registered on Python < 3.13.8, which is the only range affected by the crash. --- impit-python/src/exit_gate.rs | 7 +++++++ impit-python/test/exit_test.py | 22 ++++++++++++++++++++++ 2 files changed, 29 insertions(+) diff --git a/impit-python/src/exit_gate.rs b/impit-python/src/exit_gate.rs index caad1e6b..af285c94 100644 --- a/impit-python/src/exit_gate.rs +++ b/impit-python/src/exit_gate.rs @@ -64,6 +64,10 @@ fn close(py: Python<'_>) { } pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> { + // Newer versions hang such threads instead of killing them. + if m.py().version_info() >= (3, 13, 8) { + return Ok(()); + } let close = wrap_pyfunction!(close, m)?; m.py() .import("atexit")? @@ -127,5 +131,8 @@ where F: Future> + Send + 'static, T: for<'py> IntoPyObject<'py> + Send + 'static, { + // Python code is still running, so the interpreter isn't finalizing yet. This reopens the gate + // for atexit hooks that run after ours, whose results would never arrive otherwise. + STATE.lock().unwrap().closed = false; pyo3_async_runtimes::generic::future_into_py::(py, fut) } diff --git a/impit-python/test/exit_test.py b/impit-python/test/exit_test.py index 5289695a..b69fcac0 100644 --- a/impit-python/test/exit_test.py +++ b/impit-python/test/exit_test.py @@ -50,3 +50,25 @@ async def main(): def test_exit_right_after_await_does_not_crash() -> None: result = subprocess.run([sys.executable, '-c', SCRIPT], capture_output=True, timeout=30, check=False) # noqa: S603 assert result.returncode == 0, result.stderr.decode() + + +# atexit hooks run in reverse order, so this one runs after the hook impit registers on import. +LATE_ATEXIT_SCRIPT = textwrap.dedent(""" + import asyncio + import atexit + + + async def main(): + async with impit.AsyncClient(): + pass + + + atexit.register(lambda: asyncio.run(main())) + + import impit +""") + + +def test_async_client_works_in_atexit_hook_registered_before_import() -> None: + result = subprocess.run([sys.executable, '-c', LATE_ATEXIT_SCRIPT], capture_output=True, timeout=10, check=False) # noqa: S603 + assert result.returncode == 0, result.stderr.decode() From 4a4b82131d158b150b91d4374f329678939d15e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jind=C5=99ich=20B=C3=A4r?= Date: Mon, 9 Nov 2026 10:40:50 +0100 Subject: [PATCH 3/4] chore(python): guard the exit gate with clippy instead of subprocess tests The exit tests only exercised the gate on Python < 3.13.8, which CI doesn't run, and wouldn't catch a new call site that skips the gate. A clippy ban on the ungated `future_into_py` does, on every Python version. --- impit-python/clippy.toml | 4 ++ impit-python/test/exit_test.py | 74 ---------------------------------- 2 files changed, 4 insertions(+), 74 deletions(-) delete mode 100644 impit-python/test/exit_test.py diff --git a/impit-python/clippy.toml b/impit-python/clippy.toml index 0d4e02f0..9421bd94 100644 --- a/impit-python/clippy.toml +++ b/impit-python/clippy.toml @@ -1 +1,5 @@ too-many-arguments-threshold = 20 + +disallowed-methods = [ + { path = "pyo3_async_runtimes::tokio::future_into_py", reason = "use crate::exit_gate::future_into_py, it keeps result deliveries from racing interpreter shutdown" }, +] diff --git a/impit-python/test/exit_test.py b/impit-python/test/exit_test.py deleted file mode 100644 index b69fcac0..00000000 --- a/impit-python/test/exit_test.py +++ /dev/null @@ -1,74 +0,0 @@ -import subprocess -import sys -import textwrap - -# The result delivery of the last awaited call releases the GIL inside `call_soon_threadsafe` -# (`socket.send`). If the interpreter starts finalizing before the delivering tokio thread gets the -# GIL back, CPython < 3.13.8 kills that thread with `pthread_exit`, which crashes the process. -# The script below widens both windows so the race is hit on every run. -SCRIPT = textwrap.dedent(""" - import asyncio - import asyncio.selector_events as selector_events - import sys - import time - import types - - import impit - - - class SlowTeardown: - def __del__(self, _now=time.perf_counter): - end = _now() + 0.1 - while _now() < end: - pass - - - # Torn down after the interpreter is marked as finalizing. - module = types.ModuleType('_slow_teardown') - module.obj = SlowTeardown() - sys.modules[module.__name__] = module - del module - - - def _write_to_self(self, _orig=selector_events.BaseSelectorEventLoop._write_to_self, _sleep=time.sleep): - _orig(self) - _sleep(0.02) - - - selector_events.BaseSelectorEventLoop._write_to_self = _write_to_self - - - async def main(): - async with impit.AsyncClient(): - pass - - - asyncio.run(main()) -""") - - -def test_exit_right_after_await_does_not_crash() -> None: - result = subprocess.run([sys.executable, '-c', SCRIPT], capture_output=True, timeout=30, check=False) # noqa: S603 - assert result.returncode == 0, result.stderr.decode() - - -# atexit hooks run in reverse order, so this one runs after the hook impit registers on import. -LATE_ATEXIT_SCRIPT = textwrap.dedent(""" - import asyncio - import atexit - - - async def main(): - async with impit.AsyncClient(): - pass - - - atexit.register(lambda: asyncio.run(main())) - - import impit -""") - - -def test_async_client_works_in_atexit_hook_registered_before_import() -> None: - result = subprocess.run([sys.executable, '-c', LATE_ATEXIT_SCRIPT], capture_output=True, timeout=10, check=False) # noqa: S603 - assert result.returncode == 0, result.stderr.decode() From 4af486bdf5411e307c809d4d0dc4c057b0f00f57 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jind=C5=99ich=20B=C3=A4r?= Date: Mon, 9 Nov 2026 13:53:31 +0100 Subject: [PATCH 4/4] chore(test): drop the httpbin token --- impit-node/test/basics.test.ts | 9 +-------- impit-python/test/httpbin.py | 12 ++---------- 2 files changed, 3 insertions(+), 18 deletions(-) diff --git a/impit-node/test/basics.test.ts b/impit-node/test/basics.test.ts index aa9cac99..2794cdf7 100644 --- a/impit-node/test/basics.test.ts +++ b/impit-node/test/basics.test.ts @@ -13,14 +13,7 @@ import { Server as ProxyServer } from 'proxy-chain'; function getHttpBinUrl(path: string, https?: boolean): string { https ??= true; - let url: URL; - if (process.env.APIFY_HTTPBIN_TOKEN) { - url = new URL(path, 'https://httpbin.apify.actor'); - url.searchParams.set('token', process.env.APIFY_HTTPBIN_TOKEN); - } else { - url = new URL(path, 'https://httpbin.org'); - } - + const url = new URL(path, 'https://httpbin.apify.actor'); url.protocol = https ? 'https:' : 'http:'; return url.href; } diff --git a/impit-python/test/httpbin.py b/impit-python/test/httpbin.py index aeb20499..780d30b7 100644 --- a/impit-python/test/httpbin.py +++ b/impit-python/test/httpbin.py @@ -1,20 +1,12 @@ from __future__ import annotations -import os import urllib.parse def get_httpbin_url(path: str, *, query: dict[str, list[str] | str] | None = None, https: bool = True) -> str: - query = query or {} - url = None - if os.environ.get('APIFY_HTTPBIN_TOKEN'): - url = urllib.parse.urlparse('https://httpbin.apify.actor') - query['token'] = os.environ['APIFY_HTTPBIN_TOKEN'] + url = urllib.parse.urlparse('https://httpbin.apify.actor') + if query: url = url._replace(query=urllib.parse.urlencode(query, doseq=True)) - else: - url = urllib.parse.urlparse('https://httpbin.org') - if query: - url = url._replace(query=urllib.parse.urlencode(query, doseq=True)) scheme = 'https' if https else 'http' url = url._replace(scheme=scheme)