From b5505b4e73a2112059e95766fc3cab97db55b7e0 Mon Sep 17 00:00:00 2001 From: coolreader18 <33094578+coolreader18@users.noreply.github.com> Date: Thu, 7 Nov 2019 22:32:27 -0600 Subject: [PATCH] Add asyncio.run from CPython 3.7 --- Lib/asyncio/__init__.py | 4 +++ Lib/asyncio/runners.py | 72 ++++++++++++++++++++++++++++++++++++++ Lib/asyncio/tasks.py | 77 +++++++++++++++++++++++++++++++++++++++++ vm/src/exceptions.rs | 7 ++++ vm/src/stdlib/io.rs | 5 +++ 5 files changed, 165 insertions(+) create mode 100644 Lib/asyncio/runners.py diff --git a/Lib/asyncio/__init__.py b/Lib/asyncio/__init__.py index cd88f5387..ff69378ba 100644 --- a/Lib/asyncio/__init__.py +++ b/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__ + diff --git a/Lib/asyncio/runners.py b/Lib/asyncio/runners.py new file mode 100644 index 000000000..c3a696ef5 --- /dev/null +++ b/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, + }) diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py index 5a43ef257..d31c0109f 100644 --- a/Lib/asyncio/tasks.py +++ b/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 diff --git a/vm/src/exceptions.rs b/vm/src/exceptions.rs index ab3be0a0c..ffde7f79e 100644 --- a/vm/src/exceptions.rs +++ b/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) + }), + }); } diff --git a/vm/src/stdlib/io.rs b/vm/src/stdlib/io.rs index 87ab1dcd4..fa3e1a918 100644 --- a/vm/src/stdlib/io.rs +++ b/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