Compare commits
10 Commits
8805cbee3d
...
d04a754655
Author | SHA1 | Date |
---|---|---|
Tyler Goodlet | d04a754655 | |
Tyler Goodlet | 6e9562ffd8 | |
Tyler Goodlet | 9c8a120ccc | |
Tyler Goodlet | a0b885e9d7 | |
Tyler Goodlet | b7d5ff79b4 | |
Tyler Goodlet | 60b892cc80 | |
Tyler Goodlet | cbc18f94ec | |
Tyler Goodlet | e016884ce0 | |
Tyler Goodlet | d42f628597 | |
Tyler Goodlet | a572fde6c6 |
|
@ -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()
|
||||
|
||||
|
|
|
@ -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
|
||||
|
||||
|
|
|
@ -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))
|
||||
|
||||
elif inspect.isasyncgen(coro):
|
||||
task = asyncio.create_task(consume_asyncgen(to_trio, coro))
|
||||
task = asyncio.create_task(
|
||||
wait_on_coro_final_result(to_trio, coro, aio_task_complete)
|
||||
)
|
||||
|
||||
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}")
|
||||
|
||||
# cancel_scope.cancel()
|
||||
from_aio._err = aio_err
|
||||
to_trio.close()
|
||||
from_aio._err = aio_err
|
||||
cancel_scope.cancel()
|
||||
from_aio.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:
|
||||
# async with from_aio:
|
||||
# return single value
|
||||
return await from_aio.receive()
|
||||
# return single value
|
||||
with cs:
|
||||
# naively expect the mem chan api to do the job
|
||||
# of handling cross-framework cancellations / errors
|
||||
return await from_aio.receive()
|
||||
|
||||
# if cancel_scope.cancelled_caught:
|
||||
# # always raise from any captured asyncio error
|
||||
# if aio_err:
|
||||
# raise aio_err
|
||||
if cs.cancelled_caught:
|
||||
# always raise from any captured asyncio error
|
||||
if from_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
|
||||
|
||||
|
||||
# async def stream_from_task
|
||||
# pass
|
||||
if not task.done():
|
||||
task.cancel()
|
||||
|
||||
|
||||
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.
|
||||
|
|
Loading…
Reference in New Issue