Drop old (and deluded) "streaming" cruft

tractor/_debug.py

@@ -2,6 +2,7 @@
 Multi-core debugging for da peeps!
 
 """
+from __future__ import annotations
 import bdb
 import sys
 from functools import partial
@@ -27,12 +28,14 @@ from ._exceptions import is_multi_cancelled
 try:
     # wtf: only exported when installed in dev mode?
     import pdbpp
+
 except ImportError:
     # pdbpp is installed in regular mode...it monkey patches stuff
     import pdb
     assert pdb.xpm, "pdbpp is not installed?"  # type: ignore
     pdbpp = pdb
 
+
 log = get_logger(__name__)
 
 
@@ -92,6 +95,23 @@ class PdbwTeardown(pdbpp.Pdb):
             _pdb_release_hook()
 
 
+def _mk_pdb() -> PdbwTeardown:
+
+    # XXX: setting these flags on the pdb instance are absolutely
+    # critical to having ctrl-c work in the ``trio`` standard way!  The
+    # stdlib's pdb supports entering the current sync frame on a SIGINT,
+    # with ``trio`` we pretty much never want this and if we did we can
+    # handle it in the ``tractor`` task runtime.
+    # global pdb
+
+    pdb = PdbwTeardown()
+    pdb.nosigint = True
+    pdb.allow_kbdint = True
+    opts = (allow_kbdint, nosigint) = pdb.allow_kbdint, pdb.nosigint
+    print(f'`pdbp` was configured with {opts}')
+    return pdb
+
+
 # TODO: will be needed whenever we get to true remote debugging.
 # XXX see https://github.com/goodboy/tractor/issues/130
 
@@ -461,21 +481,6 @@ async def _breakpoint(
     debug_func(actor)
 
 
-def _mk_pdb() -> PdbwTeardown:
-
-    # XXX: setting these flags on the pdb instance are absolutely
-    # critical to having ctrl-c work in the ``trio`` standard way!  The
-    # stdlib's pdb supports entering the current sync frame on a SIGINT,
-    # with ``trio`` we pretty much never want this and if we did we can
-    # handle it in the ``tractor`` task runtime.
-
-    pdb = PdbwTeardown()
-    pdb.allow_kbdint = True
-    pdb.nosigint = True
-
-    return pdb
-
-
 def _set_trace(actor=None):
     pdb = _mk_pdb()
 

tractor/_spawn.py

@@ -192,6 +192,7 @@ async def new_proc(
     _runtime_vars: Dict[str, Any],  # serialized and sent to _child
 
     *,
+
     infect_asyncio: bool = False,
     task_status: TaskStatus[Portal] = trio.TASK_STATUS_IGNORED
 

tractor/to_asyncio.py

@@ -1,18 +1,21 @@
-"""
+'''
 Infection apis for ``asyncio`` loops running ``trio`` using guest mode.
-"""
+
+'''
 import asyncio
+from contextlib import asynccontextmanager as acm
 import inspect
 from typing import (
     Any,
     Callable,
     AsyncIterator,
     Awaitable,
+    Optional,
 )
 
 import trio
 
-from .log import get_logger
+from .log import get_logger, get_console_log
 from ._state import current_actor
 
 log = get_logger(__name__)
@@ -21,40 +24,21 @@ log = get_logger(__name__)
 __all__ = ['run_task', 'run_as_asyncio_guest']
 
 
-async def run_coro(
-    to_trio: trio.MemorySendChannel,
-    coro: Awaitable,
-) -> None:
-    """Await ``coro`` and relay result back to ``trio``.
-    """
-    to_trio.send_nowait(await coro)
-
-
-async def consume_asyncgen(
-    to_trio: trio.MemorySendChannel,
-    coro: AsyncIterator,
-) -> None:
-    """Stream async generator results back to ``trio``.
-
-    ``from_trio`` might eventually be used here for
-    bidirectional streaming.
-    """
-    async for item in coro:
-        to_trio.send_nowait(item)
-
-
 def _run_asyncio_task(
     func: Callable,
     *,
     qsize: int = 1,
-    _treat_as_stream: bool = False,
+    provide_channels: bool = False,
     **kwargs,
+
 ) -> Any:
-    """Run an ``asyncio`` async function or generator in a task, return
+    '''
+    Run an ``asyncio`` async function or generator in a task, return
     or stream the result back to ``trio``.
 
-    """
+    '''
-    assert current_actor().is_infected_aio()
+    if not current_actor().is_infected_aio():
+        raise RuntimeError("`infect_asyncio` mode is not enabled!?")
 
     # ITC (inter task comms)
     from_trio = asyncio.Queue(qsize)  # type: ignore
@@ -68,9 +52,11 @@ def _run_asyncio_task(
         # the assumption is that the target async routine accepts the
         # send channel then it intends to yield more then one return
         # value otherwise it would just return ;P
-        # _treat_as_stream = True
         assert qsize > 1
 
+    if provide_channels:
+        assert 'to_trio' in args
+
     # allow target func to accept/stream results manually by name
     if 'to_trio' in args:
         kwargs['to_trio'] = to_trio
@@ -78,164 +64,145 @@ def _run_asyncio_task(
     if 'from_trio' in args:
         kwargs['from_trio'] = from_trio
 
-    # if 'from_aio' in args:
-    #     kwargs['from_aio'] = from_aio
-
     coro = func(**kwargs)
 
-    # cancel_scope = trio.CancelScope()
+    cancel_scope = trio.CancelScope()
+    aio_task_complete = trio.Event()
+    aio_err: Optional[BaseException] = None
+
+    async def wait_on_coro_final_result(
+        to_trio: trio.MemorySendChannel,
+        coro: Awaitable,
+        aio_task_complete: trio.Event,
+
+    ) -> None:
+        '''
+        Await ``coro`` and relay result back to ``trio``.
+
+        '''
+        nonlocal aio_err
+        orig = result = id(coro)
+        try:
+            result = await coro
+        except BaseException as err:
+            aio_err = err
+            from_aio._err = aio_err
+            raise
+        finally:
+            aio_task_complete.set()
+            if result != orig and aio_err is None:
+                to_trio.send_nowait(result)
 
     # start the asyncio task we submitted from trio
     if inspect.isawaitable(coro):
-        task = asyncio.create_task(run_coro(to_trio, coro))
+        task = asyncio.create_task(
-
+            wait_on_coro_final_result(to_trio, coro, aio_task_complete)
-    elif inspect.isasyncgen(coro):
+        )
-        task = asyncio.create_task(consume_asyncgen(to_trio, coro))
 
     else:
         raise TypeError(f"No support for invoking {coro}")
 
-    aio_err = None
+    def cancel_trio(task) -> None:
+        '''
+        Cancel the calling ``trio`` task on error.
 
-    def cancel_trio(task):
+        '''
-        """Cancel the calling ``trio`` task on error.
-        """
         nonlocal aio_err
-        aio_err = task.exception()
+        try:
+            aio_err = task.exception()
+        except asyncio.CancelledError as cerr:
+            aio_err = cerr
 
         if aio_err:
             log.exception(f"asyncio task errorred:\n{aio_err}")
-
+            from_aio._err = aio_err
-        # cancel_scope.cancel()
+            cancel_scope.cancel()
-        from_aio._err = aio_err
+            from_aio.close()
-        to_trio.close()
 
     task.add_done_callback(cancel_trio)
 
-    return task, from_aio, to_trio
+    return task, from_aio, to_trio, cancel_scope, aio_task_complete
 
 
 async def run_task(
     func: Callable,
     *,
+
     qsize: int = 2**10,
-    _treat_as_stream: bool = False,
     **kwargs,
+
 ) -> Any:
     """Run an ``asyncio`` async function or generator in a task, return
     or stream the result back to ``trio``.
 
     """
-    # assert current_actor().is_infected_aio()
-
-    # # ITC (inter task comms)
-    # from_trio = asyncio.Queue(qsize)  # type: ignore
-    # to_trio, from_aio = trio.open_memory_channel(qsize)  # type: ignore
-
-    # args = tuple(inspect.getfullargspec(func).args)
-
-    # if getattr(func, '_tractor_steam_function', None):
-    #     # the assumption is that the target async routine accepts the
-    #     # send channel then it intends to yield more then one return
-    #     # value otherwise it would just return ;P
-    #     _treat_as_stream = True
-
-    # # allow target func to accept/stream results manually by name
-    # if 'to_trio' in args:
-    #     kwargs['to_trio'] = to_trio
-    # if 'from_trio' in args:
-    #     kwargs['from_trio'] = from_trio
-
-    # coro = func(**kwargs)
-
-    # cancel_scope = trio.CancelScope()
-
-    # # start the asyncio task we submitted from trio
-    # if inspect.isawaitable(coro):
-    #     task = asyncio.create_task(run_coro(to_trio, coro))
-
-    # elif inspect.isasyncgen(coro):
-    #     task = asyncio.create_task(consume_asyncgen(to_trio, coro))
-
-    # else:
-    #     raise TypeError(f"No support for invoking {coro}")
-
-    # aio_err = None
-
-    # def cancel_trio(task):
-    #     """Cancel the calling ``trio`` task on error.
-    #     """
-    #     nonlocal aio_err
-    #     aio_err = task.exception()
-
-    #     if aio_err:
-    #         log.exception(f"asyncio task errorred:\n{aio_err}")
-
-    #     cancel_scope.cancel()
-
-    # task.add_done_callback(cancel_trio)
-
-    # async iterator
-    # if inspect.isasyncgen(coro) or _treat_as_stream:
-
-    # if inspect.isasyncgenfunction(meth) or :
-    if _treat_as_stream:
-
-        task, from_aio, to_trio = _run_asyncio_task(
-            func,
-            qsize=2**8,
-            **kwargs,
-        )
-
-        return from_aio
-
-        # async def stream_results():
-        #     try:
-        #         with cancel_scope:
-        #             # stream values upward
-        #             async with from_aio:
-        #                 async for item in from_aio:
-        #                     yield item
-
-        #         if cancel_scope.cancelled_caught:
-        #             # always raise from any captured asyncio error
-        #             if aio_err:
-        #                 raise aio_err
-
-        #     except BaseException as err:
-        #         if aio_err is not None:
-        #             # always raise from any captured asyncio error
-        #             raise err from aio_err
-        #         else:
-        #             raise
-        #     finally:
-        #         # breakpoint()
-        #         task.cancel()
-
-        # return stream_results()
-
     # simple async func
     try:
-        task, from_aio, to_trio = _run_asyncio_task(
+        task, from_aio, to_trio, cs, _ = _run_asyncio_task(
             func,
             qsize=1,
             **kwargs,
         )
 
-        # with cancel_scope:
+        # return single value
-        # async with from_aio:
+        with cs:
-            # return single value
+            # naively expect the mem chan api to do the job
-        return await from_aio.receive()
+            # of handling cross-framework cancellations / errors
+            return await from_aio.receive()
 
-        # if cancel_scope.cancelled_caught:
+        if cs.cancelled_caught:
-        #     # always raise from any captured asyncio error
+            # always raise from any captured asyncio error
-        #     if aio_err:
+            if from_aio._err:
-        #         raise aio_err
+                raise from_aio._err
 
     # Do we need this?
     except BaseException as err:
-        # await tractor.breakpoint()
+
         aio_err = from_aio._err
+
+        if aio_err is not None:
+            # always raise from any captured asyncio error
+            raise err from aio_err
+        else:
+            raise
+    finally:
+        if not task.done():
+            task.cancel()
+
+
+# TODO: explicitly api for the streaming case where
+# we pull from the mem chan in an async generator?
+# This ends up looking more like our ``Portal.open_stream_from()``
+# NB: code below is untested.
+
+
+@acm
+async def open_channel_from(
+    target: Callable[[Any, ...], Any],
+    **kwargs,
+
+) -> AsyncIterator[Any]:
+
+    try:
+        task, from_aio, to_trio, cs, aio_task_complete = _run_asyncio_task(
+            target,
+            qsize=2**8,
+            provide_channels=True,
+            **kwargs,
+        )
+
+        with cs:
+            # sync to "started()" call.
+            first = await from_aio.receive()
+            # stream values upward
+            async with from_aio:
+                yield first, from_aio
+                # await aio_task_complete.wait()
+
+    except BaseException as err:
+
+        aio_err = from_aio._err
+
         if aio_err is not None:
             # always raise from any captured asyncio error
             raise err from aio_err
@@ -243,17 +210,20 @@ async def run_task(
             raise
 
     finally:
-        task.cancel()
+        if cs.cancelled_caught:
+            # always raise from any captured asyncio error
+            if from_aio._err:
+                raise from_aio._err
 
-
+        if not task.done():
-# async def stream_from_task
+            task.cancel()
-#     pass
 
 
 def run_as_asyncio_guest(
     trio_main: Callable,
 ) -> None:
-    """Entry for an "infected ``asyncio`` actor".
+    '''
+    Entry for an "infected ``asyncio`` actor".
 
     Uh, oh. :o
 
@@ -267,16 +237,22 @@ def run_as_asyncio_guest(
     it.
 
     :)
-    """
+
+    '''
+    # Disable sigint handling in children? (nawp)
+    # import signal
+    # signal.signal(signal.SIGINT, signal.SIG_IGN)
+
+    get_console_log('runtime')
+
     async def aio_main(trio_main):
 
         loop = asyncio.get_running_loop()
-
         trio_done_fut = asyncio.Future()
 
         def trio_done_callback(main_outcome):
 
-            log.info(f"trio_main finished: {main_outcome!r}")
+            print(f"trio_main finished: {main_outcome!r}")
             trio_done_fut.set_result(main_outcome)
 
         # start the infection: run trio on the asyncio loop in "guest mode"
@@ -287,7 +263,6 @@ def run_as_asyncio_guest(
             run_sync_soon_threadsafe=loop.call_soon_threadsafe,
             done_callback=trio_done_callback,
         )
-
         (await trio_done_fut).unwrap()
 
     # might as well if it's installed.