Record the owning thread state in register_task() only and rework test

Author: kumaraditya303

Lib/test/test_asyncio/test_free_threading.py

@@ -116,62 +116,58 @@ def runner():
         support.gc_collect()
         self.assertIsNone(wr())
 
-    def test_lingering_task_different_thread_finalized(self) -> None:
-        tls = threading.local()
+    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
 
-        async def never_finishes():
-            await asyncio.Future()
+            def is_running(self):
+                return False
+
+            def call_soon(self, *args, **kwargs):
+                pass
+
+        loop = InertLoop()
+        done = threading.Event()
+
+        async def coro():
+            pass
 
-        def make_pending_tasks(loop, count):
+        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(count):
-                task = loop.create_task(never_finishes())
+            for _ in range(32):
+                c = coro()
+                task = self.factory(loop, c)
+                c.close()
                 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)))
+            handoff.append(tasks)
 
-        stop = threading.Event()
+        def spawner():
+            handoff = []
+            for _ in range(100):
+                thread = Thread(target=worker, args=(handoff,))
+                thread.start()
+                thread.join()
 
         def inspector():
-            loop = asyncio.new_event_loop()
-            try:
-                while not stop.is_set():
-                    asyncio.all_tasks(loop)
-            finally:
-                loop.close()
+            while not done.is_set():
+                asyncio.all_tasks(loop)
 
         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()
+                spawners = [Thread(target=spawner) for _ in range(4)]
+                with threading_helper.start_threads(spawners):
+                    pass
             finally:
-                stop.set()
+                done.set()
 
     def test_run_coroutine_threadsafe(self) -> None:
         results = []

Modules/_asynciomodule.c

@@ -68,7 +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
+    // id of the thread state whose asyncio_tasks_head the task was
+    // registered on; unlike _Py_ThreadId() it is never reused
     uint64_t task_tstate_id;
 #endif
 } TaskObj;
@@ -2208,14 +2209,14 @@ 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_tstate_id == tstate->id && !tstate->_status.finalizing) {
+    // 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
+        // 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);
@@ -2343,11 +2344,6 @@ _asyncio_Task___init___impl(TaskObj *self, PyObject *coro, PyObject *loop,
     }
 
     Py_CLEAR(self->task_fut_waiter);
-#ifdef Py_GIL_DISABLED
-    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;
     self->task_num_cancels_requested = 0;