testing: Cancel all pending coroutines in tearDown

It's difficult to synchronize test shutdown with the exits of all
coroutines, so explicitly cancel all native coroutines (which are
spammy when allowed to be GC'd).

Suppress logging of CancelledErrors in IOLoop.

tornado/http1connection.py

@@ -812,7 +812,11 @@ async def _server_request_loop(
                 request_delegate = delegate.start_request(self, conn)
                 try:
                     ret = await conn.read_response(request_delegate)
-                except (iostream.StreamClosedError, iostream.UnsatisfiableReadError):
+                except (
+                    iostream.StreamClosedError,
+                    iostream.UnsatisfiableReadError,
+                    asyncio.CancelledError,
+                ):
                     return
                 except _QuietException:
                     # This exception was already logged.

tornado/ioloop.py

@@ -724,7 +724,9 @@ def set_default_executor(self, executor: concurrent.futures.Executor) -> None:
     def _run_callback(self, callback: Callable[[], Any]) -> None:
         """Runs a callback with error handling.
 
-        For use in subclasses.
+        .. versionchanged:: 6.0
+
+           CancelledErrors are no longer logged.
         """
         try:
             ret = callback()
@@ -744,6 +746,8 @@ def _run_callback(self, callback: Callable[[], Any]) -> None:
                     pass
                 else:
                     self.add_future(ret, self._discard_future_result)
+        except asyncio.CancelledError:
+            pass
         except Exception:
             app_log.error("Exception in callback %r", callback, exc_info=True)
 

tornado/iostream.py

@@ -33,7 +33,7 @@
 import sys
 import re
 
-from tornado.concurrent import Future
+from tornado.concurrent import Future, future_set_result_unless_cancelled
 from tornado import ioloop
 from tornado.log import gen_log
 from tornado.netutil import ssl_wrap_socket, _client_ssl_defaults, _server_ssl_defaults
@@ -803,7 +803,7 @@ def _finish_read(self, size: int, streaming: bool) -> None:
         if self._read_future is not None:
             future = self._read_future
             self._read_future = None
-            future.set_result(result)
+            future_set_result_unless_cancelled(future, result)
         self._maybe_add_error_listener()
 
     def _try_inline_read(self) -> None:
@@ -972,7 +972,7 @@ def _handle_write(self) -> None:
             if index > self._total_write_done_index:
                 break
             self._write_futures.popleft()
-            future.set_result(None)
+            future_set_result_unless_cancelled(future, None)
 
     def _consume(self, loc: int) -> bytes:
         # Consume loc bytes from the read buffer and return them
@@ -1311,7 +1311,7 @@ def _handle_connect(self) -> None:
         if self._connect_future is not None:
             future = self._connect_future
             self._connect_future = None
-            future.set_result(self)
+            future_set_result_unless_cancelled(future, self)
         self._connecting = False
 
     def set_nodelay(self, value: bool) -> None:
@@ -1429,7 +1429,7 @@ def _finish_ssl_connect(self) -> None:
         if self._ssl_connect_future is not None:
             future = self._ssl_connect_future
             self._ssl_connect_future = None
-            future.set_result(self)
+            future_set_result_unless_cancelled(future, self)
 
     def _verify_cert(self, peercert: Any) -> bool:
         """Returns True if peercert is valid according to the configured

tornado/test/httpclient_test.py

@@ -664,6 +664,7 @@ def stop_server():
 
             @gen.coroutine
             def slow_stop():
+                yield self.server.close_all_connections()
                 # The number of iterations is difficult to predict. Typically,
                 # one is sufficient, although sometimes it needs more.
                 for i in range(5):

tornado/test/testing_test.py

@@ -1,9 +1,11 @@
 from tornado import gen, ioloop
 from tornado.httpserver import HTTPServer
+from tornado.locks import Event
 from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, bind_unused_port, gen_test
 from tornado.web import Application
 import asyncio
 import contextlib
+import gc
 import os
 import platform
 import traceback
@@ -55,6 +57,30 @@ def test_subsequent_wait_calls(self):
         self.wait(timeout=0.15)
 
 
+class LeakTest(AsyncTestCase):
+    def tearDown(self):
+        super().tearDown()
+        # Trigger a gc to make warnings more deterministic.
+        gc.collect()
+
+    def test_leaked_coroutine(self):
+        # This test verifies that "leaked" coroutines are shut down
+        # without triggering warnings like "task was destroyed but it
+        # is pending". If this test were to fail, it would fail
+        # because runtests.py detected unexpected output to stderr.
+        event = Event()
+
+        async def callback():
+            try:
+                await event.wait()
+            except asyncio.CancelledError:
+                pass
+
+        self.io_loop.add_callback(callback)
+        self.io_loop.add_callback(self.stop)
+        self.wait()
+
+
 class AsyncHTTPTestCaseTest(AsyncHTTPTestCase):
     def setUp(self):
         super(AsyncHTTPTestCaseTest, self).setUp()

tornado/testing.py

@@ -9,6 +9,7 @@
   for the tornado.autoreload module to rerun the tests when code changes.
 """
 
+import asyncio
 from collections.abc import Generator
 import functools
 import inspect
@@ -178,6 +179,36 @@ def setUp(self) -> None:
         self.io_loop.make_current()
 
     def tearDown(self) -> None:
+        # Native coroutines tend to produce warnings if they're not
+        # allowed to run to completion. It's difficult to ensure that
+        # this always happens in tests, so cancel any tasks that are
+        # still pending by the time we get here.
+        asyncio_loop = self.io_loop.asyncio_loop  # type: ignore
+        if hasattr(asyncio, "all_tasks"):  # py37
+            tasks = asyncio.all_tasks(asyncio_loop)  # type: ignore
+        else:
+            tasks = asyncio.Task.all_tasks(asyncio_loop)
+        # Tasks that are done may still appear here and may contain
+        # non-cancellation exceptions, so filter them out.
+        tasks = [t for t in tasks if not t.done()]
+        for t in tasks:
+            t.cancel()
+        # Allow the tasks to run and finalize themselves (which means
+        # raising a CancelledError inside the coroutine). This may
+        # just transform the "task was destroyed but it is pending"
+        # warning into a "uncaught CancelledError" warning, but
+        # catching CancelledErrors in coroutines that may leak is
+        # simpler than ensuring that no coroutines leak.
+        if tasks:
+            done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))
+            assert not pending
+            # If any task failed with anything but a CancelledError, raise it.
+            for f in done:
+                try:
+                    f.result()
+                except asyncio.CancelledError:
+                    pass
+
         # Clean up Subprocess, so it can be used again with a new ioloop.
         Subprocess.uninitialize()
         self.io_loop.clear_current()