concurrent: Add future_set_exception_unless_cancelled

Tornado's limited support for cancellation means that errors after a
Future is cancelled could get raised (and probably logged) as
InvalidStateErrors.

This new function effectively changes the behavior to log the real
exception instead of the InvalidStateError. We log them instead of
ignoring them because it's generally not a good idea to let errors
pass silently even if they're after the point that no one is listening
for them (this is consistent with the way gen.multi() handles multiple
errors, for example).

Fixes #2540

diff --git a/tornado/concurrent.py b/tornado/concurrent.py
index 3c65d9a35d9c4d45567dc65e0fc39eabffddd50e..e25f3454c80b4fad1e3995288aafbd65911e439c 100644 (file)
--- a/tornado/concurrent.py
+++ b/tornado/concurrent.py
@@ -33,6 +33,8 @@ import functools
 import sys
 import types
 
+from tornado.log import app_log
+
 import typing
 from typing import Any, Callable, Optional, Tuple, Union
 
@@ -185,6 +187,28 @@ def future_set_result_unless_cancelled(
         future.set_result(value)
 
 
+def future_set_exception_unless_cancelled(
+    future: Union["futures.Future[_T]", "Future[_T]"], exc: BaseException
+) -> None:
+    """Set the given ``exc`` as the `Future`'s exception.
+
+    If the Future is already canceled, logs the exception instead. If
+    this logging is not desired, the caller should explicitly check
+    the state of the Future and call Future.set_exception instead of
+    this wrapper.
+
+    Avoids asyncio.InvalidStateError when calling set_exception() on
+    a cancelled `asyncio.Future`.
+
+    .. versionadded:: 6.0
+
+    """
+    if not future.cancelled():
+        future.set_exception(exc)
+    else:
+        app_log.error("Exception after Future was cancelled", exc_info=exc)
+
+
 def future_set_exc_info(
     future: Union["futures.Future[_T]", "Future[_T]"],
     exc_info: Tuple[
@@ -197,6 +221,12 @@ def future_set_exc_info(
     enable better tracebacks on Python 2.
 
     .. versionadded:: 5.0
+
+    .. versionchanged:: 6.0
+
+       If the future is already cancelled, this function is a no-op.
+       (previously asyncio.InvalidStateError would be raised)
+
     """
     if hasattr(future, "set_exc_info"):
         # Tornado's Future
@@ -205,7 +235,7 @@ def future_set_exc_info(
         # asyncio.Future
         if exc_info[1] is None:
             raise Exception("future_set_exc_info called with no exception")
-        future.set_exception(exc_info[1])
+        future_set_exception_unless_cancelled(future, exc_info[1])
 
 
 @typing.overload

diff --git a/tornado/httpclient.py b/tornado/httpclient.py
index d1c92a49f1fce27d30cd3e70b1ab1a37836ff699..00df064fbeafa84cf1bcc325dc88ddf4345faa12 100644 (file)
--- a/tornado/httpclient.py
+++ b/tornado/httpclient.py
@@ -45,7 +45,11 @@ import ssl
 import time
 import weakref
 
-from tornado.concurrent import Future, future_set_result_unless_cancelled
+from tornado.concurrent import (
+    Future,
+    future_set_result_unless_cancelled,
+    future_set_exception_unless_cancelled,
+)
 from tornado.escape import utf8, native_str
 from tornado import gen, httputil
 from tornado.ioloop import IOLoop
@@ -290,7 +294,7 @@ class AsyncHTTPClient(Configurable):
         def handle_response(response: "HTTPResponse") -> None:
             if response.error:
                 if raise_error or not response._error_is_response_code:
-                    future.set_exception(response.error)
+                    future_set_exception_unless_cancelled(future, response.error)
                     return
             future_set_result_unless_cancelled(future, response)
 

diff --git a/tornado/iostream.py b/tornado/iostream.py
index 2fd2814d5970f719aad6ef8bbf117f1ea2896c3b..5e98e7fe582ba012157f9bd313738f0d4cf4eb52 100644 (file)
--- a/tornado/iostream.py
+++ b/tornado/iostream.py
@@ -627,15 +627,17 @@ class BaseIOStream(object):
             futures.append(self._connect_future)
             self._connect_future = None
         for future in futures:
-            future.set_exception(StreamClosedError(real_error=self.error))
+            if not future.done():
+                future.set_exception(StreamClosedError(real_error=self.error))
             future.exception()
         if self._ssl_connect_future is not None:
             # _ssl_connect_future expects to see the real exception (typically
             # an ssl.SSLError), not just StreamClosedError.
-            if self.error is not None:
-                self._ssl_connect_future.set_exception(self.error)
-            else:
-                self._ssl_connect_future.set_exception(StreamClosedError())
+            if not self._ssl_connect_future.done():
+                if self.error is not None:
+                    self._ssl_connect_future.set_exception(self.error)
+                else:
+                    self._ssl_connect_future.set_exception(StreamClosedError())
             self._ssl_connect_future.exception()
             self._ssl_connect_future = None
         if self._close_callback is not None:

diff --git a/tornado/process.py b/tornado/process.py
index 51bb840c5b70f361faa3afb472351c3442461ecd..c1a5fce7ec486624b5f1e86e73a8fcceb9bd1727 100644 (file)
--- a/tornado/process.py
+++ b/tornado/process.py
@@ -27,7 +27,11 @@ import time
 
 from binascii import hexlify
 
-from tornado.concurrent import Future, future_set_result_unless_cancelled
+from tornado.concurrent import (
+    Future,
+    future_set_result_unless_cancelled,
+    future_set_exception_unless_cancelled,
+)
 from tornado import ioloop
 from tornado.iostream import PipeIOStream
 from tornado.log import gen_log
@@ -296,7 +300,9 @@ class Subprocess(object):
         def callback(ret: int) -> None:
             if ret != 0 and raise_error:
                 # Unfortunately we don't have the original args any more.
-                future.set_exception(CalledProcessError(ret, "unknown"))
+                future_set_exception_unless_cancelled(
+                    future, CalledProcessError(ret, "unknown")
+                )
             else:
                 future_set_result_unless_cancelled(future, ret)
 

diff --git a/tornado/test/httpclient_test.py b/tornado/test/httpclient_test.py
index 0a732ddc3a1249a8d87e40c677f77895da328bb6..745a9cb12f76918f64589dacd0ca3ff954dcc96c 100644 (file)
--- a/tornado/test/httpclient_test.py
+++ b/tornado/test/httpclient_test.py
@@ -23,7 +23,7 @@ from tornado.httpclient import (
 from tornado.httpserver import HTTPServer
 from tornado.ioloop import IOLoop
 from tornado.iostream import IOStream
-from tornado.log import gen_log
+from tornado.log import gen_log, app_log
 from tornado import netutil
 from tornado.testing import AsyncHTTPTestCase, bind_unused_port, gen_test, ExpectLog
 from tornado.test.util import skipOnTravis
@@ -570,6 +570,20 @@ X-XSS-Protection: 1;
         for k, v in response.time_info.items():
             self.assertTrue(0 <= v < 1.0, "time_info[%s] out of bounds: %s" % (k, v))
 
+    @gen_test
+    def test_error_after_cancel(self):
+        fut = self.http_client.fetch(self.get_url("/404"))
+        self.assertTrue(fut.cancel())
+        with ExpectLog(app_log, "Exception after Future was cancelled") as el:
+            # We can't wait on the cancelled Future any more, so just
+            # let the IOLoop run until the exception gets logged (or
+            # not, in which case we exit the loop and ExpectLog will
+            # raise).
+            for i in range(100):
+                yield gen.sleep(0.01)
+                if el.logged_stack:
+                    break
+
 
 class RequestProxyTest(unittest.TestCase):
     def test_request_set(self):