Add asyncio.run from CPython 3.7

Lib/asyncio/__init__.py

@@ -1,5 +1,6 @@
 """The asyncio package, tracking PEP 3156."""
 
+# flake8: noqa
 import sys
 
 import selectors
@@ -11,6 +12,7 @@ if sys.platform == 'win32' and False:
     except ImportError:
         import _overlapped  # Will also be exported.
 
+
 # This relies on each of the submodules having an __all__ variable.
 from .base_events import *
 from .coroutines import *
@@ -18,6 +20,7 @@ from .events import *
 from .futures import *
 from .locks import *
 from .protocols import *
+from .runners import *
 from .queues import *
 from .streams import *
 from .subprocess import *
@@ -30,6 +33,7 @@ __all__ = (base_events.__all__ +
            futures.__all__ +
            locks.__all__ +
            protocols.__all__ +
+           runners.__all__ +
            queues.__all__ +
            streams.__all__ +
            subprocess.__all__ +

Lib/asyncio/runners.py

@@ -0,0 +1,72 @@
+__all__ = ['run']
+
+from . import coroutines
+from . import events
+from . import tasks
+
+
+def run(main, *, debug=False):
+    """Run a coroutine.
+
+    This function runs the passed coroutine, taking care of
+    managing the asyncio event loop and finalizing asynchronous
+    generators.
+
+    This function cannot be called when another asyncio event loop is
+    running in the same thread.
+
+    If debug is True, the event loop will be run in debug mode.
+
+    This function always creates a new event loop and closes it at the end.
+    It should be used as a main entry point for asyncio programs, and should
+    ideally only be called once.
+
+    Example:
+
+        async def main():
+            await asyncio.sleep(1)
+            print('hello')
+
+        asyncio.run(main())
+    """
+    if events._get_running_loop() is not None:
+        raise RuntimeError(
+            "asyncio.run() cannot be called from a running event loop")
+
+    if not coroutines.iscoroutine(main):
+        raise ValueError("a coroutine was expected, got {!r}".format(main))
+
+    loop = events.new_event_loop()
+    try:
+        events.set_event_loop(loop)
+        loop.set_debug(debug)
+        return loop.run_until_complete(main)
+    finally:
+        try:
+            _cancel_all_tasks(loop)
+            loop.run_until_complete(loop.shutdown_asyncgens())
+        finally:
+            events.set_event_loop(None)
+            loop.close()
+
+
+def _cancel_all_tasks(loop):
+    to_cancel = tasks.all_tasks(loop)
+    if not to_cancel:
+        return
+
+    for task in to_cancel:
+        task.cancel()
+
+    loop.run_until_complete(
+        tasks.gather(*to_cancel, loop=loop, return_exceptions=True))
+
+    for task in to_cancel:
+        if task.cancelled():
+            continue
+        if task.exception() is not None:
+            loop.call_exception_handler({
+                'message': 'unhandled exception during asyncio.run() shutdown',
+                'exception': task.exception(),
+                'task': task,
+            })

Lib/asyncio/tasks.py

@@ -20,6 +20,30 @@ from . import futures
 from .coroutines import coroutine
 
 
+def current_task(loop=None):
+    """Return a currently executed task."""
+    if loop is None:
+        loop = events.get_running_loop()
+    return _current_tasks.get(loop)
+
+
+def all_tasks(loop=None):
+    """Return a set of all tasks for the loop."""
+    if loop is None:
+        loop = events.get_running_loop()
+    return {t for t in _all_tasks
+            if futures._get_loop(t) is loop and not t.done()}
+
+
+def _all_tasks_compat(loop=None):
+    # Different from "all_task()" by returning *all* Tasks, including
+    # the completed ones.  Used to implement deprecated "Tasks.all_task()"
+    # method.
+    if loop is None:
+        loop = events.get_event_loop()
+    return {t for t in _all_tasks if futures._get_loop(t) is loop}
+
+
 class Task(futures.Future):
     """A coroutine wrapped in a Future."""
 
@@ -707,3 +731,56 @@ def run_coroutine_threadsafe(coro, loop):
 
     loop.call_soon_threadsafe(callback)
     return future
+
+
+# WeakSet containing all alive tasks.
+_all_tasks = weakref.WeakSet()
+
+# Dictionary containing tasks that are currently active in
+# all running event loops.  {EventLoop: Task}
+_current_tasks = {}
+
+
+def _register_task(task):
+    """Register a new task in asyncio as executed by loop."""
+    _all_tasks.add(task)
+
+
+def _enter_task(loop, task):
+    current_task = _current_tasks.get(loop)
+    if current_task is not None:
+        raise RuntimeError(f"Cannot enter into task {task!r} while another "
+                           f"task {current_task!r} is being executed.")
+    _current_tasks[loop] = task
+
+
+def _leave_task(loop, task):
+    current_task = _current_tasks.get(loop)
+    if current_task is not task:
+        raise RuntimeError(f"Leaving task {task!r} does not match "
+                           f"the current task {current_task!r}.")
+    del _current_tasks[loop]
+
+
+def _unregister_task(task):
+    """Unregister a task."""
+    _all_tasks.discard(task)
+
+
+_py_register_task = _register_task
+_py_unregister_task = _unregister_task
+_py_enter_task = _enter_task
+_py_leave_task = _leave_task
+
+
+try:
+    from _asyncio import (_register_task, _unregister_task,
+                          _enter_task, _leave_task,
+                          _all_tasks, _current_tasks)
+except ImportError:
+    pass
+else:
+    _c_register_task = _register_task
+    _c_unregister_task = _unregister_task
+    _c_enter_task = _enter_task
+    _c_leave_task = _leave_task

vm/src/exceptions.rs

@@ -1,4 +1,5 @@
 use crate::function::PyFuncArgs;
+use crate::obj::objiter;
 use crate::obj::objtraceback::PyTracebackRef;
 use crate::obj::objtuple::{PyTuple, PyTupleRef};
 use crate::obj::objtype;
@@ -551,4 +552,10 @@ pub fn init(context: &PyContext) {
     extend_class!(context, import_error_type, {
         "__init__" => context.new_rustfunc(import_error_init)
     });
+
+    extend_class!(context, &context.exceptions.stop_iteration, {
+        "value" => context.new_rustfunc(|obj: PyObjectRef, vm: &VirtualMachine| {
+            objiter::stop_iter_value(vm, &obj)
+        }),
+    });
 }

vm/src/stdlib/io.rs

@@ -166,6 +166,10 @@ impl PyStringIORef {
             None => Err(vm.new_value_error("Error Performing Operation".to_string())),
         }
     }
+
+    fn close(self, _vm: &VirtualMachine) {
+        // TODO: discard the text buffer on close
+    }
 }
 
 fn string_io_new(
@@ -759,6 +763,7 @@ pub fn make_module(vm: &VirtualMachine) -> PyObjectRef {
         "getvalue" => ctx.new_rustfunc(PyStringIORef::getvalue),
         "tell" => ctx.new_rustfunc(PyStringIORef::tell),
         "readline" => ctx.new_rustfunc(PyStringIORef::readline),
+        "close" => ctx.new_rustfunc(PyStringIORef::close),
     });
 
     //BytesIO: in-memory bytes