diff --git a/Lib/test/test_asyncio/test_free_threading.py b/Lib/test/test_asyncio/test_free_threading.py index c60b61c7d41115f..85b74a8b1e9b528 100644 --- a/Lib/test/test_asyncio/test_free_threading.py +++ b/Lib/test/test_asyncio/test_free_threading.py @@ -116,6 +116,72 @@ def runner(): support.gc_collect() self.assertIsNone(wr()) + def test_lingering_task_different_thread_finalized(self) -> None: + # When a thread exits with pending tasks, PyThreadState_Clear() moves + # them onto interp->asyncio_tasks_head. Unregistering those tasks from + # later threads (which may reuse the exited thread's OS thread ID) or + # during PyThreadState_Clear() while all_tasks() walks the task lists + # must not corrupt the shared list. + 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) + # Tasks held in thread-local storage are cleared by PyThreadState_Clear + # after llist_concat moves tstate->asyncio_tasks_head to interp. + tls_tasks = make_pending_tasks(loop, 8) + for task in tls_tasks: + task.get_coro().close() + tls.tasks = tls_tasks + # Tasks returned in batches stay alive until a subsequent worker + # thread (often reusing this thread's OS thread ID) unregisters them. + 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 = [] diff --git a/Misc/NEWS.d/next/Core_and_Builtins/2026-10-15-10-15-27.gh-issue-159449.xsdzFf.rst b/Misc/NEWS.d/next/Core_and_Builtins/2026-10-15-10-15-27.gh-issue-159449.xsdzFf.rst new file mode 100644 index 000000000000000..f592e76fdbff707 --- /dev/null +++ b/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. diff --git a/Modules/_asynciomodule.c b/Modules/_asynciomodule.c index 18e731336d6da1d..155e343988a2236 100644 --- a/Modules/_asynciomodule.c +++ b/Modules/_asynciomodule.c @@ -68,8 +68,9 @@ 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 (PyThreadState.id) of the thread state that created this + // task, see unregister_task() + uint64_t task_tstate_id; #endif } TaskObj; @@ -2151,6 +2152,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); } @@ -2170,15 +2174,19 @@ 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 lock-free path is only safe while the task is still linked into the + // current thread's own list. That holds if this thread state created the + // task and has not started PyThreadState_Clear(), which moves lingering + // tasks onto the shared interp->asyncio_tasks_head list. Compare thread + // state ids rather than _Py_ThreadId(), which is reused by threads that + // start after the creating thread has exited. + 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(); + // the task may be linked into another thread's list or into the + // interpreter's list: stop the world then check and remove the task _PyEval_StopTheWorld(tstate->interp); unregister_task_safe(task); _PyEval_StartTheWorld(tstate->interp); @@ -2307,7 +2315,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; @@ -3948,7 +3958,10 @@ 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 task may already be done but not yet unlinked: its state is set + // before unregister_task() removes it, and that removal may be + // waiting for a stop-the-world pause. add_one_task() filters out + // tasks that are done. // The linked list holds borrowed references to task // as such it is possible that the task is concurrently // deallocated while added to this list.