Compare commits

..

No commits in common. "f4cab2ddb3f88d28657f9b8617b7e2b348aa7efe" and "89551ef37180383543daf495b22abb77cf5ce07e" have entirely different histories.

2 changed files with 36 additions and 91 deletions

View File

@ -326,19 +326,16 @@ async def spawn_and_error(breadth, depth) -> None:
) )
kwargs = { kwargs = {
'name': f'{name}_errorer_{i}', 'name': f'{name}_errorer_{i}',
# 'delay': 1,
} }
await nursery.run_in_actor(*args, **kwargs) await nursery.run_in_actor(*args, **kwargs)
@tractor_test @tractor_test
async def test_nested_multierrors(loglevel, start_method): async def test_nested_multierrors(loglevel, start_method):
''' """Test that failed actor sets are wrapped in `trio.MultiError`s.
Test that failed actor sets are wrapped in `trio.MultiError`s.
This test goes only 2 nurseries deep but we should eventually have tests This test goes only 2 nurseries deep but we should eventually have tests
for arbitrary n-depth actor trees. for arbitrary n-depth actor trees.
"""
'''
if start_method == 'trio': if start_method == 'trio':
depth = 3 depth = 3
subactor_breadth = 2 subactor_breadth = 2
@ -362,7 +359,6 @@ async def test_nested_multierrors(loglevel, start_method):
breadth=subactor_breadth, breadth=subactor_breadth,
depth=depth, depth=depth,
) )
except trio.MultiError as err: except trio.MultiError as err:
assert len(err.exceptions) == subactor_breadth assert len(err.exceptions) == subactor_breadth
for subexc in err.exceptions: for subexc in err.exceptions:
@ -398,13 +394,16 @@ async def test_nested_multierrors(loglevel, start_method):
assert isinstance(subexc, tractor.RemoteActorError) assert isinstance(subexc, tractor.RemoteActorError)
if depth > 0 and subactor_breadth > 1: if depth > 0 and subactor_breadth > 1:
# XXX it's race whether or not a parent containing # XXX not sure what's up with this..
# a nursery *may* get multiple child failures before # on windows sometimes spawning is just too slow and
# it cancels and tears down. # we get back the (sent) cancel signal instead
if isinstance(subexc, tractor.RemoteActorError): if platform.system() == 'Windows':
assert subexc.type in (trio.MultiError, tractor.RemoteActorError) if isinstance(subexc, tractor.RemoteActorError):
assert subexc.type in (trio.MultiError, tractor.RemoteActorError)
else:
assert isinstance(subexc, trio.MultiError)
else: else:
assert isinstance(subexc, trio.MultiError) assert subexc.type is trio.MultiError
else: else:
assert subexc.type in (tractor.RemoteActorError, trio.Cancelled) assert subexc.type in (tractor.RemoteActorError, trio.Cancelled)
@ -487,11 +486,9 @@ def test_cancel_while_childs_child_in_sync_sleep(
start_method, start_method,
spawn_backend, spawn_backend,
): ):
""" """Verify that a child cancelled while executing sync code is torn
Verify that a child cancelled while executing sync code is torn down down even when that cancellation is triggered by the parent
even when that cancellation is triggered by the parent
2 nurseries "up". 2 nurseries "up".
""" """
if start_method == 'forkserver': if start_method == 'forkserver':
pytest.skip("Forksever sux hard at resuming from sync sleep...") pytest.skip("Forksever sux hard at resuming from sync sleep...")
@ -503,7 +500,7 @@ def test_cancel_while_childs_child_in_sync_sleep(
spawn, spawn,
name='spawn', name='spawn',
) )
await trio.sleep(0.5) await trio.sleep(1)
assert 0 assert 0
with pytest.raises(AssertionError): with pytest.raises(AssertionError):

View File

@ -57,7 +57,7 @@ from .log import get_logger
from ._portal import Portal from ._portal import Portal
from ._actor import Actor from ._actor import Actor
from ._entry import _mp_main from ._entry import _mp_main
from ._exceptions import ActorFailure, NoResult from ._exceptions import ActorFailure
log = get_logger('tractor') log = get_logger('tractor')
@ -136,74 +136,47 @@ async def exhaust_portal(
# always be established and shutdown using a context manager api # always be established and shutdown using a context manager api
final = await portal.result() final = await portal.result()
except ( except (Exception, trio.MultiError) as err:
Exception,
trio.MultiError
) as err:
# we reraise in the parent task via a ``trio.MultiError`` # we reraise in the parent task via a ``trio.MultiError``
return err return err
except trio.Cancelled as err: except trio.Cancelled as err:
# lol, of course we need this too ;P # lol, of course we need this too ;P
# TODO: merge with above? # TODO: merge with above?
log.warning(f"Cancelled result waiter for {portal.actor.uid}") log.warning(f"Cancelled result waiter for {portal.actor.uid}")
return err return err
else: else:
log.debug(f"Returning final result: {final}") log.debug(f"Returning final result: {final}")
return final return final
async def pack_and_report_errors(
portal: Portal,
subactor: Actor,
errors: Dict[Tuple[str, str], Exception],
) -> Any:
# if this call errors we store the exception for later
# in ``errors`` which will be reraised inside
# a MultiError and we still send out a cancel request
result = await exhaust_portal(portal, subactor)
uid = portal.channel.uid
if (
isinstance(result, Exception)
# or isinstance(result, trio.MultiError)
):
errors[subactor.uid] = result
log.warning(f"{uid} received remote error:\n{result}")
raise result
elif isinstance(result, trio.Cancelled):
errors[subactor.uid] = result
log.runtime(f"{uid} was cancelled before result")
else:
log.runtime( f"{uid} received final result:\n{result}")
return result
async def cancel_on_completion( async def cancel_on_completion(
portal: Portal, portal: Portal,
subactor: Actor, actor: Actor,
errors: Dict[Tuple[str, str], Exception], errors: Dict[Tuple[str, str], Exception],
) -> None: ) -> None:
''' '''
Cancel subactor gracefully once it's "main" portal's Cancel actor gracefully once it's "main" portal's
result arrives. result arrives.
Should only be called for actors spawned with `run_in_actor()`. Should only be called for actors spawned with `run_in_actor()`.
''' '''
await pack_and_report_errors( # if this call errors we store the exception for later
portal, # in ``errors`` which will be reraised inside
subactor, # a MultiError and we still send out a cancel request
errors, result = await exhaust_portal(portal, actor)
) if isinstance(result, Exception):
errors[actor.uid] = result
log.warning(
f"Cancelling {portal.channel.uid} after error {result}"
)
else:
log.runtime(
f"Cancelling {portal.channel.uid} gracefully "
f"after result {result}")
# cancel the process now that we have a final result # cancel the process now that we have a final result
await portal.cancel_actor() await portal.cancel_actor()
@ -371,9 +344,8 @@ async def new_proc(
with trio.CancelScope(shield=True): with trio.CancelScope(shield=True):
await actor_nursery._join_procs.wait() await actor_nursery._join_procs.wait()
cancel_on_complete = portal in actor_nursery._cancel_after_result_on_exit
async with trio.open_nursery() as nursery: async with trio.open_nursery() as nursery:
if cancel_on_complete: if portal in actor_nursery._cancel_after_result_on_exit:
nursery.start_soon( nursery.start_soon(
cancel_on_completion, cancel_on_completion,
portal, portal,
@ -397,11 +369,6 @@ async def new_proc(
f"{subactor.uid}") f"{subactor.uid}")
nursery.cancel_scope.cancel() nursery.cancel_scope.cancel()
# if errors:
# log.warning(
# f'Remote errors retreived from child: {subactor.uid}')
# actor_nursery._ria_nursery.cancel_scope.cancel()
finally: finally:
# The "hard" reap since no actor zombies are allowed! # The "hard" reap since no actor zombies are allowed!
# XXX: do this **after** cancellation/tearfown to avoid # XXX: do this **after** cancellation/tearfown to avoid
@ -431,30 +398,11 @@ async def new_proc(
else: else:
log.warning('Nursery cancelled before sub-proc started') log.warning('Nursery cancelled before sub-proc started')
uid = subactor.uid
if not cancelled_during_spawn: if not cancelled_during_spawn:
subactor, _, portal = actor_nursery._children.pop(uid) # pop child entry to indicate we no longer managing this
# subactor
actor_nursery._children.pop(subactor.uid)
# check for a late delivery of an error from
# the target remote task and overwrite any cancel
# that was captured as part of teardown.
if cancel_on_complete:
error = errors.get(uid)
if type(error) is trio.Cancelled:
# actor was cancelled before it's final result was
# retreived so check now for any result and pack as
# an error to be raised in the surrounding
# nursery's multierror handling.
errors.pop(uid)
with trio.move_on_after(0.001) as cs:
cs.shield = True
err = await pack_and_report_errors(
portal,
subactor,
errors,
)
if type(err) is trio.Cancelled:
errors.pop(uid)
else: else:
# `multiprocessing` # `multiprocessing`
# async with trio.open_nursery() as nursery: # async with trio.open_nursery() as nursery: