From 2fee5bc7d12f189553eda065c684bb6c849e9e19 Mon Sep 17 00:00:00 2001 From: Yehor Smoliakov Date: Sat, 10 Oct 2026 13:27:14 -0700 Subject: [PATCH 1/2] Fix greenlet re-entry wakeups and closed transport protocol leaks --- docs/async-ecosystem-tests.md | 31 +++++ src/engine/loop_core.rs | 16 ++- src/transport/process/core_protocol.rs | 16 ++- src/transport/stream/core_protocol.rs | 27 +++- src/transport/stream/protocol.rs | 19 +++ tests/ecosystem/test_greenlet_reentry.py | 97 +++++++++++++++ tests/test_transport_protocol_release.py | 150 +++++++++++++++++++++++ 7 files changed, 343 insertions(+), 13 deletions(-) create mode 100644 tests/ecosystem/test_greenlet_reentry.py create mode 100644 tests/test_transport_protocol_release.py diff --git a/docs/async-ecosystem-tests.md b/docs/async-ecosystem-tests.md index baf81d7..81ef478 100644 --- a/docs/async-ecosystem-tests.md +++ b/docs/async-ecosystem-tests.md @@ -85,6 +85,37 @@ aiosqlite 0.22.1 and uvloop 0.23.0. These cases did not reproduce a new rsloop defect; no production implementation change was needed. See the PR checks for platform-specific execution results. +## Greenlet re-entry and transport ownership regressions + +`tests/ecosystem/test_greenlet_reentry.py` covers the two-run lifecycle used by +ASGI servers: start background tasks, then re-enter the loop from a deeper C +stack. It tests a minimal greenlet bridge and SQLAlchemy's `greenlet_spawn` at +depths 0, 2, 4, 8, 50 and 100. Each case runs in a subprocess with a deadline and +faulthandler, verifies progress by both contending tasks, and verifies cancellation +completes. No database is required. The active ready queue lives on the heap so +greenlet stack copying cannot invalidate its thread-local pointer. + +`tests/test_transport_protocol_release.py` checks that closed TCP and subprocess +transports release protocols that retain their transport, including when +`connection_lost` raises. It also checks release of `StreamReaderProtocol` and its +cached reader. These tests use weak references and garbage collection rather than +RSS, which can vary with allocator behavior. Stream teardown clears the protocol, +cached bound methods, fast-reader references and saved context after delivering +`connection_lost`; `get_protocol()` then returns `None`. This breaks the closed +connection cycle without making native transports cyclic-GC types. + +Run both regression groups with: + +```sh +uv run --no-sync python scripts/run_python_tests.py \ + tests/test_transport_protocol_release.py tests/ecosystem/test_greenlet_reentry.py -m '' +``` + +These regressions reproduce the scheduling and ownership mechanisms reported in +[issue #115](https://github.com/RustedBytes/rsloop/issues/115) and +[issue #116](https://github.com/RustedBytes/rsloop/issues/116). They do not replace +end-to-end Granian/PostgreSQL load or shutdown testing. + ## PostgreSQL contracts `tests/ecosystem/test_postgresql.py` adds 36 cases: asyncpg and SQLAlchemy diff --git a/src/engine/loop_core.rs b/src/engine/loop_core.rs index a056038..5d3a671 100644 --- a/src/engine/loop_core.rs +++ b/src/engine/loop_core.rs @@ -723,8 +723,10 @@ impl LoopCore { ensure_running_loop(py, &loop_obj)?; self.mark_runtime_thread(); let mut local_timers = LocalTimers::new(self); - let mut local_ready = VecDeque::new(); - self.install_local_ready_queue(&mut local_ready); + // Greenlets can copy this run's C stack out while a callback executes. + // TLS must point to heap storage, not a queue header on that stack. + let mut local_ready = Box::new(VecDeque::new()); + self.install_local_ready_queue(&mut *local_ready); let mut pending_signal_error: Option = None; let mut ready_batch = VecDeque::new(); @@ -787,7 +789,7 @@ impl LoopCore { // hot stream of locally-scheduled callbacks. if !local_ready.is_empty() { if ready_batch.is_empty() { - std::mem::swap(&mut ready_batch, &mut local_ready); + std::mem::swap(&mut ready_batch, &mut *local_ready); } else { ready_batch.extend(local_ready.drain(..)); } @@ -1686,9 +1688,11 @@ impl LoopCore { return Err(item); } - // SAFETY: `ready` points to run_forever's stack-local queue on this - // thread. Neither callback invocation nor runtime polling holds a - // mutable reference to it across this call. + // SAFETY: `ready` points to run_forever's heap-allocated queue on + // this thread, stable even across greenlet stack switches. Neither + // callback invocation nor runtime polling holds a mutable reference + // to it across this call. TLS is cleared before the allocation + // drops. unsafe { (*ready).push_back(item) }; if !tls.drain_active.get() { // I/O futures are polled with Python detached, but still on diff --git a/src/transport/process/core_protocol.rs b/src/transport/process/core_protocol.rs index d419446..bf7e003 100644 --- a/src/transport/process/core_protocol.rs +++ b/src/transport/process/core_protocol.rs @@ -256,8 +256,20 @@ impl ProcessTransportCore { let arg = exc .map(|err| err.value(py).clone().unbind().into_any()) .unwrap_or_else(|| py.None()); - self.call_protocol_method1(py, "connection_lost", arg)?; - Ok(()) + let result = self.call_protocol_method1(py, "connection_lost", arg); + // Subprocess protocols commonly store their transport too. Release + // references even if the callback fails, outside the state lock so + // Python finalizers can safely reenter the transport. + let released = { + let mut state = self.state.lock().expect("poisoned process state"); + state.context_needs_run = false; + ( + std::mem::replace(&mut state.protocol, py.None()), + std::mem::replace(&mut state.context, py.None()), + ) + }; + drop(released); + result.map(|_| ()) } #[cfg_attr( diff --git a/src/transport/stream/core_protocol.rs b/src/transport/stream/core_protocol.rs index efbf9d8..a726c53 100644 --- a/src/transport/stream/core_protocol.rs +++ b/src/transport/stream/core_protocol.rs @@ -485,14 +485,15 @@ impl StreamTransportCore { ) }; - if let Some(fast_path) = fast_path.as_ref() { - fast_path.connection_lost(py, exc)?; + let result = if let Some(fast_path) = fast_path.as_ref() { + fast_path.connection_lost(py, exc) } else { let arg = exc .map(|err| err.value(py).clone().unbind().into_any()) .unwrap_or_else(|| py.None()); - self.call_protocol_method1(py, &callback, &context, context_needs_run, arg)?; - } + self.call_protocol_method1(py, &callback, &context, context_needs_run, arg) + .map(|_| ()) + }; // Preserve asyncio's post-close get_extra_info("socket") behavior // without retaining the live shared owner through Python reference @@ -504,6 +505,22 @@ impl StreamTransportCore { if let Some(server) = server.and_then(|weak| weak.upgrade()) { server.connection_lost(); } - Ok(()) + // Even a failing connection_lost must break protocol -> transport -> + // protocol cycles. Bound methods and fast-reader caches also own Python + // references. Drop them outside the mutex: finalizers may reenter us. + let released = { + let mut state = self.state.lock().expect("poisoned transport state"); + state.context_needs_run = false; + ( + std::mem::replace(&mut state.protocol, py.None()), + std::mem::replace( + &mut state.callbacks, + super::protocol::ProtocolCallbacks::cleared(py), + ), + std::mem::replace(&mut state.context, py.None()), + ) + }; + drop(released); + result } } diff --git a/src/transport/stream/protocol.rs b/src/transport/stream/protocol.rs index 43583af..c95743e 100644 --- a/src/transport/stream/protocol.rs +++ b/src/transport/stream/protocol.rs @@ -36,6 +36,25 @@ pub(super) struct ProtocolCallbacks { pub(super) stream_reader_fast_path: Option, } +impl ProtocolCallbacks { + /// No protocol callbacks may run after connection_lost has been delivered. + /// Release cached bound methods and fast readers as well as the protocol. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "ProtocolCallbacks"))] + pub(super) fn cleared(py: Python<'_>) -> Self { + Self { + connection_made: py.None(), + data_received: None, + eof_received: None, + connection_lost: py.None(), + pause_writing: py.None(), + resume_writing: py.None(), + get_buffer: None, + buffer_updated: None, + stream_reader_fast_path: None, + } + } +} + pub(super) enum StreamReaderFastPath { Native { protocol: Py, diff --git a/tests/ecosystem/test_greenlet_reentry.py b/tests/ecosystem/test_greenlet_reentry.py new file mode 100644 index 0000000..a19936b --- /dev/null +++ b/tests/ecosystem/test_greenlet_reentry.py @@ -0,0 +1,97 @@ +"""Exercise greenlet stack copying in a subprocess: regressions can segfault.""" + +import asyncio +import subprocess +import sys +from pathlib import Path + +import pytest +import rsloop + +pytestmark = pytest.mark.ecosystem + + +def _exercise(depth, bridge): + import greenlet + + class AsyncGreenlet(greenlet.greenlet): + def __init__(self, fn, driver): + super().__init__(fn, driver) + self.driver = driver + + def await_only(awaitable): + return greenlet.getcurrent().driver.switch(awaitable) + + async def spawn(fn): + child = AsyncGreenlet(fn, greenlet.getcurrent()) + result = child.switch() + while not child.dead: + try: + value = await result + except BaseException: # noqa: BLE001 - forward cancellation into the greenlet + result = child.throw(*sys.exc_info()) + else: + result = child.switch(value) + return result + + if bridge == "sqlalchemy": + from sqlalchemy.util.concurrency import await_only, greenlet_spawn + + spawn = greenlet_spawn + + lock = asyncio.Lock() + progress = [0, 0] + + def critical_section(): + await_only(lock.acquire()) + try: + await_only(asyncio.sleep(0.001)) + finally: + lock.release() + + async def worker(index): + while True: + await spawn(critical_section) + progress[index] += 1 + await asyncio.sleep(0) + + async def start(): + return [asyncio.create_task(worker(i)) for i in range(2)] + + def deeper(n, fn): + # map.__next__ deliberately adds C frames; a generator changes the repro. + return fn() if n == 0 else next(map(lambda _: deeper(n - 1, fn), [None])) # noqa: C417 + + loop = rsloop.new_event_loop() + asyncio.set_event_loop(loop) + tasks = loop.run_until_complete(start()) + deeper(depth, lambda: loop.run_until_complete(asyncio.sleep(0.15))) + assert min(progress) > 1, progress + for task in tasks: + task.cancel() + loop.run_until_complete( + asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), 2) + ) + assert all(task.cancelled() for task in tasks) + loop.close() + asyncio.set_event_loop(None) + + +@pytest.mark.parametrize("depth", [0, 2, 4, 8, 50, 100]) +@pytest.mark.parametrize("bridge", ["minimal", "sqlalchemy"]) +def test_greenlet_wakeup_after_deeper_loop_reentry(depth, bridge): + pytest.importorskip("greenlet") + if bridge == "sqlalchemy": + pytest.importorskip("sqlalchemy") + script = ( + f"import runpy; runpy.run_path({str(Path(__file__))!r})" + f"['_exercise']({depth}, {bridge!r})" + ) + result = subprocess.run( + [sys.executable, "-X", "faulthandler", "-c", script], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + assert result.returncode == 0, result.stdout + result.stderr diff --git a/tests/test_transport_protocol_release.py b/tests/test_transport_protocol_release.py new file mode 100644 index 0000000..5083163 --- /dev/null +++ b/tests/test_transport_protocol_release.py @@ -0,0 +1,150 @@ +"""Closed transports must release protocols that keep their transport alive.""" + +import asyncio +import gc +import sys +import weakref + +import pytest +import rsloop + + +@pytest.mark.parametrize("close_method", ["close", "abort", "peer"]) +@pytest.mark.parametrize("raise_on_lost", [False, True]) +def test_closed_transport_releases_protocol(close_method, raise_on_lost): + async def main(): + loop = asyncio.get_running_loop() + loop.set_exception_handler(lambda *_: None) + peers = [] + accepted = asyncio.Queue() + + class Peer(asyncio.Protocol): + def connection_made(self, transport): + peers.append(transport) + accepted.put_nowait(transport) + + class Client(asyncio.Protocol): + def __init__(self): + self.lost = loop.create_future() + + def connection_made(self, transport): + self.transport = transport + + def connection_lost(self, exc): + assert self.transport.get_protocol() is self + self.lost.set_result(None) + if raise_on_lost: + raise RuntimeError("intentional connection_lost failure") + + server = await loop.create_server(Peer, "127.0.0.1", 0) + refs = [] + transports = [] + try: + for _ in range(20): + transport, protocol = await loop.create_connection( + Client, "127.0.0.1", server.sockets[0].getsockname()[1] + ) + peer = await asyncio.wait_for(accepted.get(), 5) + refs.append(weakref.ref(protocol)) + transports.append(transport) + if close_method == "peer": + peer.close() + else: + getattr(transport, close_method)() + await asyncio.wait_for(protocol.lost, 5) + del protocol, transport + await asyncio.sleep(0) + gc.collect() + assert all(ref() is None for ref in refs) + assert all(transport.get_protocol() is None for transport in transports) + finally: + for transport in peers: + transport.close() + server.close() + await server.wait_closed() + + loop = rsloop.new_event_loop() + try: + loop.run_until_complete(main()) + finally: + loop.close() + + +def test_closed_stream_reader_releases_fast_path_references(): + async def main(): + loop = asyncio.get_running_loop() + peers = [] + + class Peer(asyncio.Protocol): + def connection_made(self, transport): + peers.append(transport) + transport.write(b"reply") + + server = await loop.create_server(Peer, "127.0.0.1", 0) + try: + reader = asyncio.StreamReader() + protocol = asyncio.StreamReaderProtocol(reader) + transport, _ = await loop.create_connection( + lambda protocol=protocol: protocol, + "127.0.0.1", + server.sockets[0].getsockname()[1], + ) + assert await asyncio.wait_for(reader.readexactly(5), 5) == b"reply" + reader_ref, protocol_ref = weakref.ref(reader), weakref.ref(protocol) + transport.close() + await asyncio.wait_for(protocol._get_close_waiter(transport), 5) + del reader, protocol, _ + await asyncio.sleep(0) + gc.collect() + assert reader_ref() is None + assert protocol_ref() is None + assert transport.get_protocol() is None + finally: + for peer in peers: + peer.close() + server.close() + await server.wait_closed() + + loop = rsloop.new_event_loop() + try: + loop.run_until_complete(main()) + finally: + loop.close() + + +@pytest.mark.parametrize("raise_on_lost", [False, True]) +def test_exited_subprocess_releases_protocol(raise_on_lost): + async def main(): + loop = asyncio.get_running_loop() + loop.set_exception_handler(lambda *_: None) + + class Client(asyncio.SubprocessProtocol): + def __init__(self): + self.lost = loop.create_future() + + def connection_made(self, transport): + self.transport = transport + + def connection_lost(self, exc): + self.lost.set_result(None) + if raise_on_lost: + raise RuntimeError("intentional connection_lost failure") + + transport, protocol = await loop.subprocess_exec( + Client, sys.executable, "-c", "pass" + ) + try: + ref = weakref.ref(protocol) + await asyncio.wait_for(protocol.lost, 5) + del protocol + await asyncio.sleep(0) + gc.collect() + assert ref() is None + finally: + transport.close() + + loop = rsloop.new_event_loop() + try: + loop.run_until_complete(main()) + finally: + loop.close() From 39553c6a517866975d1020f8e29c86e15f93e440 Mon Sep 17 00:00:00 2001 From: Yehor Smoliakov Date: Sat, 10 Oct 2026 13:31:07 -0700 Subject: [PATCH 2/2] Fix Pyright types in greenlet and transport regression tests --- tests/ecosystem/test_greenlet_reentry.py | 7 ++++--- tests/test_transport_protocol_release.py | 10 ++++++---- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/tests/ecosystem/test_greenlet_reentry.py b/tests/ecosystem/test_greenlet_reentry.py index a19936b..507a13f 100644 --- a/tests/ecosystem/test_greenlet_reentry.py +++ b/tests/ecosystem/test_greenlet_reentry.py @@ -4,6 +4,7 @@ import subprocess import sys from pathlib import Path +from typing import cast import pytest import rsloop @@ -20,7 +21,7 @@ def __init__(self, fn, driver): self.driver = driver def await_only(awaitable): - return greenlet.getcurrent().driver.switch(awaitable) + return cast(AsyncGreenlet, greenlet.getcurrent()).driver.switch(awaitable) async def spawn(fn): child = AsyncGreenlet(fn, greenlet.getcurrent()) @@ -28,8 +29,8 @@ async def spawn(fn): while not child.dead: try: value = await result - except BaseException: # noqa: BLE001 - forward cancellation into the greenlet - result = child.throw(*sys.exc_info()) + except BaseException as exc: # noqa: BLE001 - forward cancellation into the greenlet + result = child.throw(type(exc), exc, exc.__traceback__) else: result = child.switch(value) return result diff --git a/tests/test_transport_protocol_release.py b/tests/test_transport_protocol_release.py index 5083163..566c5e9 100644 --- a/tests/test_transport_protocol_release.py +++ b/tests/test_transport_protocol_release.py @@ -4,6 +4,7 @@ import gc import sys import weakref +from typing import cast import pytest import rsloop @@ -78,7 +79,7 @@ async def main(): class Peer(asyncio.Protocol): def connection_made(self, transport): peers.append(transport) - transport.write(b"reply") + cast(asyncio.Transport, transport).write(b"reply") server = await loop.create_server(Peer, "127.0.0.1", 0) try: @@ -91,9 +92,10 @@ def connection_made(self, transport): ) assert await asyncio.wait_for(reader.readexactly(5), 5) == b"reply" reader_ref, protocol_ref = weakref.ref(reader), weakref.ref(protocol) - transport.close() - await asyncio.wait_for(protocol._get_close_waiter(transport), 5) - del reader, protocol, _ + writer = asyncio.StreamWriter(transport, protocol, reader, loop) + writer.close() + await asyncio.wait_for(writer.wait_closed(), 5) + del reader, protocol, writer, _ await asyncio.sleep(0) gc.collect() assert reader_ref() is None