gen: Hold strong references to all asyncio.Tasks

Per the warning in the asyncio documentation, we need to hold a strong
reference to all asyncio Tasks to prevent premature GC. Following
discussions in cpython (python/cpython#91887),
we hold these references on the IOLoop instance to ensure that they are
strongly held but do not cause leaks if the event loop itself is
discarded.

This is expected to fix all of the various "task was destroyed but
it is pending" warnings that have been reported. The
IOLoop._pending_tasks set is expected to become obsolete if
corresponding changes are made to asyncio in Python 3.13.

Fixes #3209
Fixes #3047
Fixes #2763

Some issues involve this warning as their most visible symptom,
but have an underlying cause that should still be addressed.
Updates #2914
Updates #2356

tornado/gen.py

@@ -840,13 +840,17 @@ def handle_exception(
             return False
 
 
-# Convert Awaitables into Futures.
-try:
-    _wrap_awaitable = asyncio.ensure_future
-except AttributeError:
-    # asyncio.ensure_future was introduced in Python 3.4.4, but
-    # Debian jessie still ships with 3.4.2 so try the old name.
-    _wrap_awaitable = getattr(asyncio, "async")
+def _wrap_awaitable(awaitable: Awaitable) -> Future:
+    # Convert Awaitables into Futures.
+    # Note that we use ensure_future, which handles both awaitables
+    # and coroutines, rather than create_task, which only accepts
+    # coroutines. (ensure_future calls create_task if given a coroutine)
+    fut = asyncio.ensure_future(awaitable)
+    # See comments on IOLoop._pending_tasks.
+    loop = IOLoop.current()
+    loop._register_task(fut)
+    fut.add_done_callback(lambda f: loop._unregister_task(f))
+    return fut
 
 
 def convert_yielded(yielded: _Yieldable) -> Future:

tornado/ioloop.py

@@ -50,7 +50,7 @@
 from typing import Union, Any, Type, Optional, Callable, TypeVar, Tuple, Awaitable
 
 if typing.TYPE_CHECKING:
-    from typing import Dict, List  # noqa: F401
+    from typing import Dict, List, Set  # noqa: F401
 
     from typing_extensions import Protocol
 else:
@@ -159,6 +159,18 @@ async def main():
     # In Python 3, _ioloop_for_asyncio maps from asyncio loops to IOLoops.
     _ioloop_for_asyncio = dict()  # type: Dict[asyncio.AbstractEventLoop, IOLoop]
 
+    # Maintain a set of all pending tasks to follow the warning in the docs
+    # of asyncio.create_tasks:
+    # https://docs.python.org/3.11/library/asyncio-task.html#asyncio.create_task
+    # This ensures that all pending tasks have a strong reference so they
+    # will not be garbage collected before they are finished.
+    # (Thus avoiding "task was destroyed but it is pending" warnings)
+    # An analogous change has been proposed in cpython for 3.13:
+    # https://github.com/python/cpython/issues/91887
+    # If that change is accepted, this can eventually be removed.
+    # If it is not, we will consider the rationale and may remove this.
+    _pending_tasks = set()  # type: Set[Future]
+
     @classmethod
     def configure(
         cls, impl: "Union[None, str, Type[Configurable]]", **kwargs: Any
@@ -805,6 +817,12 @@ def close_fd(self, fd: Union[int, _Selectable]) -> None:
         except OSError:
             pass
 
+    def _register_task(self, f: Future) -> None:
+        self._pending_tasks.add(f)
+
+    def _unregister_task(self, f: Future) -> None:
+        self._pending_tasks.discard(f)
+
 
 class _Timeout(object):
     """An IOLoop timeout, a UNIX timestamp and a callback"""