jc211
/
tractor
forked from goodboy/tractor
1
0
Fork 0
Code Pull Requests Activity

First draft: `.to_asyncio.open_channel_from()`

iaio_backup
Tyler Goodlet 2021-10-07 23:14:34 -04:00
parent b7d5ff79b4
commit a0b885e9d7
1 changed files with 113 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
tractor/to_asyncio.py
View File

@ -3,17 +3,19 @@ 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 from .log import get_logger, get_console_log
from ._state import current_actor from ._state import current_actor
log = get_logger(__name__) log = get_logger(__name__)
@ -22,36 +24,30 @@ log = get_logger(__name__)
__all__ = ['run_task', 'run_as_asyncio_guest'] __all__ = ['run_task', 'run_as_asyncio_guest']
async def run_coro( # async def consume_asyncgen(
to_trio: trio.MemorySendChannel, # to_trio: trio.MemorySendChannel,
coro: Awaitable, # coro: AsyncIterator,
) -> None: # ) -> None:
"""Await ``coro`` and relay result back to ``trio``. # """Stream async generator results back to ``trio``.
"""
to_trio.send_nowait(await coro)
# ``from_trio`` might eventually be used here for
async def consume_asyncgen( # bidirectional streaming.
to_trio: trio.MemorySendChannel, # """
coro: AsyncIterator, # async for item in coro:
) -> None: # to_trio.send_nowait(item)
"""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,
_treat_as_stream: bool = False, # _treat_as_stream: bool = False,
provide_channels: 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``.
""" """
@ -73,6 +69,9 @@ def _run_asyncio_task(
# _treat_as_stream = True # _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
@ -80,25 +79,46 @@ 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
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(run_coro(to_trio, coro)) task = asyncio.create_task(
wait_on_coro_final_result(to_trio, coro, aio_task_complete)
)
elif inspect.isasyncgen(coro): # elif inspect.isasyncgen(coro):
task = asyncio.create_task(consume_asyncgen(to_trio, 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}")
aio_err = None
def cancel_trio(task): def cancel_trio(task):
"""Cancel the calling ``trio`` task on error. """Cancel the calling ``trio`` task on error.
""" """
@ -110,23 +130,21 @@ def _run_asyncio_task(
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._err = aio_err from_aio.close()
from_aio.close()
task.add_done_callback(cancel_trio) task.add_done_callback(cancel_trio)
return task, from_aio, to_trio, cancel_scope return task, from_aio, to_trio, cancel_scope, aio_task_complete
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, # _treat_as_stream: bool = False,
**kwargs, **kwargs,
) -> Any: ) -> Any:
@ -134,22 +152,9 @@ async def run_task(
or stream the result back to ``trio``. or stream the result back to ``trio``.
""" """
# streaming ``asyncio`` task
if _treat_as_stream:
task, from_aio, to_trio, cs = _run_asyncio_task(
func,
qsize=2**8,
**kwargs,
)
# naively expect the mem chan api to do the job
# of handling cross-framework cancellations / errors
return from_aio
# simple async func # simple async func
try: try:
task, from_aio, to_trio, cs = _run_asyncio_task( task, from_aio, to_trio, cs, _ = _run_asyncio_task(
func, func,
qsize=1, qsize=1,
**kwargs, **kwargs,
@ -157,6 +162,8 @@ async def run_task(
# return single value # return single value
with cs: with cs:
# naively expect the mem chan api to do the job
# of handling cross-framework cancellations / errors
return await from_aio.receive() return await from_aio.receive()
if cs.cancelled_caught: if cs.cancelled_caught:
@ -165,7 +172,7 @@ async def run_task(
raise from_aio._err raise from_aio._err
# Do we need this? # Do we need this?
except Exception as err: except BaseException as err:
aio_err = from_aio._err aio_err = from_aio._err
@ -178,8 +185,8 @@ async def run_task(
# except trio.Cancelled: # except trio.Cancelled:
# raise # raise
finally: finally:
# if not task.done(): if not task.done():
task.cancel() task.cancel()
# TODO: explicit api for the streaming case where # TODO: explicit api for the streaming case where
@ -187,21 +194,54 @@ async def run_task(
# This ends up looking more like our ``Portal.open_stream_from()`` # This ends up looking more like our ``Portal.open_stream_from()``
# NB: code below is untested. # NB: code below is untested.
# @asynccontextmanager # async def _start_and_sync_aio_task(
# async def stream_from_task( # from_trio,
# to_trio,
# from_aio,
# target: Callable[Any],
# **kwargs,
# ) -> AsyncIterator[Any]: @acm
async def open_channel_from(
# from_aoi = await run_task(target, _treat_as_stream=True, **kwargs) target: Callable[[Any, ...], Any],
**kwargs,
# with cancel_scope: ) -> AsyncIterator[Any]:
# # stream values upward
# async with from_aio: try:
# async for item in from_aio: task, from_aio, to_trio, cs, aio_task_complete = _run_asyncio_task(
# yield item 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
else:
raise
finally:
if cs.cancelled_caught:
# always raise from any captured asyncio error
if from_aio._err:
raise from_aio._err
if not task.done():
task.cancel()
def run_as_asyncio_guest( def run_as_asyncio_guest(
@ -221,16 +261,22 @@ def run_as_asyncio_guest(
it. it.
:) :)
""" """
# Disable sigint handling in children?
# 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):
log.info(f"trio_main finished: {main_outcome!r}") print(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"
@ -241,7 +287,6 @@ 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.