Compare commits
3 Commits
89551ef371
...
f4cab2ddb3
Author | SHA1 | Date |
---|---|---|
Tyler Goodlet | f4cab2ddb3 | |
Tyler Goodlet | 97b79064f5 | |
Tyler Goodlet | a8d71f15b7 |
|
@ -326,16 +326,19 @@ 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
|
||||||
|
@ -359,6 +362,7 @@ 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:
|
||||||
|
@ -394,16 +398,13 @@ 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 not sure what's up with this..
|
# XXX it's race whether or not a parent containing
|
||||||
# on windows sometimes spawning is just too slow and
|
# a nursery *may* get multiple child failures before
|
||||||
# we get back the (sent) cancel signal instead
|
# it cancels and tears down.
|
||||||
if platform.system() == 'Windows':
|
|
||||||
if isinstance(subexc, tractor.RemoteActorError):
|
if isinstance(subexc, tractor.RemoteActorError):
|
||||||
assert subexc.type in (trio.MultiError, tractor.RemoteActorError)
|
assert subexc.type in (trio.MultiError, tractor.RemoteActorError)
|
||||||
else:
|
else:
|
||||||
assert isinstance(subexc, trio.MultiError)
|
assert isinstance(subexc, trio.MultiError)
|
||||||
else:
|
|
||||||
assert subexc.type is trio.MultiError
|
|
||||||
else:
|
else:
|
||||||
assert subexc.type in (tractor.RemoteActorError, trio.Cancelled)
|
assert subexc.type in (tractor.RemoteActorError, trio.Cancelled)
|
||||||
|
|
||||||
|
@ -486,9 +487,11 @@ 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
|
"""
|
||||||
down even when that cancellation is triggered by the parent
|
Verify that a child cancelled while executing sync code is torn down
|
||||||
|
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...")
|
||||||
|
@ -500,7 +503,7 @@ def test_cancel_while_childs_child_in_sync_sleep(
|
||||||
spawn,
|
spawn,
|
||||||
name='spawn',
|
name='spawn',
|
||||||
)
|
)
|
||||||
await trio.sleep(1)
|
await trio.sleep(0.5)
|
||||||
assert 0
|
assert 0
|
||||||
|
|
||||||
with pytest.raises(AssertionError):
|
with pytest.raises(AssertionError):
|
||||||
|
|
|
@ -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
|
from ._exceptions import ActorFailure, NoResult
|
||||||
|
|
||||||
|
|
||||||
log = get_logger('tractor')
|
log = get_logger('tractor')
|
||||||
|
@ -136,48 +136,75 @@ 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 (Exception, trio.MultiError) as err:
|
except (
|
||||||
|
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,
|
||||||
actor: Actor,
|
subactor: Actor,
|
||||||
errors: Dict[Tuple[str, str], Exception],
|
errors: Dict[Tuple[str, str], Exception],
|
||||||
|
|
||||||
) -> None:
|
) -> None:
|
||||||
'''
|
'''
|
||||||
Cancel actor gracefully once it's "main" portal's
|
Cancel subactor 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()`.
|
||||||
|
|
||||||
'''
|
'''
|
||||||
# if this call errors we store the exception for later
|
await pack_and_report_errors(
|
||||||
# in ``errors`` which will be reraised inside
|
portal,
|
||||||
# a MultiError and we still send out a cancel request
|
subactor,
|
||||||
result = await exhaust_portal(portal, actor)
|
errors,
|
||||||
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()
|
||||||
|
|
||||||
|
@ -344,8 +371,9 @@ 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 portal in actor_nursery._cancel_after_result_on_exit:
|
if cancel_on_complete:
|
||||||
nursery.start_soon(
|
nursery.start_soon(
|
||||||
cancel_on_completion,
|
cancel_on_completion,
|
||||||
portal,
|
portal,
|
||||||
|
@ -369,6 +397,11 @@ 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
|
||||||
|
@ -398,11 +431,30 @@ 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:
|
||||||
# pop child entry to indicate we no longer managing this
|
subactor, _, portal = actor_nursery._children.pop(uid)
|
||||||
# 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:
|
||||||
|
|
Loading…
Reference in New Issue