gh-159041: Fix asyncio task tracking during thread state clear

Lib/test/test_asyncio/test_free_threading.py

@@ -1,4 +1,6 @@
 import asyncio
+import contextvars
+import _thread
 import threading
 import unittest
 from threading import Thread
@@ -232,6 +234,68 @@ def factory(self, loop, coro, **kwargs):
 @unittest.skipUnless(hasattr(asyncio.tasks, "_c_all_tasks"), "requires _asyncio")
 class TestCFreeThreading(TestFreeThreading, TestCase):
 
+    def test_task_created_during_thread_state_clear_is_tracked(self):
+        """Tasks created by a thread-state finalizer must not be orphaned."""
+        loop = asyncio.new_event_loop()
+        loop_ready = threading.Event()
+        task_started = threading.Event()
+        task_finished = threading.Event()
+        thread_done = threading.Event()
+        blocker = []
+
+        async def blocked():
+            task_started.set()
+            await blocker[0]
+            task_finished.set()
+
+        def initialize_loop():
+            blocker.append(loop.create_future())
+            loop_ready.set()
+
+        def run_loop():
+            asyncio.set_event_loop(loop)
+            loop.call_soon(initialize_loop)
+            try:
+                loop.run_forever()
+            finally:
+                loop.close()
+
+        loop_thread = Thread(target=run_loop)
+        loop_thread.start()
+        try:
+            self.assertTrue(loop_ready.wait(10))
+
+            created = []
+
+            class CreatesTaskOnClear:
+                def __del__(self):
+                    task = loop.create_task(blocked())
+                    created.append(task)
+                    # loop.create_task() does not wake a loop running in
+                    # another thread.
+                    loop.call_soon_threadsafe(lambda: None)
+
+            var = contextvars.ContextVar("var")
+
+            def worker():
+                var.set(CreatesTaskOnClear())
+                thread_done.set()
+
+            _thread.start_new_thread(worker, ())
+            self.assertTrue(thread_done.wait(10))
+
+            self.assertTrue(task_started.wait(10))
+            self.assertEqual(len(created), 1)
+            self.assertIn(created[0], asyncio.all_tasks(loop))
+        finally:
+            if blocker:
+                loop.call_soon_threadsafe(blocker[0].set_result, None)
+                if task_started.is_set():
+                    self.assertTrue(task_finished.wait(10))
+            loop.call_soon_threadsafe(loop.stop)
+            loop_thread.join(10)
+            self.assertFalse(loop_thread.is_alive())
+
     def setUp(self):
         self._old_current_task = asyncio.current_task
         asyncio.current_task = asyncio.tasks.current_task = asyncio.tasks._c_current_task

Misc/NEWS.d/next/Library/2026-10-11-00-00-00.gh-issue-159041.Q5k8m2.rst

@@ -0,0 +1,2 @@
+Fix a use-after-free in the free-threaded build when an :mod:`asyncio` task is
+created by a thread-state finalizer.

Python/pystate.c

@@ -1837,13 +1837,6 @@ PyThreadState_Clear(PyThreadState *tstate)
     Py_CLEAR(((_PyThreadStateImpl *)tstate)->asyncio_running_task);
 
 
-    PyMutex_Lock(&tstate->interp->asyncio_tasks_lock);
-    // merge any lingering tasks from thread state to interpreter's
-    // tasks list
-    llist_concat(&tstate->interp->asyncio_tasks_head,
-                 &((_PyThreadStateImpl *)tstate)->asyncio_tasks_head);
-    PyMutex_Unlock(&tstate->interp->asyncio_tasks_lock);
-
     Py_CLEAR(tstate->dict);
     Py_CLEAR(tstate->async_exc);
 
@@ -1877,6 +1870,14 @@ PyThreadState_Clear(PyThreadState *tstate)
     // Finalizers above may resolve imports and create this set.
     Py_CLEAR(((_PyThreadStateImpl *)tstate)->lazy_imports);
 
+    // Finalizers run by the Py_CLEAR() calls above can create asyncio tasks.
+    // Merge the task list only after all of those finalizers have run, so
+    // those tasks are not left linked to this thread state when it is freed.
+    PyMutex_Lock(&tstate->interp->asyncio_tasks_lock);
+    llist_concat(&tstate->interp->asyncio_tasks_head,
+                 &((_PyThreadStateImpl *)tstate)->asyncio_tasks_head);
+    PyMutex_Unlock(&tstate->interp->asyncio_tasks_lock);
+
 #ifdef Py_GIL_DISABLED
     // Each thread should clear own freelists in free-threading builds.
     struct _Py_freelists *freelists = _Py_freelists_GET();