Repository: tornadoweb/tornado
Pull request #2253 (2018-01-22 02:37:31)

Misc small fixes for 5.0


Patch:
diff --git a/.travis.yml b/.travis.yml
index dadc4790..64caeb3d 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -6,6 +6,9 @@ sudo: false
 
 matrix:
     fast_finish: true
+    # Nightly python is failing as of 2018-01-26
+    allow_failures:
+    - python: nightly
 
 language: python
 # For a list of available versions, run
diff --git a/docs/releases/v5.0.0.rst b/docs/releases/v5.0.0.rst
index 33d42b67..8d9b56ba 100644
--- a/docs/releases/v5.0.0.rst
+++ b/docs/releases/v5.0.0.rst
@@ -309,7 +309,6 @@ Other notes
   improve performance.
 - Requests with invalid websocket headers now get a response with
   status code 400 instead of a closed connection.
-- `.WebSocketHandler.write_message` now raises `.StreamClosedError` if
-  called after the connection is closed.
-  TODO(bdarnell): fix this up with WebSocketClosedError
+- `.WebSocketHandler.write_message` now raises `.WebSocketClosedError` if
+  the connection closes while the write is in progress.
 - The ``io_loop`` argument to `.websocket_connect` has been removed.
diff --git a/tornado/concurrent.py b/tornado/concurrent.py
index cca6468c..65da0ac8 100644
--- a/tornado/concurrent.py
+++ b/tornado/concurrent.py
@@ -337,11 +337,12 @@ class Future(object):
 
     def _set_done(self):
         self._done = True
-        from tornado.ioloop import IOLoop
-        loop = IOLoop.current()
-        for cb in self._callbacks:
-            loop.add_callback(cb, self)
-        self._callbacks = None
+        if self._callbacks:
+            from tornado.ioloop import IOLoop
+            loop = IOLoop.current()
+            for cb in self._callbacks:
+                loop.add_callback(cb, self)
+            self._callbacks = None
 
     # On Python 3.3 or older, objects with a destructor part of a reference
     # cycle are never destroyed. It's no longer the case on Python 3.4 thanks to
diff --git a/tornado/locks.py b/tornado/locks.py
index 44222a43..94adb322 100644
--- a/tornado/locks.py
+++ b/tornado/locks.py
@@ -15,6 +15,7 @@
 from __future__ import absolute_import, division, print_function
 
 import collections
+from concurrent.futures import CancelledError
 
 from tornado import gen, ioloop
 from tornado.concurrent import Future, future_set_result_unless_cancelled
@@ -196,7 +197,8 @@ class Event(object):
         Done
     """
     def __init__(self):
-        self._future = Future()
+        self._value = False
+        self._waiters = set()
 
     def __repr__(self):
         return '<%s %s>' % (
@@ -204,23 +206,26 @@ class Event(object):
 
     def is_set(self):
         """Return ``True`` if the internal flag is true."""
-        return self._future.done()
+        return self._value
 
     def set(self):
         """Set the internal flag to ``True``. All waiters are awakened.
 
         Calling `.wait` once the flag is set will not block.
         """
-        if not self._future.done():
-            self._future.set_result(None)
+        if not self._value:
+            self._value = True
+
+            for fut in self._waiters:
+                if not fut.done():
+                    fut.set_result(None)
 
     def clear(self):
         """Reset the internal flag to ``False``.
 
         Calls to `.wait` will block until `.set` is called.
         """
-        if self._future.done():
-            self._future = Future()
+        self._value = False
 
     def wait(self, timeout=None):
         """Block until the internal flag is true.
@@ -228,10 +233,21 @@ class Event(object):
         Returns a Future, which raises `tornado.util.TimeoutError` after a
         timeout.
         """
+        fut = Future()
+        if self._value:
+            fut.set_result(None)
+            return fut
+        self._waiters.add(fut)
+        fut.add_done_callback(lambda fut: self._waiters.remove(fut))
         if timeout is None:
-            return self._future
+            return fut
         else:
-            return gen.with_timeout(timeout, self._future)
+            timeout_fut = gen.with_timeout(timeout, fut, quiet_exceptions=(CancelledError,))
+            # This is a slightly clumsy workaround for the fact that
+            # gen.with_timeout doesn't cancel its futures. Cancelling
+            # fut will remove it from the waiters list.
+            timeout_fut.add_done_callback(lambda tf: fut.cancel() if not fut.done() else None)
+            return timeout_fut
 
 
 class _ReleasingContextManager(object):
diff --git a/tornado/websocket.py b/tornado/websocket.py
index 32012780..53832d2b 100644
--- a/tornado/websocket.py
+++ b/tornado/websocket.py
@@ -237,6 +237,7 @@ class WebSocketHandler(tornado.web.RequestHandler):
         is allowed.
 
         If the connection is already closed, raises `WebSocketClosedError`.
+        Returns a `.Future` which can be used for flow control.
 
         .. versionchanged:: 3.2
            `WebSocketClosedError` was added (previously a closed connection
@@ -244,6 +245,10 @@ class WebSocketHandler(tornado.web.RequestHandler):
 
         .. versionchanged:: 4.3
            Returns a `.Future` which can be used for flow control.
+
+        .. versionchanged:: 5.0
+           Consistently raises `WebSocketClosedError`. Previously could
+           sometimes raise `.StreamClosedError`.
         """
         if self.ws_connection is None:
             raise WebSocketClosedError()
@@ -788,7 +793,23 @@ class WebSocketProtocol13(WebSocketProtocol):
         if self._compressor:
             message = self._compressor.compress(message)
             flags |= self.RSV1
-        return self._write_frame(True, opcode, message, flags=flags)
+        # For historical reasons, write methods in Tornado operate in a semi-synchronous
+        # mode in which awaiting the Future they return is optional (But errors can
+        # still be raised). This requires us to go through an awkward dance here
+        # to transform the errors that may be returned while presenting the same
+        # semi-synchronous interface.
+        try:
+            fut = self._write_frame(True, opcode, message, flags=flags)
+        except StreamClosedError:
+            raise WebSocketClosedError()
+
+        @gen.coroutine
+        def wrapper():
+            try:
+                yield fut
+            except StreamClosedError:
+                raise WebSocketClosedError()
+        return wrapper()
 
     def write_ping(self, data):
         """Send ping frame."""
@@ -1144,8 +1165,16 @@ class WebSocketClientConnection(simple_httpclient._HTTPConnection):
         future_set_result_unless_cancelled(self.connect_future, self)
 
     def write_message(self, message, binary=False):
-        """Sends a message to the WebSocket server."""
-        return self.protocol.write_message(message, binary)
+        """Sends a message to the WebSocket server.
+
+        If the stream is closed, raises `WebSocketClosedError`.
+        Returns a `.Future` which can be used for flow control.
+
+        .. versionchanged:: 5.0
+           Exception raised on a closed stream changed from `.StreamClosedError`
+           to `WebSocketClosedError`.
+        """
+        return self.protocol.write_message(message, binary=binary)
 
     def read_message(self, callback=None):
         """Reads a message from the WebSocket server.


Test patch:
diff --git a/maint/test/cython/tox.ini b/maint/test/cython/tox.ini
index 725e1704..bbf8f157 100644
--- a/maint/test/cython/tox.ini
+++ b/maint/test/cython/tox.ini
@@ -1,6 +1,6 @@
 [tox]
 # This currently segfaults on pypy.
-envlist = py27,py34,py35,py36
+envlist = py27,py35,py36
 
 [testenv]
 deps =
@@ -13,6 +13,5 @@ commands = python -m unittest cythonapp_test
 # defaults for the others.
 basepython =
            py27: python2.7
-           py34: python3.4
            py35: python3.5
            py36: python3.6
diff --git a/tornado/test/asyncio_test.py b/tornado/test/asyncio_test.py
index b08f065f..bbabb521 100644
--- a/tornado/test/asyncio_test.py
+++ b/tornado/test/asyncio_test.py
@@ -47,7 +47,8 @@ class AsyncIOLoopTest(AsyncTestCase):
         if hasattr(asyncio, 'ensure_future'):
             ensure_future = asyncio.ensure_future
         else:
-            ensure_future = asyncio.async
+            # async is a reserved word in Python 3.7
+            ensure_future = getattr(asyncio, 'async')
 
         x = yield ensure_future(
             asyncio.get_event_loop().run_in_executor(None, lambda: 42))
diff --git a/tornado/test/gen_test.py b/tornado/test/gen_test.py
index ff7d9770..6850af81 100644
--- a/tornado/test/gen_test.py
+++ b/tornado/test/gen_test.py
@@ -1069,6 +1069,20 @@ class GenCoroutineTest(AsyncTestCase):
         actual = repr(coro)
         self.assertIn(expected, actual)
 
+    @unittest.skipIf(asyncio is None, "asyncio module not present")
+    @gen_test
+    def test_asyncio_gather(self):
+        # This demonstrates that tornado coroutines can be understood
+        # by asyncio (This failed prior to Tornado 5.0).
+        @gen.coroutine
+        def f():
+            yield gen.moment
+            raise gen.Return(1)
+
+        ret = yield asyncio.gather(f(), f())
+        self.assertEqual(ret, [1, 1])
+        self.finished = True
+
 
 class GenSequenceHandler(RequestHandler):
     @asynchronous
diff --git a/tornado/test/template_test.py b/tornado/test/template_test.py
index 41c1c90c..c19728e1 100644
--- a/tornado/test/template_test.py
+++ b/tornado/test/template_test.py
@@ -6,8 +6,8 @@ import traceback
 
 from tornado.escape import utf8, native_str, to_unicode
 from tornado.template import Template, DictLoader, ParseError, Loader
-from tornado.test.util import unittest, is_coverage_running
-from tornado.util import ObjectDict, unicode_type, PY3
+from tornado.test.util import unittest
+from tornado.util import ObjectDict, unicode_type
 
 
 class TemplateTest(unittest.TestCase):
@@ -175,11 +175,6 @@ try{% set y = 1/x %}
         self.assertEqual(template.generate(), '0')
 
     def test_non_ascii_name(self):
-        if PY3 and is_coverage_running():
-            try:
-                os.fsencode(u"t\u00e9st.html")
-            except UnicodeEncodeError:
-                self.skipTest("coverage tries to access unencodable filename")
         loader = DictLoader({u"t\u00e9st.html": "hello"})
         self.assertEqual(loader.load(u"t\u00e9st.html").generate(), b"hello")
 
diff --git a/tornado/test/util.py b/tornado/test/util.py
index 1d998f0e..63a9762f 100644
--- a/tornado/test/util.py
+++ b/tornado/test/util.py
@@ -97,24 +97,6 @@ def exec_test(caller_globals, caller_locals, s):
     return local_namespace
 
 
-def is_coverage_running():
-    """Return whether coverage is currently running.
-    """
-    if 'coverage' not in sys.modules:
-        return False
-    tracer = sys.gettrace()
-    if tracer is None:
-        return False
-    try:
-        mod = tracer.__module__
-    except AttributeError:
-        try:
-            mod = tracer.__class__.__module__
-        except AttributeError:
-            return False
-    return mod.startswith('coverage')
-
-
 def subTest(test, *args, **kwargs):
     """Compatibility shim for unittest.TestCase.subTest.
 
diff --git a/tornado/test/websocket_test.py b/tornado/test/websocket_test.py
index 54734d81..f7e8016a 100644
--- a/tornado/test/websocket_test.py
+++ b/tornado/test/websocket_test.py
@@ -7,7 +7,6 @@ import traceback
 from tornado.concurrent import Future
 from tornado import gen
 from tornado.httpclient import HTTPError, HTTPRequest
-from tornado.iostream import StreamClosedError
 from tornado.log import gen_log, app_log
 from tornado.template import DictLoader
 from tornado.testing import AsyncHTTPTestCase, gen_test, bind_unused_port, ExpectLog
@@ -25,7 +24,7 @@ except ImportError:
     traceback.print_exc()
     raise
 
-from tornado.websocket import WebSocketHandler, websocket_connect, WebSocketError
+from tornado.websocket import WebSocketHandler, websocket_connect, WebSocketError, WebSocketClosedError
 
 try:
     from tornado import speedups
@@ -50,10 +49,11 @@ class TestWebSocketHandler(WebSocketHandler):
 
 
 class EchoHandler(TestWebSocketHandler):
+    @gen.coroutine
     def on_message(self, message):
         try:
-            self.write_message(message, isinstance(message, bytes))
-        except StreamClosedError:
+            yield self.write_message(message, isinstance(message, bytes))
+        except WebSocketClosedError:
             pass
 
 
@@ -342,7 +342,7 @@ class WebSocketTest(WebSocketBaseTestCase):
         ws = yield self.ws_connect('/close_reason')
         msg = yield ws.read_message()
         self.assertIs(msg, None)
-        with self.assertRaises(StreamClosedError):
+        with self.assertRaises(WebSocketClosedError):
             ws.write_message('hello')
 
     @gen_test