Issue python#28003: Implement PEP 525 -- Asynchronous Generators.

Author: 1st1

Include/ceval.h

@@ -25,6 +25,10 @@ PyAPI_FUNC(void) PyEval_SetProfile(Py_tracefunc, PyObject *);
 PyAPI_FUNC(void) PyEval_SetTrace(Py_tracefunc, PyObject *);
 PyAPI_FUNC(void) _PyEval_SetCoroutineWrapper(PyObject *);
 PyAPI_FUNC(PyObject *) _PyEval_GetCoroutineWrapper(void);
+PyAPI_FUNC(void) _PyEval_SetAsyncGenFirstiter(PyObject *);
+PyAPI_FUNC(PyObject *) _PyEval_GetAsyncGenFirstiter(void);
+PyAPI_FUNC(void) _PyEval_SetAsyncGenFinalizer(PyObject *);
+PyAPI_FUNC(PyObject *) _PyEval_GetAsyncGenFinalizer(void);
 #endif
 
 struct _frame; /* Avoid including frameobject.h */

Include/code.h

@@ -59,6 +59,7 @@ typedef struct {
    ``async def`` keywords) */
 #define CO_COROUTINE            0x0080
 #define CO_ITERABLE_COROUTINE   0x0100
+#define CO_ASYNC_GENERATOR      0x0200
 
 /* These are no longer used. */
 #if 0

Include/genobject.h

@@ -61,6 +61,37 @@ PyObject *_PyAIterWrapper_New(PyObject *aiter);
 PyObject *_PyCoro_GetAwaitableIter(PyObject *o);
 PyAPI_FUNC(PyObject *) PyCoro_New(struct _frame *,
     PyObject *name, PyObject *qualname);
+
+/* Asynchronous Generators */
+
+typedef struct {
+    _PyGenObject_HEAD(ag)
+    PyObject *ag_finalizer;
+
+    /* Flag is set to 1 when hooks set up by sys.set_asyncgen_hooks
+       were called on the generator, to avoid calling them more
+       than once. */
+    int ag_hooks_inited;
+
+    /* Flag is set to 1 when aclose() is called for the first time, or
+       when a StopAsyncIteration exception is raised. */
+    int ag_closed;
+} PyAsyncGenObject;
+
+PyAPI_DATA(PyTypeObject) PyAsyncGen_Type;
+PyAPI_DATA(PyTypeObject) _PyAsyncGenASend_Type;
+PyAPI_DATA(PyTypeObject) _PyAsyncGenWrappedValue_Type;
+PyAPI_DATA(PyTypeObject) _PyAsyncGenAThrow_Type;
+
+PyAPI_FUNC(PyObject *) PyAsyncGen_New(struct _frame *,
+    PyObject *name, PyObject *qualname);
+
+#define PyAsyncGen_CheckExact(op) (Py_TYPE(op) == &PyAsyncGen_Type)
+
+PyObject *_PyAsyncGenValueWrapperNew(PyObject *);
+
+int PyAsyncGen_ClearFreeLists(void);
+
 #endif
 
 #undef _PyGenObject_HEAD

Include/pylifecycle.h

@@ -107,6 +107,7 @@ PyAPI_FUNC(void) _PyGC_Fini(void);
 PyAPI_FUNC(void) PySlice_Fini(void);
 PyAPI_FUNC(void) _PyType_Fini(void);
 PyAPI_FUNC(void) _PyRandom_Fini(void);
+PyAPI_FUNC(void) PyAsyncGen_Fini(void);
 
 PyAPI_DATA(PyThreadState *) _Py_Finalizing;
 #endif

Include/pystate.h

@@ -148,6 +148,9 @@ typedef struct _ts {
     Py_ssize_t co_extra_user_count;
     freefunc co_extra_freefuncs[MAX_CO_EXTRA_USERS];
 
+    PyObject *async_gen_firstiter;
+    PyObject *async_gen_finalizer;
+
     /* XXX signal handlers should also be here */
 
 } PyThreadState;

Include/symtable.h

@@ -48,6 +48,7 @@ typedef struct _symtable_entry {
     unsigned ste_child_free : 1;  /* true if a child block has free vars,
                                      including free refs to globals */
     unsigned ste_generator : 1;   /* true if namespace is a generator */
+    unsigned ste_coroutine : 1;   /* true if namespace is a coroutine */
     unsigned ste_varargs : 1;     /* true if block has varargs */
     unsigned ste_varkeywords : 1; /* true if block has varkeywords */
     unsigned ste_returns_value : 1;  /* true if namespace uses return with

Lib/asyncio/base_events.py

@@ -13,7 +13,6 @@
 to modify the meaning of the API call itself.
 """
 
-
 import collections
 import concurrent.futures
 import heapq
@@ -28,6 +27,7 @@
 import traceback
 import sys
 import warnings
+import weakref
 
 from . import compat
 from . import coroutines
@@ -242,6 +242,13 @@ def __init__(self):
         self._task_factory = None
         self._coroutine_wrapper_set = False
 
+        # A weak set of all asynchronous generators that are being iterated
+        # by the loop.
+        self._asyncgens = weakref.WeakSet()
+
+        # Set to True when `loop.shutdown_asyncgens` is called.
+        self._asyncgens_shutdown_called = False
+
     def __repr__(self):
         return ('<%s running=%s closed=%s debug=%s>'
                 % (self.__class__.__name__, self.is_running(),
@@ -333,13 +340,56 @@ def _check_closed(self):
         if self._closed:
             raise RuntimeError('Event loop is closed')
 
+    def _asyncgen_finalizer_hook(self, agen):
+        self._asyncgens.discard(agen)
+        if not self.is_closed():
+            self.create_task(agen.aclose())
+
+    def _asyncgen_firstiter_hook(self, agen):
+        if self._asyncgens_shutdown_called:
+            warnings.warn(
+                "asynchronous generator {!r} was scheduled after "
+                "loop.shutdown_asyncgens() call".format(agen),
+                ResourceWarning, source=self)
+
+        self._asyncgens.add(agen)
+
+    @coroutine
+    def shutdown_asyncgens(self):
+        """Shutdown all active asynchronous generators."""
+        self._asyncgens_shutdown_called = True
+
+        if not len(self._asyncgens):
+            return
+
+        closing_agens = list(self._asyncgens)
+        self._asyncgens.clear()
+
+        shutdown_coro = tasks.gather(
+            *[ag.aclose() for ag in closing_agens],
+            return_exceptions=True,
+            loop=self)
+
+        results = yield from shutdown_coro
+        for result, agen in zip(results, closing_agens):
+            if isinstance(result, Exception):
+                self.call_exception_handler({
+                    'message': 'an error occurred during closing of '
+                               'asynchronous generator {!r}'.format(agen),
+                    'exception': result,
+                    'asyncgen': agen
+                })
+
     def run_forever(self):
         """Run until stop() is called."""
         self._check_closed()
         if self.is_running():
             raise RuntimeError('Event loop is running.')
         self._set_coroutine_wrapper(self._debug)
         self._thread_id = threading.get_ident()
+        old_agen_hooks = sys.get_asyncgen_hooks()
+        sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook,
+                               finalizer=self._asyncgen_finalizer_hook)
         try:
             while True:
                 self._run_once()
@@ -349,6 +399,7 @@ def run_forever(self):
             self._stopping = False
             self._thread_id = None
             self._set_coroutine_wrapper(False)
+            sys.set_asyncgen_hooks(*old_agen_hooks)
 
     def run_until_complete(self, future):
         """Run until the Future is done.
@@ -1179,7 +1230,9 @@ def call_exception_handler(self, context):
         - 'handle' (optional): Handle instance;
         - 'protocol' (optional): Protocol instance;
         - 'transport' (optional): Transport instance;
-        - 'socket' (optional): Socket instance.
+        - 'socket' (optional): Socket instance;
+        - 'asyncgen' (optional): Asynchronous generator that caused
+                                 the exception.
 
         New keys maybe introduced in the future.
 

Lib/asyncio/coroutines.py

@@ -276,7 +276,10 @@ def _format_coroutine(coro):
     try:
         coro_code = coro.gi_code
     except AttributeError:
-        coro_code = coro.cr_code
+        try:
+            coro_code = coro.cr_code
+        except AttributeError:
+            return repr(coro)
 
     try:
         coro_frame = coro.gi_frame

Lib/asyncio/events.py

@@ -248,6 +248,10 @@ def close(self):
         """
         raise NotImplementedError
 
+    def shutdown_asyncgens(self):
+        """Shutdown all active asynchronous generators."""
+        raise NotImplementedError
+
     # Methods scheduling callbacks.  All these return Handles.
 
     def _timer_handle_cancelled(self, handle):

Lib/dis.py

@@ -87,6 +87,7 @@ def distb(tb=None, *, file=None):
     64: "NOFREE",
    128: "COROUTINE",
    256: "ITERABLE_COROUTINE",
+   512: "ASYNC_GENERATOR",
 }
 
 def pretty_flags(flags):

Lib/inspect.py

@@ -185,6 +185,13 @@ def iscoroutinefunction(object):
     return bool((isfunction(object) or ismethod(object)) and
                 object.__code__.co_flags & CO_COROUTINE)
 
+def isasyncgenfunction(object):
+    return bool((isfunction(object) or ismethod(object)) and
+                object.__code__.co_flags & CO_ASYNC_GENERATOR)
+
+def isasyncgen(object):
+    return isinstance(object, types.AsyncGeneratorType)
+
 def isgenerator(object):
     """Return true if the object is a generator.