Allow to run async code in a loop that already running

_pydevd_bundle/pydevconsole_code.py

@@ -212,33 +212,8 @@ def __call__(self, source, filename="<input>", symbol="single"):
 
 __all__ = ["InteractiveInterpreter", "InteractiveConsole", "interact", "compile_command"]
 
-from _pydev_bundle._pydev_saved_modules import threading
 
-
-class _EvalAwaitInNewEventLoop(threading.Thread):
-    def __init__(self, compiled, updated_globals, updated_locals):
-        threading.Thread.__init__(self)
-        self.daemon = True
-        self._compiled = compiled
-        self._updated_globals = updated_globals
-        self._updated_locals = updated_locals
-
-        # Output
-        self.evaluated_value = None
-        self.exc = None
-
-    async def _async_func(self):
-        return await eval(self._compiled, self._updated_locals, self._updated_globals)
-
-    def run(self):
-        try:
-            import asyncio
-
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            self.evaluated_value = asyncio.run(self._async_func())
-        except:
-            self.exc = sys.exc_info()
+from _pydevd_bundle.pydevd_async_utils import eval_async_coro
 
 
 class InteractiveInterpreter:
@@ -321,13 +296,7 @@ def runcode(self, code):
                 is_async = inspect.CO_COROUTINE & code.co_flags == inspect.CO_COROUTINE
 
             if is_async:
-                t = _EvalAwaitInNewEventLoop(code, self.locals, None)
-                t.start()
-                t.join()
-
-                if t.exc:
-                    raise t.exc[1].with_traceback(t.exc[2])
-
+                eval_async_coro(code, self.locals, None)
             else:
                 exec(code, self.locals)
         except SystemExit:

_pydevd_bundle/pydevd_async_utils.py

@@ -0,0 +1,45 @@
+__all__ = ["eval_async_coro"]
+
+import types
+
+
+def _get_current_loop():
+    import asyncio
+
+    try:
+        return asyncio.get_running_loop()
+    except (RuntimeError, AttributeError):
+        return asyncio.new_event_loop()
+
+
+def _prepare_coro(coro, _locals, _globals):
+    if isinstance(coro, types.CodeType):
+        return eval(coro, _locals, _globals)
+
+    return coro
+
+
+def eval_async_coro(coro, _locals, _globals):
+    import asyncio
+
+    coro = _prepare_coro(coro, _locals, _globals)
+    loop = _get_current_loop()
+
+    if not loop.is_running():
+        return loop.run_until_complete(coro)
+
+    current = asyncio.current_task(loop)
+
+    t = loop.create_task(coro)
+
+    try:
+        if current is not None:
+            asyncio._leave_task(loop, current)
+
+        while not t.done():
+            loop._run_once()
+
+        return t.result()
+    finally:
+        if current is not None:
+            asyncio._enter_task(loop, current)

_pydevd_bundle/pydevd_console.py

@@ -2,12 +2,13 @@
 
 import sys
 import traceback
-from _pydevd_bundle.pydevconsole_code import InteractiveConsole, _EvalAwaitInNewEventLoop
+from _pydevd_bundle.pydevconsole_code import InteractiveConsole
 from _pydev_bundle import _pydev_completer
 from _pydev_bundle.pydev_console_utils import BaseInterpreterInterface, BaseStdIn
 from _pydev_bundle.pydev_imports import Exec
 from _pydev_bundle.pydev_override import overrides
 from _pydevd_bundle import pydevd_save_locals
+from _pydevd_bundle.pydevd_async_utils import eval_async_coro
 from _pydevd_bundle.pydevd_io import IOBuf
 from pydevd_tracing import get_exception_traceback_str
 from _pydevd_bundle.pydevd_xml import make_valid_xml_value
@@ -160,14 +161,10 @@ def runcode(self, code):
                 is_async = inspect.CO_COROUTINE & code.co_flags == inspect.CO_COROUTINE
 
             if is_async:
-                t = _EvalAwaitInNewEventLoop(code, updated_globals, updated_locals)
-                t.start()
-                t.join()
-
-                update_globals_and_locals(updated_globals, initial_globals, self.frame)
-                if t.exc:
-                    raise t.exc[1].with_traceback(t.exc[2])
-
+                try:
+                    eval_async_coro(code, updated_globals, updated_locals)
+                finally:
+                    update_globals_and_locals(updated_globals, initial_globals, self.frame)
             else:
                 try:
                     exec(code, updated_globals, updated_locals)

_pydevd_bundle/pydevd_vars.py

@@ -3,6 +3,8 @@
 """
 
 import pickle
+
+from _pydevd_bundle.pydevd_async_utils import eval_async_coro
 from _pydevd_bundle.pydevd_constants import get_frame, get_current_thread_id, iter_chars, silence_warnings_decorator, get_global_debugger
 
 from _pydevd_bundle.pydevd_xml import ExceptionOnEvaluate, get_type, var_to_xml
@@ -253,18 +255,7 @@ def eval_in_context(expression, global_vars, local_vars, py_db=None):
         is_async = inspect.CO_COROUTINE & compiled.co_flags == inspect.CO_COROUTINE
 
         if is_async:
-            if py_db is None:
-                py_db = get_global_debugger()
-                if py_db is None:
-                    raise RuntimeError("Cannot evaluate async without py_db.")
-            t = _EvalAwaitInNewEventLoop(py_db, compiled, global_vars, local_vars)
-            t.start()
-            t.join()
-
-            if t.exc:
-                raise t.exc[1].with_traceback(t.exc[2])
-            else:
-                result = t.evaluated_value
+            result = eval_async_coro(compiled, local_vars, global_vars)
         else:
             result = eval(compiled, global_vars, local_vars)
     except (Exception, KeyboardInterrupt):
@@ -416,31 +407,6 @@ def _compile_as_exec(expression):
         return compile(expression_to_evaluate, "<string>", "exec")
 
 
-class _EvalAwaitInNewEventLoop(PyDBDaemonThread):
-    def __init__(self, py_db, compiled, updated_globals, updated_locals):
-        PyDBDaemonThread.__init__(self, py_db)
-        self._compiled = compiled
-        self._updated_globals = updated_globals
-        self._updated_locals = updated_locals
-
-        # Output
-        self.evaluated_value = None
-        self.exc = None
-
-    async def _async_func(self):
-        return await eval(self._compiled, self._updated_locals, self._updated_globals)
-
-    def _on_run(self):
-        try:
-            import asyncio
-
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            self.evaluated_value = asyncio.run(self._async_func())
-        except:
-            self.exc = sys.exc_info()
-
-
 @_evaluate_with_timeouts
 def evaluate_expression(py_db, frame, expression, is_exec):
     """
@@ -550,12 +516,7 @@ def method():
                     compiled = _compile_as_exec(expression)
                     is_async = inspect.CO_COROUTINE & compiled.co_flags == inspect.CO_COROUTINE
                     if is_async:
-                        t = _EvalAwaitInNewEventLoop(py_db, compiled, updated_globals, updated_locals)
-                        t.start()
-                        t.join()
-
-                        if t.exc:
-                            raise t.exc[1].with_traceback(t.exc[2])
+                        eval_async_coro(compiled, updated_globals, updated_locals)
                     else:
                         Exec(compiled, updated_globals, updated_locals)
                 finally:
@@ -564,14 +525,7 @@ def method():
             else:
                 is_async = inspect.CO_COROUTINE & compiled.co_flags == inspect.CO_COROUTINE
                 if is_async:
-                    t = _EvalAwaitInNewEventLoop(py_db, compiled, updated_globals, updated_locals)
-                    t.start()
-                    t.join()
-
-                    if t.exc:
-                        raise t.exc[1].with_traceback(t.exc[2])
-                    else:
-                        result = t.evaluated_value
+                    result = eval_async_coro(compiled, updated_globals, updated_locals)
                 else:
                     result = eval(compiled, updated_globals, updated_locals)
                 if result is not None:  # Only print if it's not None (as python does)