Solve our abandonment issues..

To make the recent set of tests pass this (hopefully) finally solves all
`asyncio` embedded `trio` guest-run abandonment by ensuring we "pump the
event loop" until the guest-run future is fully complete.

Accomplished via simple poll loop of the form `while not
trio_done_fut.done(): await asyncio.sleep(.1)` in the `aio_main()`
task's exception teardown sequence. The loop does a naive 10ms
"pump-via-sleep & poll" for the `trio` side to complete before finally
exiting (and presumably raising) from the SIGINT cancellation.

Other related cleanups and refinements:
- use `asyncio.Task.result()` inside `cancel_trio()` since it also
  inline-raises any exception outcome and we can also log-report the
  result in non-error cases.
- comment out buncha not-sure-we-need-it stuff in `cancel_trio()`.
- remove the botched `AsyncioCancelled(CancelledError):` idea obvi XD
- comment `greenback` init for now in `aio_main()` since (pretty sure)
  we don't ever want to actually REPL in that specific func-as-task?
- always capture any `fute_err: BaseException` from the `main_outcome:
  Outcome` delivered by the `trio` side guest-run task.
- add and raise a new super noisy `AsyncioRuntimeTranslationError`
  whenever we detect that the guest-run `trio_done_fut` has not
  completed before task exit; should avoid abandonment issues ever
  happening again without knowing!
aio_abandons
Tyler Goodlet 2024-06-26 13:48:36 -04:00
parent 268bd0d8ec
commit 9133f42b07
1 changed files with 176 additions and 70 deletions

View File

@ -33,11 +33,12 @@ from typing import (
) )
import tractor import tractor
from tractor._exceptions import AsyncioCancelled
from tractor._state import ( from tractor._state import (
debug_mode, debug_mode,
) )
from tractor.log import get_logger
from tractor.devx import _debug from tractor.devx import _debug
from tractor.log import get_logger
from tractor.trionics._broadcast import ( from tractor.trionics._broadcast import (
broadcast_receiver, broadcast_receiver,
BroadcastReceiver, BroadcastReceiver,
@ -51,7 +52,10 @@ from outcome import (
log = get_logger(__name__) log = get_logger(__name__)
__all__ = ['run_task', 'run_as_asyncio_guest'] __all__ = [
'run_task',
'run_as_asyncio_guest',
]
@dataclass @dataclass
@ -155,15 +159,16 @@ def _run_asyncio_task(
*, *,
qsize: int = 1, qsize: int = 1,
provide_channels: bool = False, provide_channels: bool = False,
hide_tb: bool = False,
**kwargs, **kwargs,
) -> LinkedTaskChannel: ) -> LinkedTaskChannel:
''' '''
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 the caller `trio.lowleve.Task`.
''' '''
__tracebackhide__ = True __tracebackhide__: bool = hide_tb
if not tractor.current_actor().is_infected_aio(): if not tractor.current_actor().is_infected_aio():
raise RuntimeError( raise RuntimeError(
"`infect_asyncio` mode is not enabled!?" "`infect_asyncio` mode is not enabled!?"
@ -224,6 +229,7 @@ def _run_asyncio_task(
try: try:
result = await coro result = await coro
except BaseException as aio_err: except BaseException as aio_err:
chan._aio_err = aio_err
if isinstance(aio_err, CancelledError): if isinstance(aio_err, CancelledError):
log.runtime( log.runtime(
'`asyncio` task was cancelled..\n' '`asyncio` task was cancelled..\n'
@ -232,7 +238,6 @@ def _run_asyncio_task(
log.exception( log.exception(
'`asyncio` task errored\n' '`asyncio` task errored\n'
) )
chan._aio_err = aio_err
raise raise
else: else:
@ -268,7 +273,7 @@ def _run_asyncio_task(
aio_task_complete aio_task_complete
) )
) )
chan._aio_task = task chan._aio_task: asyncio.Task = task
# XXX TODO XXX get this actually workin.. XD # XXX TODO XXX get this actually workin.. XD
# maybe setup `greenback` for `asyncio`-side task REPLing # maybe setup `greenback` for `asyncio`-side task REPLing
@ -284,19 +289,19 @@ def _run_asyncio_task(
def cancel_trio(task: asyncio.Task) -> None: def cancel_trio(task: asyncio.Task) -> None:
''' '''
Cancel the calling ``trio`` task on error. Cancel the calling `trio` task on error.
''' '''
nonlocal chan nonlocal chan
aio_err = chan._aio_err aio_err: BaseException|None = chan._aio_err
task_err: BaseException|None = None task_err: BaseException|None = None
# only to avoid ``asyncio`` complaining about uncaptured # only to avoid `asyncio` complaining about uncaptured
# task exceptions # task exceptions
try: try:
task.exception() res: Any = task.result()
except BaseException as terr: except BaseException as terr:
task_err = terr task_err: BaseException = terr
msg: str = ( msg: str = (
'Infected `asyncio` task {etype_str}\n' 'Infected `asyncio` task {etype_str}\n'
@ -328,42 +333,49 @@ def _run_asyncio_task(
if task_err is None: if task_err is None:
assert aio_err assert aio_err
aio_err.with_traceback(aio_err.__traceback__) # wait, wut?
# log.error( # aio_err.with_traceback(aio_err.__traceback__)
# 'infected task errorred'
# )
# TODO: show that the cancellation originated # TODO: show when cancellation originated
# from the ``trio`` side? right? # from each side more pedantically?
# elif type(aio_err) is CancelledError: # elif (
# type(aio_err) is CancelledError
# and # trio was the cause?
# cancel_scope.cancel_called
# ):
# log.cancel( # log.cancel(
# 'infected task was cancelled' # 'infected task was cancelled by `trio`-side'
# ) # )
# raise aio_err from task_err
# if cancel_scope.cancelled: # XXX: if not already, alway cancel the scope
# raise aio_err from err # on a task error in case the trio task is blocking on
# a checkpoint.
# XXX: alway cancel the scope on error
# in case the trio task is blocking
# on a checkpoint.
cancel_scope.cancel() cancel_scope.cancel()
if (
task_err
and
aio_err is not task_err
):
raise aio_err from task_err
# raise any `asyncio` side error. # raise any `asyncio` side error.
raise aio_err raise aio_err
log.info(
'`trio` received final result from {task}\n'
f'|_{res}\n'
)
# TODO: do we need this?
# if task_err:
# cancel_scope.cancel()
# raise task_err
task.add_done_callback(cancel_trio) task.add_done_callback(cancel_trio)
return chan return chan
class AsyncioCancelled(CancelledError):
'''
Asyncio cancelled translation (non-base) error
for use with the ``to_asyncio`` module
to be raised in the ``trio`` side task
'''
@acm @acm
async def translate_aio_errors( async def translate_aio_errors(
@ -386,7 +398,9 @@ async def translate_aio_errors(
) -> None: ) -> None:
aio_err = chan._aio_err aio_err = chan._aio_err
if ( if (
aio_err is not None and aio_err is not None
and
# not isinstance(aio_err, CancelledError)
type(aio_err) != CancelledError type(aio_err) != CancelledError
): ):
# always raise from any captured asyncio error # always raise from any captured asyncio error
@ -418,13 +432,17 @@ async def translate_aio_errors(
): ):
aio_err = chan._aio_err aio_err = chan._aio_err
if ( if (
task.cancelled() and task.cancelled()
and
type(aio_err) is CancelledError type(aio_err) is CancelledError
): ):
# if an underlying ``asyncio.CancelledError`` triggered this # if an underlying `asyncio.CancelledError` triggered this
# channel close, raise our (non-``BaseException``) wrapper # channel close, raise our (non-``BaseException``) wrapper
# error: ``AsyncioCancelled`` from that source error. # error: ``AsyncioCancelled`` from that source error.
raise AsyncioCancelled from aio_err raise AsyncioCancelled(
f'Task cancelled\n'
f'|_{task}\n'
) from aio_err
else: else:
raise raise
@ -467,8 +485,8 @@ async def run_task(
) -> 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`.
''' '''
# simple async func # simple async func
@ -526,10 +544,27 @@ async def open_channel_from(
chan._to_trio.close() chan._to_trio.close()
class AsyncioRuntimeTranslationError(RuntimeError):
'''
We failed to correctly relay runtime semantics and/or maintain SC
supervision rules cross-event-loop.
'''
def run_as_asyncio_guest( def run_as_asyncio_guest(
trio_main: Callable, trio_main: Callable,
# ^-NOTE-^ when spawned with `infected_aio=True` this func is
# normally `Actor._async_main()` as is passed by some boostrap
# entrypoint like `._entry._trio_main()`.
) -> None: ) -> None:
# ^-TODO-^ technically whatever `trio_main` returns.. we should
# try to use func-typevar-params at leaast by 3.13!
# -[ ] https://typing.readthedocs.io/en/latest/spec/callables.html#callback-protocols
# -[ ] https://peps.python.org/pep-0646/#using-type-variable-tuples-in-functions
# -[ ] https://typing.readthedocs.io/en/latest/spec/callables.html#unpack-for-keyword-arguments
# -[ ] https://peps.python.org/pep-0718/
''' '''
Entry for an "infected ``asyncio`` actor". Entry for an "infected ``asyncio`` actor".
@ -555,7 +590,13 @@ def run_as_asyncio_guest(
# :) # :)
async def aio_main(trio_main): async def aio_main(trio_main):
'''
Main `asyncio.Task` which calls
`trio.lowlevel.start_guest_run()` to "infect" the `asyncio`
event-loop by embedding the `trio` scheduler allowing us to
boot the `tractor` runtime and connect back to our parent.
'''
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
trio_done_fut = asyncio.Future() trio_done_fut = asyncio.Future()
startup_msg: str = ( startup_msg: str = (
@ -564,17 +605,22 @@ def run_as_asyncio_guest(
'-> built a `trio`-done future\n' '-> built a `trio`-done future\n'
) )
if debug_mode(): # TODO: shoudn't this be done in the guest-run trio task?
# XXX make it obvi we know this isn't supported yet! # if debug_mode():
log.error( # # XXX make it obvi we know this isn't supported yet!
'Attempting to enter unsupported `greenback` init ' # log.error(
'from `asyncio` task..' # 'Attempting to enter unsupported `greenback` init '
) # 'from `asyncio` task..'
await _debug.maybe_init_greenback( # )
force_reload=True, # await _debug.maybe_init_greenback(
) # force_reload=True,
# )
def trio_done_callback(main_outcome): def trio_done_callback(main_outcome):
log.info(
f'trio_main finished with\n'
f'|_{main_outcome!r}'
)
if isinstance(main_outcome, Error): if isinstance(main_outcome, Error):
error: BaseException = main_outcome.error error: BaseException = main_outcome.error
@ -594,7 +640,6 @@ def run_as_asyncio_guest(
else: else:
trio_done_fut.set_result(main_outcome) trio_done_fut.set_result(main_outcome)
log.runtime(f'trio_main finished: {main_outcome!r}')
startup_msg += ( startup_msg += (
f'-> created {trio_done_callback!r}\n' f'-> created {trio_done_callback!r}\n'
@ -613,26 +658,48 @@ 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,
) )
fute_err: BaseException|None = None
try: try:
# TODO: better SIGINT handling since shielding seems to
# make NO DIFFERENCE XD
# -[ ] maybe this is due to 3.11's recent SIGINT handling
# changes and we can better work with/around it?
# https://docs.python.org/3/library/asyncio-runner.html#handling-keyboard-interruption
out: Outcome = await asyncio.shield(trio_done_fut) out: Outcome = await asyncio.shield(trio_done_fut)
# NOTE `Error.unwrap()` will raise
# NOTE will raise (via `Error.unwrap()`) from any
# exception packed into the guest-run's `main_outcome`.
return out.unwrap() return out.unwrap()
except asyncio.CancelledError: except (
# XXX special SIGINT-handling is required since
# `asyncio.shield()`-ing seems to NOT handle that case as
# per recent changes in 3.11:
# https://docs.python.org/3/library/asyncio-runner.html#handling-keyboard-interruption
#
# NOTE: further, apparently ONLY need to handle this
# special SIGINT case since all other `asyncio`-side
# errors can be processed via our `chan._aio_err`
# relaying (right?); SIGINT seems to be totally diff
# error path in `asyncio`'s runtime..?
asyncio.CancelledError,
) as fute_err:
err_message: str = (
'main `asyncio` task '
)
if isinstance(fute_err, asyncio.CancelledError):
err_message += 'was cancelled!\n'
else:
err_message += f'errored with {out.error!r}\n'
actor: tractor.Actor = tractor.current_actor() actor: tractor.Actor = tractor.current_actor()
log.exception( log.exception(
'`asyncio`-side main task was cancelled!\n' err_message
'Cancelling actor-runtime..\n' +
'Cancelling `trio`-side `tractor`-runtime..\n'
f'c)>\n' f'c)>\n'
f' |_{actor}.cancel_soon()\n' f' |_{actor}.cancel_soon()\n'
) )
# TODO: reduce this comment bloc since abandon issues are
# now solved?
#
# XXX NOTE XXX the next LOC is super important!!! # XXX NOTE XXX the next LOC is super important!!!
# => without it, we can get a guest-run abandonment case # => without it, we can get a guest-run abandonment case
# where asyncio will not trigger `trio` in a final event # where asyncio will not trigger `trio` in a final event
@ -681,16 +748,55 @@ def run_as_asyncio_guest(
# is apparently a working fix! # is apparently a working fix!
actor.cancel_soon() actor.cancel_soon()
# XXX NOTE XXX PUMP the asyncio event loop to allow `trio`-side to # XXX NOTE XXX pump the `asyncio` event-loop to allow
# `trio`-guest-run to complete and teardown !! # `trio`-side to `trio`-guest-run to complete and
# teardown !!
# #
# XXX WITHOUT THIS the guest-run gets race-conditionally # *WITHOUT THIS* the guest-run can get race-conditionally abandoned!!
# abandoned by `asyncio`!! # XD
# XD XD XD #
await asyncio.shield( await asyncio.sleep(.1) # `delay` can't be 0 either XD
asyncio.sleep(.1) # NOPE! it can't be 0 either XD while not trio_done_fut.done():
) log.runtime(
raise 'Waiting on main guest-run `asyncio` task to complete..\n'
f'|_trio_done_fut: {trio_done_fut}\n'
)
await asyncio.sleep(.1)
# XXX: don't actually need the shield.. seems to
# make no difference (??) and we know it spawns an
# internal task..
# await asyncio.shield(asyncio.sleep(.1))
# XXX alt approach but can block indefinitely..
# so don't use?
# loop._run_once()
try:
return trio_done_fut.result()
except asyncio.exceptions.InvalidStateError as state_err:
# XXX be super dupere noisy about abandonment issues!
aio_task: asyncio.Task = asyncio.current_task()
message: str = (
'The `asyncio`-side task likely exited before the '
'`trio`-side guest-run completed!\n\n'
)
if fute_err:
message += (
f'The main {aio_task}\n'
f'STOPPED due to {type(fute_err)}\n\n'
)
message += (
f'Likely something inside our guest-run-as-task impl is '
f'not effectively waiting on the `trio`-side to complete ?!\n'
f'This code -> {aio_main!r}\n\n'
'Below you will likely see a '
'"RuntimeWarning: Trio guest run got abandoned.." !!\n'
)
raise AsyncioRuntimeTranslationError(message) from state_err
# might as well if it's installed. # might as well if it's installed.
try: try:
@ -698,7 +804,7 @@ def run_as_asyncio_guest(
loop = uvloop.new_event_loop() loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
except ImportError: except ImportError:
pass log.runtime('`uvloop` not available..')
return asyncio.run( return asyncio.run(
aio_main(trio_main), aio_main(trio_main),