Repository: tornadoweb/tornado
Pull request #1476 (2015-08-03 02:07:15)

Support Python 3.5 `async` statements
`Lock` and `Semaphore` support `async with`. All of the `queues` and `WaitIterator` support `async for`. It is now possible to yield a list of awaitables, not just `Futures`.

@ajdavis 


Patch:
diff --git a/.travis.yml b/.travis.yml
index f894e855..0af81fe0 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -7,6 +7,7 @@ python:
     - 3.2
     - 3.3
     - 3.4
+    - nightly
     - pypy3
 
 env:
@@ -47,7 +48,9 @@ script:
     # We use "python -m coverage" instead of the "bin/coverage" script
     # so we can pass additional arguments to python.  However, this doesn't
     # work on 2.6, so skip coverage on that version.
-    - if [[ $TRAVIS_PYTHON_VERSION != 2.6 ]]; then export TARGET="-m coverage run $TARGET"; fi
+    # coverage needs a function that was removed in python 3.6 so we can't
+    # run it with nightly cpython.
+    - if [[ $TRAVIS_PYTHON_VERSION != 2.6 && $TRAVIS_PYTHON_VERSION != nightly ]]; then export TARGET="-m coverage run $TARGET"; fi
     - python $TARGET
     - python $TARGET --ioloop=tornado.platform.select.SelectIOLoop
     - python -O $TARGET
diff --git a/tornado/gen.py b/tornado/gen.py
index e0ce1dde..78ddce75 100644
--- a/tornado/gen.py
+++ b/tornado/gen.py
@@ -108,6 +108,11 @@ try:
 except ImportError:
     def isawaitable(x): return False
 
+try:
+    import builtins  # py3
+except ImportError:
+    import __builtin__ as builtins
+
 
 class KeyReuseError(Exception):
     pass
@@ -333,7 +338,22 @@ class WaitIterator(object):
     arguments were used in the construction of the `WaitIterator`,
     ``current_index`` will use the corresponding keyword).
 
+    On Python 3.5, `WaitIterator` implements the async iterator
+    protocol, so it can be used with the ``async for`` statement (note
+    that in this version the entire iteration is aborted if any value
+    raises an exception, while the previous example can continue past
+    individual errors)::
+
+      async for result in gen.WaitIterator(future1, future2):
+          print("Result {} received from {} at {}".format(
+              result, wait_iterator.current_future,
+              wait_iterator.current_index))
+
     .. versionadded:: 4.1
+
+    .. versionchanged:: 4.3
+       Added ``async for`` support in Python 3.5.
+
     """
     def __init__(self, *args, **kwargs):
         if args and kwargs:
@@ -390,6 +410,16 @@ class WaitIterator(object):
         self.current_future = done
         self.current_index = self._unfinished.pop(done)
 
+    @coroutine
+    def __aiter__(self):
+        raise Return(self)
+
+    def __anext__(self):
+        if self.done():
+            # Lookup by name to silence pyflakes on older versions.
+            raise getattr(builtins, 'StopAsyncIteration')()
+        return self.next()
+
 
 class YieldPoint(object):
     """Base class for objects that may be yielded from the generator.
@@ -624,11 +654,12 @@ class Multi(YieldPoint):
 def multi_future(children, quiet_exceptions=()):
     """Wait for multiple asynchronous futures in parallel.
 
-    Takes a list of ``Futures`` (but *not* other ``YieldPoints``) and returns
-    a new Future that resolves when all the other Futures are done.
-    If all the ``Futures`` succeeded, the returned Future's result is a list
-    of their results.  If any failed, the returned Future raises the exception
-    of the first one to fail.
+    Takes a list of ``Futures`` or other yieldable objects (with the
+    exception of the legacy `.YieldPoint` interfaces) and returns a
+    new Future that resolves when all the other Futures are done. If
+    all the ``Futures`` succeeded, the returned Future's result is a
+    list of their results. If any failed, the returned Future raises
+    the exception of the first one to fail.
 
     Instead of a list, the argument may also be a dictionary whose values are
     Futures, in which case a parallel dictionary is returned mapping the same
@@ -649,12 +680,16 @@ def multi_future(children, quiet_exceptions=()):
        If multiple ``Futures`` fail, any exceptions after the first (which is
        raised) will be logged. Added the ``quiet_exceptions``
        argument to suppress this logging for selected exception types.
+
+    .. versionchanged:: 4.3
+       Added support for other yieldable objects.
     """
     if isinstance(children, dict):
         keys = list(children.keys())
         children = children.values()
     else:
         keys = None
+    children = list(map(convert_yielded, children))
     assert all(is_future(i) for i in children)
     unfinished_children = set(children)
 
diff --git a/tornado/locks.py b/tornado/locks.py
index 27e14953..a1817728 100644
--- a/tornado/locks.py
+++ b/tornado/locks.py
@@ -327,6 +327,20 @@ class Semaphore(_TimeoutGarbageCollector):
 
             # Now the semaphore has been released.
             print("Worker %d is done" % worker_id)
+
+    In Python 3.5, the semaphore itself can be used as an async context
+    manager::
+
+        async def worker(worker_id):
+            async with sem:
+                print("Worker %d is working" % worker_id)
+                await use_some_resource()
+
+            # Now the semaphore has been released.
+            print("Worker %d is done" % worker_id)
+
+    .. versionchanged:: 4.3
+       Added ``async with`` support in Python 3.5.
     """
     def __init__(self, value=1):
         super(Semaphore, self).__init__()
@@ -389,6 +403,14 @@ class Semaphore(_TimeoutGarbageCollector):
 
     __exit__ = __enter__
 
+    @gen.coroutine
+    def __aenter__(self):
+        yield self.acquire()
+
+    @gen.coroutine
+    def __aexit__(self, typ, value, tb):
+        self.release()
+
 
 class BoundedSemaphore(Semaphore):
     """A semaphore that prevents release() being called too many times.
@@ -418,7 +440,7 @@ class Lock(object):
 
     Releasing an unlocked lock raises `RuntimeError`.
 
-    `acquire` supports the context manager protocol:
+    `acquire` supports the context manager protocol in all Python versions:
 
     >>> from tornado import gen, locks
     >>> lock = locks.Lock()
@@ -430,6 +452,22 @@ class Lock(object):
     ...        pass
     ...
     ...    # Now the lock is released.
+
+    In Python 3.5, `Lock` also supports the async context manager
+    protocol. Note that in this case there is no `acquire`, because
+    ``async with`` includes both the ``yield`` and the ``acquire``
+    (just as it does with `threading.Lock`):
+
+    >>> async def f():  # doctest: +SKIP
+    ...    async with lock:
+    ...        # Do something holding the lock.
+    ...        pass
+    ...
+    ...    # Now the lock is released.
+
+    .. versionchanged:: 3.5
+       Added ``async with`` support in Python 3.5.
+
     """
     def __init__(self):
         self._block = BoundedSemaphore(value=1)
@@ -464,3 +502,11 @@ class Lock(object):
             "Use Lock like 'with (yield lock)', not like 'with lock'")
 
     __exit__ = __enter__
+
+    @gen.coroutine
+    def __aenter__(self):
+        yield self.acquire()
+
+    @gen.coroutine
+    def __aexit__(self, typ, value, tb):
+        self.release()
diff --git a/tornado/queues.py b/tornado/queues.py
index 6d694cc4..129b204e 100644
--- a/tornado/queues.py
+++ b/tornado/queues.py
@@ -44,6 +44,14 @@ def _set_timeout(future, timeout):
             lambda _: io_loop.remove_timeout(timeout_handle))
 
 
+class _QueueIterator(object):
+    def __init__(self, q):
+        self.q = q
+
+    def __anext__(self):
+        return self.q.get()
+
+
 class Queue(object):
     """Coordinate producer and consumer coroutines.
 
@@ -96,6 +104,21 @@ class Queue(object):
         Doing work on 3
         Doing work on 4
         Done
+
+    In Python 3.5, `Queue` implements the async iterator protocol, so
+    ``consumer()`` could be rewritten as::
+
+        async def consumer():
+            async for item in q:
+                try:
+                    print('Doing work on %s' % item)
+                    yield gen.sleep(0.01)
+                finally:
+                    q.task_done()
+
+    .. versionchanged:: 4.3
+       Added ``async for`` support in Python 3.5.
+
     """
     def __init__(self, maxsize=0):
         if maxsize is None:
@@ -220,6 +243,10 @@ class Queue(object):
         """
         return self._finished.wait(timeout)
 
+    @gen.coroutine
+    def __aiter__(self):
+        return _QueueIterator(self)
+
     # These three are overridable in subclasses.
     def _init(self):
         self._queue = collections.deque()
diff --git a/tox.ini b/tox.ini
index 2e227c47..78687ec7 100644
--- a/tox.ini
+++ b/tox.ini
@@ -15,8 +15,8 @@ envlist =
         # Basic configurations: Run the tests in both minimal installations
         # and with all optional dependencies.
         # (pypy3 doesn't have any optional deps yet)
-        {py26,py27,pypy,py32,py33,py34,pypy3},
-        {py26,py27,pypy,py32,py33,py34}-full,
+        {py26,py27,pypy,py32,py33,py34,py35,pypy3},
+        {py26,py27,pypy,py32,py33,py34,py35}-full,
 
         # Also run the tests with each possible replacement of a default
         # component.  Run each test on both python 2 and 3 where possible.
@@ -58,6 +58,7 @@ basepython =
            py32: python3.2
            py33: python3.3
            py34: python3.4
+           py35: python3.5
            pypy: pypy
            pypy3: pypy3
            py2: python2.7
@@ -72,10 +73,10 @@ deps =
      py3-unittest2: unittest2py3k
      # cpython-only deps: pycurl installs but curl_httpclient doesn't work;
      # twisted mostly works but is a bit flaky under pypy.
-     {py26,py27,py32,py33,py34}-full: pycurl
+     {py26,py27,py32,py33,py34,py35}-full: pycurl
      {py2,py3}: pycurl==7.19.3.1
      # twisted is cpython only and doesn't support py32.
-     {py26,py27,py33,py34}-full: twisted
+     {py26,py27,py33,py34,py35}-full: twisted
      {py2,py3}: twisted
      # pycares installation currently fails on py33
      # (https://github.com/pypa/pip/pull/816)
@@ -95,7 +96,7 @@ deps =
 
 setenv =
        # The extension is mandatory on cpython.
-       {py2,py26,py27,py3,py32,py33,py34}: TORNADO_EXTENSION=1
+       {py2,py26,py27,py3,py32,py33,py34,py35}: TORNADO_EXTENSION=1
        # In python 3, opening files in text mode uses a
        # system-dependent encoding by default.  Run the tests with "C"
        # (ascii) and "utf-8" locales to ensure we don't have hidden
@@ -104,7 +105,7 @@ setenv =
        lang_utf8: LANG=en_US.utf-8
        # tox's parser chokes if all the setenv entries are conditional.
        DUMMY=dummy
-       {py2,py26,py27,py3,py32,py33,py34}-no-ext: TORNADO_EXTENSION=0
+       {py2,py26,py27,py3,py32,py33,py34,py35}-no-ext: TORNADO_EXTENSION=0
 
 # All non-comment lines but the last must end in a backslash.
 # Tox filters line-by-line based on the environment name.
@@ -112,7 +113,7 @@ commands =
          python \
          # py3*: -b turns on an extra warning when calling
          # str(bytes), and -bb makes it an error.
-         {py3,py32,py33,py34,pypy3}: -bb \
+         {py3,py32,py33,py34,py35,pypy3}: -bb \
          # Python's optimized mode disables the assert statement, so
          # run the tests in this mode to ensure we haven't fallen into
          # the trap of relying on an assertion's side effects or using


Test patch:
diff --git a/tornado/test/gen_test.py b/tornado/test/gen_test.py
index 7b47f130..d37e0936 100644
--- a/tornado/test/gen_test.py
+++ b/tornado/test/gen_test.py
@@ -6,7 +6,6 @@ import functools
 import sys
 import textwrap
 import time
-import platform
 import weakref
 
 from tornado.concurrent import return_future, Future
@@ -16,7 +15,7 @@ from tornado.ioloop import IOLoop
 from tornado.log import app_log
 from tornado import stack_context
 from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, ExpectLog, gen_test
-from tornado.test.util import unittest, skipOnTravis
+from tornado.test.util import unittest, skipOnTravis, skipBefore33, skipBefore35, skipNotCPython, exec_test
 from tornado.web import Application, RequestHandler, asynchronous, HTTPError
 
 from tornado import gen
@@ -26,11 +25,6 @@ try:
 except ImportError:
     futures = None
 
-skipBefore33 = unittest.skipIf(sys.version_info < (3, 3), 'PEP 380 (yield from) not available')
-skipBefore35 = unittest.skipIf(sys.version_info < (3, 5), 'PEP 492 (async/await) not available')
-skipNotCPython = unittest.skipIf(platform.python_implementation() != 'CPython',
-                                 'Not CPython implementation')
-
 
 class GenEngineTest(AsyncTestCase):
     def setUp(self):
@@ -694,19 +688,13 @@ class GenCoroutineTest(AsyncTestCase):
     @skipBefore33
     @gen_test
     def test_async_return(self):
-        # It is a compile-time error to return a value in a generator
-        # before Python 3.3, so we must test this with exec.
-        # Flatten the real global and local namespace into our fake globals:
-        # it's all global from the perspective of f().
-        global_namespace = dict(globals(), **locals())
-        local_namespace = {}
-        exec(textwrap.dedent("""
+        namespace = exec_test(globals(), locals(), """
         @gen.coroutine
         def f():
             yield gen.Task(self.io_loop.add_callback)
             return 42
-        """), global_namespace, local_namespace)
-        result = yield local_namespace['f']()
+        """)
+        result = yield namespace['f']()
         self.assertEqual(result, 42)
         self.finished = True
 
@@ -716,16 +704,14 @@ class GenCoroutineTest(AsyncTestCase):
         # A yield statement exists but is not executed, which means
         # this function "returns" via an exception.  This exception
         # doesn't happen before the exception handling is set up.
-        global_namespace = dict(globals(), **locals())
-        local_namespace = {}
-        exec(textwrap.dedent("""
+        namespace = exec_test(globals(), locals(), """
         @gen.coroutine
         def f():
             if True:
                 return 42
             yield gen.Task(self.io_loop.add_callback)
-        """), global_namespace, local_namespace)
-        result = yield local_namespace['f']()
+        """)
+        result = yield namespace['f']()
         self.assertEqual(result, 42)
         self.finished = True
 
@@ -735,17 +721,33 @@ class GenCoroutineTest(AsyncTestCase):
         # This test verifies that an async function can await a
         # yield-based gen.coroutine, and that a gen.coroutine
         # (the test method itself) can yield an async function.
-        global_namespace = dict(globals(), **locals())
-        local_namespace = {}
-        exec(textwrap.dedent("""
+        namespace = exec_test(globals(), locals(), """
         async def f():
             await gen.Task(self.io_loop.add_callback)
             return 42
-        """), global_namespace, local_namespace)
-        result = yield local_namespace['f']()
+        """)
+        result = yield namespace['f']()
         self.assertEqual(result, 42)
         self.finished = True
 
+    @skipBefore35
+    @gen_test
+    def test_async_await_mixed_multi(self):
+        namespace = exec_test(globals(), locals(), """
+        async def f1():
+            await gen.Task(self.io_loop.add_callback)
+            return 42
+        """)
+
+        @gen.coroutine
+        def f2():
+            yield gen.Task(self.io_loop.add_callback)
+            raise gen.Return(43)
+
+        results = yield [namespace['f1'](), f2()]
+        self.assertEqual(results, [42, 43])
+        self.finished = True
+
     @gen_test
     def test_sync_return_no_value(self):
         @gen.coroutine
@@ -1283,6 +1285,46 @@ class WaitIteratorTest(AsyncTestCase):
                     self.assertEqual(g.current_index, 3, 'wrong index')
             i += 1
 
+    @skipBefore35
+    @gen_test
+    def test_iterator_async_await(self):
+        # Recreate the previous test with py35 syntax. It's a little clunky
+        # because of the way the previous test handles an exception on
+        # a single iteration.
+        futures = [Future(), Future(), Future(), Future()]
+        self.finish_coroutines(0, futures)
+        self.finished = False
+
+        namespace = exec_test(globals(), locals(), """
+        async def f():
+            i = 0
+            g = gen.WaitIterator(*futures)
+            try:
+                async for r in g:
+                    if i == 0:
+                        self.assertEqual(r, 24, 'iterator value incorrect')
+                        self.assertEqual(g.current_index, 2, 'wrong index')
+                    else:
+                        raise Exception("expected exception on iteration 1")
+                    i += 1
+            except ZeroDivisionError:
+                i += 1
+            async for r in g:
+                if i == 2:
+                    self.assertEqual(r, 42, 'iterator value incorrect')
+                    self.assertEqual(g.current_index, 1, 'wrong index')
+                elif i == 3:
+                    self.assertEqual(r, 84, 'iterator value incorrect')
+                    self.assertEqual(g.current_index, 3, 'wrong index')
+                else:
+                    raise Exception("didn't expect iteration %d" % i)
+                i += 1
+            self.finished = True
+        """)
+        yield namespace['f']()
+        self.assertTrue(self.finished)
+
+
     @gen_test
     def test_no_ref(self):
         # In this usage, there is no direct hard reference to the
diff --git a/tornado/test/locks_test.py b/tornado/test/locks_test.py
index 90bdafaa..43a6d516 100644
--- a/tornado/test/locks_test.py
+++ b/tornado/test/locks_test.py
@@ -15,7 +15,7 @@ from datetime import timedelta
 from tornado import gen, locks
 from tornado.gen import TimeoutError
 from tornado.testing import gen_test, AsyncTestCase
-from tornado.test.util import unittest
+from tornado.test.util import unittest, skipBefore35, exec_test
 
 
 class ConditionTest(AsyncTestCase):
@@ -328,6 +328,22 @@ class SemaphoreContextManagerTest(AsyncTestCase):
         # Semaphore was released and can be acquired again.
         self.assertTrue(sem.acquire().done())
 
+    @skipBefore35
+    @gen_test
+    def test_context_manager_async_await(self):
+        # Repeat the above test using 'async with'.
+        sem = locks.Semaphore()
+
+        namespace = exec_test(globals(), locals(), """
+        async def f():
+            async with sem as yielded:
+                self.assertTrue(yielded is None)
+        """)
+        yield namespace['f']()
+
+        # Semaphore was released and can be acquired again.
+        self.assertTrue(sem.acquire().done())
+
     @gen_test
     def test_context_manager_exception(self):
         sem = locks.Semaphore()
@@ -443,6 +459,26 @@ class LockTests(AsyncTestCase):
         yield futures
         self.assertEqual(list(range(N)), history)
 
+    @skipBefore35
+    @gen_test
+    def test_acquire_fifo_async_with(self):
+        # Repeat the above test using `async with lock:`
+        # instead of `with (yield lock.acquire()):`.
+        lock = locks.Lock()
+        self.assertTrue(lock.acquire().done())
+        N = 5
+        history = []
+
+        namespace = exec_test(globals(), locals(), """
+        async def f(idx):
+            async with lock:
+                history.append(idx)
+        """)
+        futures = [namespace['f'](i) for i in range(N)]
+        lock.release()
+        yield futures
+        self.assertEqual(list(range(N)), history)
+
     @gen_test
     def test_acquire_timeout(self):
         lock = locks.Lock()
diff --git a/tornado/test/queues_test.py b/tornado/test/queues_test.py
index f2ffb646..519dd6ae 100644
--- a/tornado/test/queues_test.py
+++ b/tornado/test/queues_test.py
@@ -12,11 +12,13 @@
 
 from datetime import timedelta
 from random import random
+import sys
+import textwrap
 
 from tornado import gen, queues
 from tornado.gen import TimeoutError
 from tornado.testing import gen_test, AsyncTestCase
-from tornado.test.util import unittest
+from tornado.test.util import unittest, skipBefore35, exec_test
 
 
 class QueueBasicTest(AsyncTestCase):
@@ -112,7 +114,7 @@ class QueueGetTest(AsyncTestCase):
         get = q.get()
         with self.assertRaises(TimeoutError):
             yield get_timeout
-        
+
         q.put_nowait(0)
         self.assertEqual(0, (yield get))
 
@@ -154,6 +156,24 @@ class QueueGetTest(AsyncTestCase):
         for getter in getters:
             self.assertRaises(TimeoutError, getter.result)
 
+    @skipBefore35
+    @gen_test
+    def test_async_for(self):
+        q = queues.Queue()
+        for i in range(5):
+            q.put(i)
+
+        namespace = exec_test(globals(), locals(), """
+        async def f():
+            results = []
+            async for i in q:
+                results.append(i)
+                if i == 4:
+                    return results
+        """)
+        results = yield namespace['f']()
+        self.assertEqual(results, list(range(5)))
+
 
 class QueuePutTest(AsyncTestCase):
     @gen_test
@@ -176,7 +196,7 @@ class QueuePutTest(AsyncTestCase):
         self.assertEqual(0, (yield get0))
         yield q.put(1)
         self.assertEqual(1, (yield get1))
-        
+
     @gen_test
     def test_nonblocking_put_with_getters(self):
         q = queues.Queue()
@@ -208,7 +228,7 @@ class QueuePutTest(AsyncTestCase):
         put = q.put(2)
         with self.assertRaises(TimeoutError):
             yield put_timeout
-        
+
         self.assertEqual(0, q.get_nowait())
         # 1 was never put in the queue.
         self.assertEqual(2, (yield q.get()))
@@ -281,7 +301,7 @@ class QueuePutTest(AsyncTestCase):
 
 class QueueJoinTest(AsyncTestCase):
     queue_class = queues.Queue
-    
+
     def test_task_done_underflow(self):
         q = self.queue_class()
         self.assertRaises(ValueError, q.task_done)
@@ -338,7 +358,7 @@ class QueueJoinTest(AsyncTestCase):
 
 class PriorityQueueJoinTest(QueueJoinTest):
     queue_class = queues.PriorityQueue
-    
+
     @gen_test
     def test_order(self):
         q = self.queue_class(maxsize=2)
diff --git a/tornado/test/util.py b/tornado/test/util.py
index 9dd9c0ce..b8feff4f 100644
--- a/tornado/test/util.py
+++ b/tornado/test/util.py
@@ -1,8 +1,10 @@
 from __future__ import absolute_import, division, print_function, with_statement
 
 import os
+import platform
 import socket
 import sys
+import textwrap
 
 from tornado.testing import bind_unused_port
 
@@ -32,6 +34,12 @@ skipIfNoNetwork = unittest.skipIf('NO_NETWORK' in os.environ,
 skipIfNoIPv6 = unittest.skipIf(not socket.has_ipv6, 'ipv6 support not present')
 
 
+skipBefore33 = unittest.skipIf(sys.version_info < (3, 3), 'PEP 380 (yield from) not available')
+skipBefore35 = unittest.skipIf(sys.version_info < (3, 5), 'PEP 492 (async/await) not available')
+skipNotCPython = unittest.skipIf(platform.python_implementation() != 'CPython',
+                                 'Not CPython implementation')
+
+
 def refusing_port():
     """Returns a local port number that will refuse all connections.
 
@@ -50,3 +58,18 @@ def refusing_port():
     conn.close()
     server_socket.close()
     return (client_socket.close, client_addr[1])
+
+
+def exec_test(caller_globals, caller_locals, s):
+    """Execute ``s`` in a given context and return the result namespace.
+
+    Used to define functions for tests in particular python
+    versions that would be syntax errors in older versions.
+    """
+    # Flatten the real global and local namespace into our fake
+    # globals: it's all global from the perspective of code defined
+    # in s.
+    global_namespace = dict(caller_globals, **caller_locals)
+    local_namespace = {}
+    exec(textwrap.dedent(s), global_namespace, local_namespace)
+    return local_namespace