gh-159449: Fix a free-threaded asyncio race that corrupts the task lists

Lib/test/test_asyncio/test_free_threading.py

@@ -116,6 +116,63 @@ def runner():
         support.gc_collect()
         self.assertIsNone(wr())
 
+    def test_lingering_task_different_thread_finalized(self) -> None:
+        tls = threading.local()
+
+        async def never_finishes():
+            await asyncio.Future()
+
+        def make_pending_tasks(loop, count):
+            tasks = []
+            for _ in range(count):
+                task = loop.create_task(never_finishes())
+                task._log_destroy_pending = False
+                tasks.append(task)
+            return tasks
+
+        def worker(batches):
+            loop = asyncio.new_event_loop()
+            loop.set_task_factory(self.factory)
+            tls_tasks = make_pending_tasks(loop, 8)
+            for task in tls_tasks:
+                task.get_coro().close()
+            tls.tasks = tls_tasks
+            batches.append((loop, make_pending_tasks(loop, 8)))
+
+        stop = threading.Event()
+
+        def inspector():
+            loop = asyncio.new_event_loop()
+            try:
+                while not stop.is_set():
+                    asyncio.all_tasks(loop)
+            finally:
+                loop.close()
+
+        inspectors = [Thread(target=inspector) for _ in range(2)]
+        with threading_helper.start_threads(inspectors):
+            try:
+                for _ in range(15):
+                    batches = []
+                    workers = [Thread(target=worker, args=(batches,)) for _ in range(4)]
+                    with threading_helper.start_threads(workers):
+                        pass
+
+                    def cleanup(batch):
+                        loop, tasks = batch
+                        for task in tasks:
+                            task.cancel()
+                            task.get_coro().close()
+                        tasks.clear()
+                        loop.close()
+
+                    cleaners = [Thread(target=cleanup, args=(b,)) for b in batches]
+                    with threading_helper.start_threads(cleaners):
+                        pass
+                support.gc_collect()
+            finally:
+                stop.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

@@ -69,7 +69,7 @@ typedef struct TaskObj {
     struct llist_node task_node;
 #ifdef Py_GIL_DISABLED
     // thread id of the thread where this task was created
-    uintptr_t task_tid;
+    uint64_t task_tstate_id;
 #endif
 } TaskObj;
 
@@ -2185,6 +2185,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);
 }
@@ -2204,15 +2207,15 @@ static void
 unregister_task(TaskObj *task)
 {
 #ifdef Py_GIL_DISABLED
+    PyThreadState *tstate = _PyThreadState_GET();
     // check if we are in the same thread
     // if so, we can avoid locking
-    if (task->task_tid == _Py_ThreadId()) {
+    if (task->task_tstate_id == tstate->id && !tstate->_status.finalizing) {
         unregister_task_safe(task);
     }
     else {
         // we are in a different thread
         // stop the world then check and remove the task
-        PyThreadState *tstate = _PyThreadState_GET();
         _PyEval_StopTheWorld(tstate->interp);
         unregister_task_safe(task);
         _PyEval_StartTheWorld(tstate->interp);
@@ -2341,7 +2344,9 @@ _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();
+    if (self->task_node.next == NULL) {
+        self->task_tstate_id = _PyThreadState_GET()->id;
+    }
 #endif
     self->task_must_cancel = 0;
     self->task_log_destroy_pending = 1;
@@ -3987,7 +3992,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.