asyncio: sync with Tulip, add a new asyncio.coroutines module

Author: vstinner

Lib/asyncio/__init__.py

@@ -18,6 +18,7 @@
         import _overlapped  # Will also be exported.
 
 # This relies on each of the submodules having an __all__ variable.
+from .coroutines import *
 from .events import *
 from .futures import *
 from .locks import *
@@ -34,7 +35,8 @@
     from .unix_events import *  # pragma: no cover
 
 
-__all__ = (events.__all__ +
+__all__ = (coroutines.__all__ +
+           events.__all__ +
            futures.__all__ +
            locks.__all__ +
            protocols.__all__ +

Lib/asyncio/base_events.py

@@ -26,9 +26,11 @@
 import os
 import sys
 
+from . import coroutines
 from . import events
 from . import futures
 from . import tasks
+from .coroutines import coroutine
 from .log import logger
 
 
@@ -118,7 +120,7 @@ def _wakeup(self):
             if not waiter.done():
                 waiter.set_result(waiter)
 
-    @tasks.coroutine
+    @coroutine
     def wait_closed(self):
         if self.sockets is None or self.waiters is None:
             return
@@ -175,7 +177,7 @@ def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
         """Create write pipe transport."""
         raise NotImplementedError
 
-    @tasks.coroutine
+    @coroutine
     def _make_subprocess_transport(self, protocol, args, shell,
                                    stdin, stdout, stderr, bufsize,
                                    extra=None, **kwargs):
@@ -298,7 +300,7 @@ def call_later(self, delay, callback, *args):
 
     def call_at(self, when, callback, *args):
         """Like call_later(), but uses an absolute time."""
-        if tasks.iscoroutinefunction(callback):
+        if coroutines.iscoroutinefunction(callback):
             raise TypeError("coroutines cannot be used with call_at()")
         if self._debug:
             self._assert_is_current_event_loop()
@@ -324,7 +326,7 @@ def call_soon(self, callback, *args):
         return handle
 
     def _call_soon(self, callback, args, check_loop):
-        if tasks.iscoroutinefunction(callback):
+        if coroutines.iscoroutinefunction(callback):
             raise TypeError("coroutines cannot be used with call_soon()")
         if self._debug and check_loop:
             self._assert_is_current_event_loop()
@@ -361,7 +363,7 @@ def call_soon_threadsafe(self, callback, *args):
         return handle
 
     def run_in_executor(self, executor, callback, *args):
-        if tasks.iscoroutinefunction(callback):
+        if coroutines.iscoroutinefunction(callback):
             raise TypeError("coroutines cannot be used with run_in_executor()")
         if isinstance(callback, events.Handle):
             assert not args
@@ -389,7 +391,7 @@ def getaddrinfo(self, host, port, *,
     def getnameinfo(self, sockaddr, flags=0):
         return self.run_in_executor(None, socket.getnameinfo, sockaddr, flags)
 
-    @tasks.coroutine
+    @coroutine
     def create_connection(self, protocol_factory, host=None, port=None, *,
                           ssl=None, family=0, proto=0, flags=0, sock=None,
                           local_addr=None, server_hostname=None):
@@ -505,7 +507,7 @@ def create_connection(self, protocol_factory, host=None, port=None, *,
             sock, protocol_factory, ssl, server_hostname)
         return transport, protocol
 
-    @tasks.coroutine
+    @coroutine
     def _create_connection_transport(self, sock, protocol_factory, ssl,
                                      server_hostname):
         protocol = protocol_factory()
@@ -521,7 +523,7 @@ def _create_connection_transport(self, sock, protocol_factory, ssl,
         yield from waiter
         return transport, protocol
 
-    @tasks.coroutine
+    @coroutine
     def create_datagram_endpoint(self, protocol_factory,
                                  local_addr=None, remote_addr=None, *,
                                  family=0, proto=0, flags=0):
@@ -593,7 +595,7 @@ def create_datagram_endpoint(self, protocol_factory,
         transport = self._make_datagram_transport(sock, protocol, r_addr)
         return transport, protocol
 
-    @tasks.coroutine
+    @coroutine
     def create_server(self, protocol_factory, host=None, port=None,
                       *,
                       family=socket.AF_UNSPEC,
@@ -672,23 +674,23 @@ def create_server(self, protocol_factory, host=None, port=None,
             self._start_serving(protocol_factory, sock, ssl, server)
         return server
 
-    @tasks.coroutine
+    @coroutine
     def connect_read_pipe(self, protocol_factory, pipe):
         protocol = protocol_factory()
         waiter = futures.Future(loop=self)
         transport = self._make_read_pipe_transport(pipe, protocol, waiter)
         yield from waiter
         return transport, protocol
 
-    @tasks.coroutine
+    @coroutine
     def connect_write_pipe(self, protocol_factory, pipe):
         protocol = protocol_factory()
         waiter = futures.Future(loop=self)
         transport = self._make_write_pipe_transport(pipe, protocol, waiter)
         yield from waiter
         return transport, protocol
 
-    @tasks.coroutine
+    @coroutine
     def subprocess_shell(self, protocol_factory, cmd, *, stdin=subprocess.PIPE,
                          stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                          universal_newlines=False, shell=True, bufsize=0,
@@ -706,7 +708,7 @@ def subprocess_shell(self, protocol_factory, cmd, *, stdin=subprocess.PIPE,
             protocol, cmd, True, stdin, stdout, stderr, bufsize, **kwargs)
         return transport, protocol
 
-    @tasks.coroutine
+    @coroutine
     def subprocess_exec(self, protocol_factory, program, *args,
                         stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE, universal_newlines=False,

Lib/asyncio/base_subprocess.py

@@ -2,8 +2,8 @@
 import subprocess
 
 from . import protocols
-from . import tasks
 from . import transports
+from .coroutines import coroutine
 
 
 class BaseSubprocessTransport(transports.SubprocessTransport):
@@ -65,7 +65,7 @@ def terminate(self):
     def kill(self):
         self._proc.kill()
 
-    @tasks.coroutine
+    @coroutine
     def _post_init(self):
         proc = self._proc
         loop = self._loop

Lib/asyncio/coroutines.py

@@ -0,0 +1,140 @@
+__all__ = ['coroutine',
+           'iscoroutinefunction', 'iscoroutine']
+
+import functools
+import inspect
+import os
+import sys
+import traceback
+
+from . import events
+from . import futures
+from .log import logger
+
+# If you set _DEBUG to true, @coroutine will wrap the resulting
+# generator objects in a CoroWrapper instance (defined below).  That
+# instance will log a message when the generator is never iterated
+# over, which may happen when you forget to use "yield from" with a
+# coroutine call.  Note that the value of the _DEBUG flag is taken
+# when the decorator is used, so to be of any use it must be set
+# before you define your coroutines.  A downside of using this feature
+# is that tracebacks show entries for the CoroWrapper.__next__ method
+# when _DEBUG is true.
+_DEBUG = (not sys.flags.ignore_environment
+          and bool(os.environ.get('PYTHONASYNCIODEBUG')))
+
+_PY35 = (sys.version_info >= (3, 5))
+
+class CoroWrapper:
+    # Wrapper for coroutine in _DEBUG mode.
+
+    def __init__(self, gen, func):
+        assert inspect.isgenerator(gen), gen
+        self.gen = gen
+        self.func = func
+        self._source_traceback = traceback.extract_stack(sys._getframe(1))
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        return next(self.gen)
+
+    def send(self, *value):
+        # We use `*value` because of a bug in CPythons prior
+        # to 3.4.1. See issue #21209 and test_yield_from_corowrapper
+        # for details.  This workaround should be removed in 3.5.0.
+        if len(value) == 1:
+            value = value[0]
+        return self.gen.send(value)
+
+    def throw(self, exc):
+        return self.gen.throw(exc)
+
+    def close(self):
+        return self.gen.close()
+
+    @property
+    def gi_frame(self):
+        return self.gen.gi_frame
+
+    @property
+    def gi_running(self):
+        return self.gen.gi_running
+
+    @property
+    def gi_code(self):
+        return self.gen.gi_code
+
+    def __del__(self):
+        # Be careful accessing self.gen.frame -- self.gen might not exist.
+        gen = getattr(self, 'gen', None)
+        frame = getattr(gen, 'gi_frame', None)
+        if frame is not None and frame.f_lasti == -1:
+            func = events._format_callback(self.func, ())
+            tb = ''.join(traceback.format_list(self._source_traceback))
+            message = ('Coroutine %s was never yielded from\n'
+                       'Coroutine object created at (most recent call last):\n'
+                       '%s'
+                       % (func, tb.rstrip()))
+            logger.error(message)
+
+
+def coroutine(func):
+    """Decorator to mark coroutines.
+
+    If the coroutine is not yielded from before it is destroyed,
+    an error message is logged.
+    """
+    if inspect.isgeneratorfunction(func):
+        coro = func
+    else:
+        @functools.wraps(func)
+        def coro(*args, **kw):
+            res = func(*args, **kw)
+            if isinstance(res, futures.Future) or inspect.isgenerator(res):
+                res = yield from res
+            return res
+
+    if not _DEBUG:
+        wrapper = coro
+    else:
+        @functools.wraps(func)
+        def wrapper(*args, **kwds):
+            w = CoroWrapper(coro(*args, **kwds), func)
+            if w._source_traceback:
+                del w._source_traceback[-1]
+            w.__name__ = func.__name__
+            if _PY35:
+                w.__qualname__ = func.__qualname__
+            w.__doc__ = func.__doc__
+            return w
+
+    wrapper._is_coroutine = True  # For iscoroutinefunction().
+    return wrapper
+
+
+def iscoroutinefunction(func):
+    """Return True if func is a decorated coroutine function."""
+    return getattr(func, '_is_coroutine', False)
+
+
+def iscoroutine(obj):
+    """Return True if obj is a coroutine object."""
+    return isinstance(obj, CoroWrapper) or inspect.isgenerator(obj)
+
+
+def _format_coroutine(coro):
+    assert iscoroutine(coro)
+    if _PY35:
+        coro_name = coro.__qualname__
+    else:
+        coro_name = coro.__name__
+
+    filename = coro.gi_code.co_filename
+    if coro.gi_frame is not None:
+        lineno = coro.gi_frame.f_lineno
+        return '%s() at %s:%s' % (coro_name, filename, lineno)
+    else:
+        lineno = coro.gi_code.co_firstlineno
+        return '%s() done at %s:%s' % (coro_name, filename, lineno)

Lib/asyncio/locks.py

@@ -6,7 +6,7 @@
 
 from . import events
 from . import futures
-from . import tasks
+from .coroutines import coroutine
 
 
 class _ContextManager:
@@ -112,7 +112,7 @@ def locked(self):
         """Return True if lock is acquired."""
         return self._locked
 
-    @tasks.coroutine
+    @coroutine
     def acquire(self):
         """Acquire a lock.
 
@@ -225,7 +225,7 @@ def clear(self):
         to true again."""
         self._value = False
 
-    @tasks.coroutine
+    @coroutine
     def wait(self):
         """Block until the internal flag is true.
 
@@ -278,7 +278,7 @@ def __repr__(self):
             extra = '{},waiters:{}'.format(extra, len(self._waiters))
         return '<{} [{}]>'.format(res[1:-1], extra)
 
-    @tasks.coroutine
+    @coroutine
     def wait(self):
         """Wait until notified.
 
@@ -306,7 +306,7 @@ def wait(self):
         finally:
             yield from self.acquire()
 
-    @tasks.coroutine
+    @coroutine
     def wait_for(self, predicate):
         """Wait until a predicate becomes true.
 
@@ -402,7 +402,7 @@ def locked(self):
         """Returns True if semaphore can not be acquired immediately."""
         return self._value == 0
 
-    @tasks.coroutine
+    @coroutine
     def acquire(self):
         """Acquire a semaphore.