Compare commits
No commits in common. "d04a754655fb8e63ccb658e9a8c811a33c9bce59" and "8805cbee3d5c165028c8c461a089212100845bd2" have entirely different histories.
d04a754655
...
8805cbee3d
|
@ -2,7 +2,6 @@
|
||||||
Multi-core debugging for da peeps!
|
Multi-core debugging for da peeps!
|
||||||
|
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
|
||||||
import bdb
|
import bdb
|
||||||
import sys
|
import sys
|
||||||
from functools import partial
|
from functools import partial
|
||||||
|
@ -28,14 +27,12 @@ from ._exceptions import is_multi_cancelled
|
||||||
try:
|
try:
|
||||||
# wtf: only exported when installed in dev mode?
|
# wtf: only exported when installed in dev mode?
|
||||||
import pdbpp
|
import pdbpp
|
||||||
|
|
||||||
except ImportError:
|
except ImportError:
|
||||||
# pdbpp is installed in regular mode...it monkey patches stuff
|
# pdbpp is installed in regular mode...it monkey patches stuff
|
||||||
import pdb
|
import pdb
|
||||||
assert pdb.xpm, "pdbpp is not installed?" # type: ignore
|
assert pdb.xpm, "pdbpp is not installed?" # type: ignore
|
||||||
pdbpp = pdb
|
pdbpp = pdb
|
||||||
|
|
||||||
|
|
||||||
log = get_logger(__name__)
|
log = get_logger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
@ -95,23 +92,6 @@ class PdbwTeardown(pdbpp.Pdb):
|
||||||
_pdb_release_hook()
|
_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.
|
# TODO: will be needed whenever we get to true remote debugging.
|
||||||
# XXX see https://github.com/goodboy/tractor/issues/130
|
# XXX see https://github.com/goodboy/tractor/issues/130
|
||||||
|
|
||||||
|
@ -481,6 +461,21 @@ async def _breakpoint(
|
||||||
debug_func(actor)
|
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):
|
def _set_trace(actor=None):
|
||||||
pdb = _mk_pdb()
|
pdb = _mk_pdb()
|
||||||
|
|
||||||
|
|
|
@ -192,7 +192,6 @@ async def new_proc(
|
||||||
_runtime_vars: Dict[str, Any], # serialized and sent to _child
|
_runtime_vars: Dict[str, Any], # serialized and sent to _child
|
||||||
|
|
||||||
*,
|
*,
|
||||||
|
|
||||||
infect_asyncio: bool = False,
|
infect_asyncio: bool = False,
|
||||||
task_status: TaskStatus[Portal] = trio.TASK_STATUS_IGNORED
|
task_status: TaskStatus[Portal] = trio.TASK_STATUS_IGNORED
|
||||||
|
|
||||||
|
|
|
@ -1,21 +1,18 @@
|
||||||
'''
|
"""
|
||||||
Infection apis for ``asyncio`` loops running ``trio`` using guest mode.
|
Infection apis for ``asyncio`` loops running ``trio`` using guest mode.
|
||||||
|
"""
|
||||||
'''
|
|
||||||
import asyncio
|
import asyncio
|
||||||
from contextlib import asynccontextmanager as acm
|
|
||||||
import inspect
|
import inspect
|
||||||
from typing import (
|
from typing import (
|
||||||
Any,
|
Any,
|
||||||
Callable,
|
Callable,
|
||||||
AsyncIterator,
|
AsyncIterator,
|
||||||
Awaitable,
|
Awaitable,
|
||||||
Optional,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
import trio
|
import trio
|
||||||
|
|
||||||
from .log import get_logger, get_console_log
|
from .log import get_logger
|
||||||
from ._state import current_actor
|
from ._state import current_actor
|
||||||
|
|
||||||
log = get_logger(__name__)
|
log = get_logger(__name__)
|
||||||
|
@ -24,21 +21,40 @@ log = get_logger(__name__)
|
||||||
__all__ = ['run_task', 'run_as_asyncio_guest']
|
__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(
|
def _run_asyncio_task(
|
||||||
func: Callable,
|
func: Callable,
|
||||||
*,
|
*,
|
||||||
qsize: int = 1,
|
qsize: int = 1,
|
||||||
provide_channels: bool = False,
|
_treat_as_stream: bool = False,
|
||||||
**kwargs,
|
**kwargs,
|
||||||
|
|
||||||
) -> Any:
|
) -> 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``.
|
or stream the result back to ``trio``.
|
||||||
|
|
||||||
'''
|
"""
|
||||||
if not current_actor().is_infected_aio():
|
assert current_actor().is_infected_aio()
|
||||||
raise RuntimeError("`infect_asyncio` mode is not enabled!?")
|
|
||||||
|
|
||||||
# ITC (inter task comms)
|
# ITC (inter task comms)
|
||||||
from_trio = asyncio.Queue(qsize) # type: ignore
|
from_trio = asyncio.Queue(qsize) # type: ignore
|
||||||
|
@ -52,11 +68,9 @@ def _run_asyncio_task(
|
||||||
# the assumption is that the target async routine accepts the
|
# the assumption is that the target async routine accepts the
|
||||||
# send channel then it intends to yield more then one return
|
# send channel then it intends to yield more then one return
|
||||||
# value otherwise it would just return ;P
|
# value otherwise it would just return ;P
|
||||||
|
# _treat_as_stream = True
|
||||||
assert qsize > 1
|
assert qsize > 1
|
||||||
|
|
||||||
if provide_channels:
|
|
||||||
assert 'to_trio' in args
|
|
||||||
|
|
||||||
# allow target func to accept/stream results manually by name
|
# allow target func to accept/stream results manually by name
|
||||||
if 'to_trio' in args:
|
if 'to_trio' in args:
|
||||||
kwargs['to_trio'] = to_trio
|
kwargs['to_trio'] = to_trio
|
||||||
|
@ -64,145 +78,164 @@ def _run_asyncio_task(
|
||||||
if 'from_trio' in args:
|
if 'from_trio' in args:
|
||||||
kwargs['from_trio'] = from_trio
|
kwargs['from_trio'] = from_trio
|
||||||
|
|
||||||
|
# if 'from_aio' in args:
|
||||||
|
# kwargs['from_aio'] = from_aio
|
||||||
|
|
||||||
coro = func(**kwargs)
|
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
|
# start the asyncio task we submitted from trio
|
||||||
if inspect.isawaitable(coro):
|
if inspect.isawaitable(coro):
|
||||||
task = asyncio.create_task(
|
task = asyncio.create_task(run_coro(to_trio, coro))
|
||||||
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:
|
else:
|
||||||
raise TypeError(f"No support for invoking {coro}")
|
raise TypeError(f"No support for invoking {coro}")
|
||||||
|
|
||||||
def cancel_trio(task) -> None:
|
aio_err = None
|
||||||
'''
|
|
||||||
Cancel the calling ``trio`` task on error.
|
|
||||||
|
|
||||||
'''
|
def cancel_trio(task):
|
||||||
|
"""Cancel the calling ``trio`` task on error.
|
||||||
|
"""
|
||||||
nonlocal aio_err
|
nonlocal aio_err
|
||||||
try:
|
aio_err = task.exception()
|
||||||
aio_err = task.exception()
|
|
||||||
except asyncio.CancelledError as cerr:
|
|
||||||
aio_err = cerr
|
|
||||||
|
|
||||||
if aio_err:
|
if aio_err:
|
||||||
log.exception(f"asyncio task errorred:\n{aio_err}")
|
log.exception(f"asyncio task errorred:\n{aio_err}")
|
||||||
from_aio._err = aio_err
|
|
||||||
cancel_scope.cancel()
|
# cancel_scope.cancel()
|
||||||
from_aio.close()
|
from_aio._err = aio_err
|
||||||
|
to_trio.close()
|
||||||
|
|
||||||
task.add_done_callback(cancel_trio)
|
task.add_done_callback(cancel_trio)
|
||||||
|
|
||||||
return task, from_aio, to_trio, cancel_scope, aio_task_complete
|
return task, from_aio, to_trio
|
||||||
|
|
||||||
|
|
||||||
async def run_task(
|
async def run_task(
|
||||||
func: Callable,
|
func: Callable,
|
||||||
*,
|
*,
|
||||||
|
|
||||||
qsize: int = 2**10,
|
qsize: int = 2**10,
|
||||||
|
_treat_as_stream: bool = False,
|
||||||
**kwargs,
|
**kwargs,
|
||||||
|
|
||||||
) -> Any:
|
) -> 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``.
|
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
|
# simple async func
|
||||||
try:
|
try:
|
||||||
task, from_aio, to_trio, cs, _ = _run_asyncio_task(
|
task, from_aio, to_trio = _run_asyncio_task(
|
||||||
func,
|
func,
|
||||||
qsize=1,
|
qsize=1,
|
||||||
**kwargs,
|
**kwargs,
|
||||||
)
|
)
|
||||||
|
|
||||||
# return single value
|
# with cancel_scope:
|
||||||
with cs:
|
# async with from_aio:
|
||||||
# naively expect the mem chan api to do the job
|
# return single value
|
||||||
# of handling cross-framework cancellations / errors
|
return await from_aio.receive()
|
||||||
return await from_aio.receive()
|
|
||||||
|
|
||||||
if cs.cancelled_caught:
|
# if cancel_scope.cancelled_caught:
|
||||||
# always raise from any captured asyncio error
|
# # always raise from any captured asyncio error
|
||||||
if from_aio._err:
|
# if aio_err:
|
||||||
raise from_aio._err
|
# raise aio_err
|
||||||
|
|
||||||
# Do we need this?
|
# Do we need this?
|
||||||
except BaseException as err:
|
except BaseException as err:
|
||||||
|
# await tractor.breakpoint()
|
||||||
aio_err = from_aio._err
|
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:
|
if aio_err is not None:
|
||||||
# always raise from any captured asyncio error
|
# always raise from any captured asyncio error
|
||||||
raise err from aio_err
|
raise err from aio_err
|
||||||
|
@ -210,20 +243,17 @@ async def open_channel_from(
|
||||||
raise
|
raise
|
||||||
|
|
||||||
finally:
|
finally:
|
||||||
if cs.cancelled_caught:
|
task.cancel()
|
||||||
# always raise from any captured asyncio error
|
|
||||||
if from_aio._err:
|
|
||||||
raise from_aio._err
|
|
||||||
|
|
||||||
if not task.done():
|
|
||||||
task.cancel()
|
# async def stream_from_task
|
||||||
|
# pass
|
||||||
|
|
||||||
|
|
||||||
def run_as_asyncio_guest(
|
def run_as_asyncio_guest(
|
||||||
trio_main: Callable,
|
trio_main: Callable,
|
||||||
) -> None:
|
) -> None:
|
||||||
'''
|
"""Entry for an "infected ``asyncio`` actor".
|
||||||
Entry for an "infected ``asyncio`` actor".
|
|
||||||
|
|
||||||
Uh, oh. :o
|
Uh, oh. :o
|
||||||
|
|
||||||
|
@ -237,22 +267,16 @@ def run_as_asyncio_guest(
|
||||||
it.
|
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):
|
async def aio_main(trio_main):
|
||||||
|
|
||||||
loop = asyncio.get_running_loop()
|
loop = asyncio.get_running_loop()
|
||||||
|
|
||||||
trio_done_fut = asyncio.Future()
|
trio_done_fut = asyncio.Future()
|
||||||
|
|
||||||
def trio_done_callback(main_outcome):
|
def trio_done_callback(main_outcome):
|
||||||
|
|
||||||
print(f"trio_main finished: {main_outcome!r}")
|
log.info(f"trio_main finished: {main_outcome!r}")
|
||||||
trio_done_fut.set_result(main_outcome)
|
trio_done_fut.set_result(main_outcome)
|
||||||
|
|
||||||
# start the infection: run trio on the asyncio loop in "guest mode"
|
# start the infection: run trio on the asyncio loop in "guest mode"
|
||||||
|
@ -263,6 +287,7 @@ def run_as_asyncio_guest(
|
||||||
run_sync_soon_threadsafe=loop.call_soon_threadsafe,
|
run_sync_soon_threadsafe=loop.call_soon_threadsafe,
|
||||||
done_callback=trio_done_callback,
|
done_callback=trio_done_callback,
|
||||||
)
|
)
|
||||||
|
|
||||||
(await trio_done_fut).unwrap()
|
(await trio_done_fut).unwrap()
|
||||||
|
|
||||||
# might as well if it's installed.
|
# might as well if it's installed.
|
||||||
|
|
Loading…
Reference in New Issue