simplify and comment

Author: graingert

Lib/asyncio/base_events.py

@@ -463,24 +463,25 @@ def create_task(self, coro, *, name=None, context=None):
 
         Return a task object.
         """
-        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]
+        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)
             else:
-                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 = self._task_factory(self, coro, context=context)
 
-                task.set_name(name)
+            task.set_name(name)
 
+        try:
             return task
         finally:
+            # gh-128552: prevent a refcycle of
+            # task.exception().__traceback__->TaskGroup.create_task->task
             del task
 
     def set_task_factory(self, factory):

Lib/asyncio/taskgroups.py

@@ -192,23 +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")
-        task = None
+        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)
         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:
+            # gh-128552: prevent a refcycle of
+            # task.exception().__traceback__->TaskGroup.create_task->task
             del task
 
     # Since Python 3.8 Tasks propagate all exceptions correctly,