gh-159449: Fix free-threaded asyncio race that corrupts the task lists (GH-159483)

(cherry picked from commit aaaf61c)

Co-authored-by: Jean-Baptiste Lespiau <jeanbaptiste.lespiau@gmail.com>
Co-authored-by: Kumar Aditya <kumaraditya@python.org>

Author: 2 people

Lib/test/test_asyncio/test_free_threading.py

@@ -116,6 +116,59 @@ def runner():
         support.gc_collect()
         self.assertIsNone(wr())
 
+    def test_unregister_lingering_tasks_of_exited_thread(self) -> None:
+        # gh-159449: a new thread can get the same OS thread id as
+        # an exited thread whose tasks linger on the interpreter's list.
+        class InertLoop:
+            def get_debug(self):
+                return False
+
+            def is_running(self):
+                return False
+
+            def call_soon(self, *args, **kwargs):
+                pass
+
+        loop = InertLoop()
+        done = threading.Event()
+
+        async def coro():
+            pass
+
+        def worker(handoff):
+            while handoff:
+                for task in handoff.pop():
+                    # complete the task right away, unlike Task.cancel()
+                    asyncio.Future.cancel(task)
+            tasks = []
+            for _ in range(32):
+                c = coro()
+                task = self.factory(loop, c)
+                c.close()
+                task._log_destroy_pending = False
+                tasks.append(task)
+            handoff.append(tasks)
+
+        def spawner():
+            handoff = []
+            for _ in range(100):
+                thread = Thread(target=worker, args=(handoff,))
+                thread.start()
+                thread.join()
+
+        def inspector():
+            while not done.is_set():
+                asyncio.all_tasks(loop)
+
+        inspectors = [Thread(target=inspector) for _ in range(2)]
+        with threading_helper.start_threads(inspectors):
+            try:
+                spawners = [Thread(target=spawner) for _ in range(4)]
+                with threading_helper.start_threads(spawners):
+                    pass
+            finally:
+                done.set()
+
     def test_run_coroutine_threadsafe(self) -> None:
         results = []
 

Misc/NEWS.d/next/Core_and_Builtins/2026-10-15-10-15-27.gh-issue-159449.xsdzFf.rst

@@ -0,0 +1 @@
+Fix a free-threaded asyncio race leading to segfaults.

Modules/_asynciomodule.c

@@ -68,8 +68,8 @@ typedef struct TaskObj {
     PyObject *task_context;
     struct llist_node task_node;
 #ifdef Py_GIL_DISABLED
-    // thread id of the thread where this task was created
-    uintptr_t task_tid;
+    // unique id of the thread state where the task was registered
+    uint64_t task_tstate_id;
 #endif
 } TaskObj;
 
@@ -2195,6 +2195,9 @@ register_task(_PyThreadStateImpl *ts, TaskObj *task)
         assert(task->task_node.prev != NULL);
         return;
     }
+#ifdef Py_GIL_DISABLED
+    task->task_tstate_id = ts->base.id;
+#endif
     struct llist_node *head = &ts->asyncio_tasks_head;
     llist_insert_tail(head, &task->task_node);
 }
@@ -2214,15 +2217,15 @@ static void
 unregister_task(TaskObj *task)
 {
 #ifdef Py_GIL_DISABLED
-    // check if we are in the same thread
-    // if so, we can avoid locking
-    if (task->task_tid == _Py_ThreadId()) {
+    PyThreadState *tstate = _PyThreadState_GET();
+    // The task can only be on this thread's list if this thread state
+    // registered it, in which case no locking is needed.
+    if (task->task_tstate_id == tstate->id) {
         unregister_task_safe(task);
     }
     else {
-        // we are in a different thread
-        // stop the world then check and remove the task
-        PyThreadState *tstate = _PyThreadState_GET();
+        // The task is on another thread's list or on the interpreter's
+        // list of lingering tasks: stop the world to remove it.
         _PyEval_StopTheWorld(tstate->interp);
         unregister_task_safe(task);
         _PyEval_StartTheWorld(tstate->interp);
@@ -2350,9 +2353,6 @@ _asyncio_Task___init___impl(TaskObj *self, PyObject *coro, PyObject *loop,
     }
 
     Py_CLEAR(self->task_fut_waiter);
-#ifdef Py_GIL_DISABLED
-    self->task_tid = _Py_ThreadId();
-#endif
     self->task_must_cancel = 0;
     self->task_log_destroy_pending = 1;
     self->task_num_cancels_requested = 0;
@@ -3994,7 +3994,6 @@ add_tasks_llist(struct llist_node *head, PyListObject *tasks)
     struct llist_node *node;
     llist_for_each_safe(node, head) {
         TaskObj *task = llist_data(node, TaskObj, task_node);
-        assert(task->task_state == STATE_PENDING);
         // The linked list holds borrowed references to task
         // as such it is possible that the task is concurrently
         // deallocated while added to this list.