avoid holding a reference to Outcome in to_thread_run_sync

Author: graingert

src/trio/_core/_tests/test_run.py

@@ -33,6 +33,7 @@
     create_asyncio_future_in_new_loop,
     gc_collect_harder,
     ignore_coroutine_never_awaited_warnings,
+    no_other_refs,
     restore_unraisablehook,
     slow,
 )
@@ -2802,17 +2803,6 @@ async def spawn_tasks_in_old_nursery(task_status: _core.TaskStatus[None]) -> Non
     assert RaisesGroup(ValueError, ValueError).matches(excinfo.value.__cause__)
 
 
-if sys.version_info >= (3, 11):
-
-    def no_other_refs() -> list[object]:
-        return []
-
-else:
-
-    def no_other_refs() -> list[object]:
-        return [sys._getframe(1)]
-
-
 @pytest.mark.skipif(
     sys.implementation.name != "cpython",
     reason="Only makes sense with refcounting GC",

src/trio/_core/_tests/tutil.py

@@ -115,3 +115,14 @@ def check_sequence_matches(seq: Sequence[T], template: Iterable[T | set[T]]) ->
 def create_asyncio_future_in_new_loop() -> asyncio.Future[object]:
     with closing(asyncio.new_event_loop()) as loop:
         return loop.create_future()
+
+
+if sys.version_info >= (3, 11):
+
+    def no_other_refs() -> list[object]:
+        return []
+
+else:
+
+    def no_other_refs() -> list[object]:
+        return [sys._getframe(1)]

src/trio/_tests/test_threads.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import contextvars
+import gc
 import queue as stdlib_queue
 import re
 import sys
@@ -29,7 +30,7 @@
     sleep_forever,
 )
 from .._core._tests.test_ki import ki_self
-from .._core._tests.tutil import slow
+from .._core._tests.tutil import gc_collect_harder, no_other_refs, slow
 from .._threads import (
     active_thread_count,
     current_default_thread_limiter,
@@ -1141,3 +1142,58 @@ async def wait_no_threads_left() -> None:
 async def test_wait_all_threads_completed_no_threads() -> None:
     await wait_all_threads_completed()
     assert active_thread_count() == 0
+
+
+@pytest.mark.skipif(
+    sys.implementation.name == "pypy",
+    reason=(
+        "gc.get_referrers is broken on PyPy (see "
+        "https://github.com/pypy/pypy/issues/5075)"
+    ),
+)
+async def test_run_sync_worker_references() -> None:
+    class Foo:
+        pass
+
+    def foo(_: Foo) -> Foo:
+        return Foo()
+
+    cvar = contextvars.ContextVar[Foo]("cvar")
+    contextval = Foo()
+    arg = Foo()
+    cvar.set(contextval)
+    v = await to_thread_run_sync(foo, arg)
+
+    cvar.set(Foo())
+    gc_collect_harder()
+
+    assert gc.get_referrers(contextval) == no_other_refs()
+    assert gc.get_referrers(foo) == no_other_refs()
+    assert gc.get_referrers(arg) == no_other_refs()
+    assert gc.get_referrers(v) == no_other_refs()
+
+
+@pytest.mark.skipif(
+    sys.implementation.name == "pypy",
+    reason=(
+        "gc.get_referrers is broken on PyPy (see "
+        "https://github.com/pypy/pypy/issues/5075)"
+    ),
+)
+async def test_run_sync_workerreferences_exc() -> None:
+
+    class MyException(Exception):
+        pass
+
+    def throw() -> None:
+        raise MyException
+
+    e = None
+    try:
+        await to_thread_run_sync(throw)
+    except MyException as err:
+        e = err
+
+    gc_collect_harder()
+
+    assert gc.get_referrers(e) == no_other_refs()

src/trio/_threads.py

@@ -443,17 +443,19 @@ def abort(raise_cancel: RaiseCancelT) -> trio.lowlevel.Abort:
             msg_from_thread: outcome.Outcome[RetT] | Run[object] | RunSync[object] = (
                 await trio.lowlevel.wait_task_rescheduled(abort)
             )
-            if isinstance(msg_from_thread, outcome.Outcome):
-                return msg_from_thread.unwrap()
-            elif isinstance(msg_from_thread, Run):
-                await msg_from_thread.run()
-            elif isinstance(msg_from_thread, RunSync):
-                msg_from_thread.run_sync()
-            else:  # pragma: no cover, internal debugging guard TODO: use assert_never
-                raise TypeError(
-                    f"trio.to_thread.run_sync received unrecognized thread message {msg_from_thread!r}.",
-                )
-            del msg_from_thread
+            try:
+                if isinstance(msg_from_thread, outcome.Outcome):
+                    return msg_from_thread.unwrap()
+                elif isinstance(msg_from_thread, Run):
+                    await msg_from_thread.run()
+                elif isinstance(msg_from_thread, RunSync):
+                    msg_from_thread.run_sync()
+                else:  # pragma: no cover, internal debugging guard TODO: use assert_never
+                    raise TypeError(
+                        f"trio.to_thread.run_sync received unrecognized thread message {msg_from_thread!r}.",
+                    )
+            finally:
+                del msg_from_thread
 
 
 def from_thread_check_cancelled() -> None: