gh-128552: fix refcycles in eager task creation

Lib/asyncio/base_events.py

@@ -463,21 +463,25 @@ def create_task(self, coro, *, name=None, context=None):
 
         Return a task object.
         """
-        self._check_closed()
-        if self._task_factory is None:
-            task = tasks.Task(coro, loop=self, name=name, context=context)
-            if task._source_traceback:
-                del task._source_traceback[-1]
-        else:
-            if context is None:
-                # Use legacy API if context is not needed
-                task = self._task_factory(self, coro)
+        task = None
+        try:
+            self._check_closed()
+            if self._task_factory is None:
+                task = tasks.Task(coro, loop=self, name=name, context=context)
+                if task._source_traceback:
+                    del task._source_traceback[-1]
             else:
-                task = self._task_factory(self, coro, context=context)
+                if context is None:
+                    # Use legacy API if context is not needed
+                    task = self._task_factory(self, coro)
+                else:
+                    task = self._task_factory(self, coro, context=context)
 
-            task.set_name(name)
+                task.set_name(name)
 
-        return task
+            return task
+        finally:
+            del task
 
     def set_task_factory(self, factory):
         """Set a task factory that will be used by loop.create_task().

Lib/asyncio/taskgroups.py

@@ -192,20 +192,24 @@ def create_task(self, coro, *, name=None, context=None):
         if self._aborting:
             coro.close()
             raise RuntimeError(f"TaskGroup {self!r} is shutting down")
-        if context is None:
-            task = self._loop.create_task(coro, name=name)
-        else:
-            task = self._loop.create_task(coro, name=name, context=context)
-
-        # optimization: Immediately call the done callback if the task is
-        # already done (e.g. if the coro was able to complete eagerly),
-        # and skip scheduling a done callback
-        if task.done():
-            self._on_task_done(task)
-        else:
-            self._tasks.add(task)
-            task.add_done_callback(self._on_task_done)
-        return task
+        task = None
+        try:
+            if context is None:
+                task = self._loop.create_task(coro, name=name)
+            else:
+                task = self._loop.create_task(coro, name=name, context=context)
+
+            # optimization: Immediately call the done callback if the task is
+            # already done (e.g. if the coro was able to complete eagerly),
+            # and skip scheduling a done callback
+            if task.done():
+                self._on_task_done(task)
+            else:
+                self._tasks.add(task)
+                task.add_done_callback(self._on_task_done)
+            return task
+        finally:
+            del task
 
     # Since Python 3.8 Tasks propagate all exceptions correctly,
     # except for KeyboardInterrupt and SystemExit which are

Lib/test/test_asyncio/test_taskgroups.py

@@ -12,11 +12,6 @@
 
 from test.test_asyncio.utils import await_without_task
 
-# To prevent a warning "test altered the execution environment"
-def tearDownModule():
-    asyncio._set_event_loop_policy(None)
-
-
 class MyExc(Exception):
     pass
 
@@ -38,7 +33,7 @@ def no_other_refs():
     return [coro]
 
 
-class TestTaskGroup(unittest.IsolatedAsyncioTestCase):
+class BaseTestTaskGroup:
 
     async def test_taskgroup_01(self):
 
@@ -832,15 +827,15 @@ async def test_taskgroup_without_parent_task(self):
         with self.assertRaisesRegex(RuntimeError, "has not been entered"):
             tg.create_task(coro)
 
-    def test_coro_closed_when_tg_closed(self):
+    async def test_coro_closed_when_tg_closed(self):
         async def run_coro_after_tg_closes():
             async with taskgroups.TaskGroup() as tg:
                 pass
             coro = asyncio.sleep(0)
             with self.assertRaisesRegex(RuntimeError, "is finished"):
                 tg.create_task(coro)
-        loop = asyncio.get_event_loop()
-        loop.run_until_complete(run_coro_after_tg_closes())
+
+        await run_coro_after_tg_closes()
 
     async def test_cancelling_level_preserved(self):
         async def raise_after(t, e):
@@ -998,5 +993,16 @@ class MyKeyboardInterrupt(KeyboardInterrupt):
         self.assertListEqual(gc.get_referrers(exc), no_other_refs())
 
 
+class TestTaskGroup(BaseTestTaskGroup, unittest.IsolatedAsyncioTestCase):
+    loop_factory = asyncio.EventLoop
+
+class TestEagerTaskTaskGroup(BaseTestTaskGroup, unittest.IsolatedAsyncioTestCase):
+    @staticmethod
+    def loop_factory():
+        loop = asyncio.EventLoop()
+        loop.set_task_factory(asyncio.eager_task_factory)
+        return loop
+
+
 if __name__ == "__main__":
     unittest.main()

Misc/NEWS.d/next/Library/2025-01-06-18-41-08.gh-issue-128552.fV-f8j.rst

@@ -0,0 +1 @@
+Fix cyclic garbage introduced by :meth:`loop.create_task` and :meth:`asyncio.TaskGroup.create_task` holding a reference to the created task.