Repository: tornadoweb/tornado

Issue:
InvalidStateError when cancelling http requests
I am using Tornado 5.1.1 with Python 3.6 asyncio.

When I cancel() an asyncio task holding a Tornado http fetch request, I get the following exception:

```
ERROR:tornado.application:Exception in callback functools.partial(<function wrap.<locals>.null_wrapper at 0x7fda5decba60>, HTTPResponse(_body=None,_error_i
Traceback (most recent call last):
File "/data/workspace/cape/attribute-pipeline/python/lib/python3.6/site-packages/tornado/ioloop.py", line 758, in _run_callback
ret = callback()
File "/data/workspace/cape/attribute-pipeline/python/lib/python3.6/site-packages/tornado/stack_context.py", line 300, in null_wrapper
return fn(*args, **kwargs)
File "/data/workspace/cape/attribute-pipeline/python/lib/python3.6/site-packages/tornado/httpclient.py", line 290, in handle_response
future.set_exception(response.error)
asyncio.base_futures.InvalidStateError: invalid state
```

The problem looks like it's here in `httpclient.py`:
```
        def handle_response(response):
            if raise_error and response.error:
                if isinstance(response.error, HTTPError):
                    response.error.response = response
                future.set_exception(response.error)
            else:
                if response.error and not response._error_is_response_code:
                    warnings.warn("raise_error=False will allow '%s' to be raised in the future" %
                                  response.error, DeprecationWarning)
                future_set_result_unless_cancelled(future, response)
        self.fetch_impl(request, handle_response)
        return future
```

For results Tornado uses the wrapper `future_set_result_unless_cancelled`.

For exceptions, Tornado just invokes `future.set_exception(response.error)` which triggers the invalid state error since the future has been cancelled externally:

https://github.com/tornadoweb/tornado/blob/master/tornado/httpclient.py#L293

Perhaps the set_exception() just needs a similar check-if-cancelled wrapper?

Discussion:

Patch:
diff --git a/tornado/concurrent.py b/tornado/concurrent.py
index 3c65d9a3..df12da87 100644
--- 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[
@@ -193,19 +217,20 @@ def future_set_exc_info(
 ) -> None:
     """Set the given ``exc_info`` as the `Future`'s exception.
 
-    Understands both `asyncio.Future` and Tornado's extensions to
-    enable better tracebacks on Python 2.
+    Understands both `asyncio.Future` and the extensions in older
+    versions of Tornado to 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
-        future.set_exc_info(exc_info)  # type: ignore
-    else:
-        # asyncio.Future
-        if exc_info[1] is None:
-            raise Exception("future_set_exc_info called with no exception")
-        future.set_exception(exc_info[1])
+    if exc_info[1] is None:
+        raise Exception("future_set_exc_info called with no exception")
+    future_set_exception_unless_cancelled(future, exc_info[1])
 
 
 @typing.overload
diff --git a/tornado/httpclient.py b/tornado/httpclient.py
index d1c92a49..00df064f 100644
--- 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 2fd2814d..5e98e7fe 100644
--- 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 51bb840c..c1a5fce7 100644
--- 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)
 


Test patch:
diff --git a/tornado/test/httpclient_test.py b/tornado/test/httpclient_test.py
index 0a732ddc..745a9cb1 100644
--- 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):