Compare commits
4 Commits
1d59f1963c
...
cf56f33be6
| Author | SHA1 | Date |
|---|---|---|
|
|
cf56f33be6 | |
|
|
8739c5fadb | |
|
|
dfdaf2b1c1 | |
|
|
37eeb7bab6 |
|
|
@ -0,0 +1,5 @@
|
||||||
|
Remove legacy ``ActorNursery.run_in_actor()``,
|
||||||
|
``Portal.wait_for_result()`` and ``Portal.result()``. Use
|
||||||
|
``tractor.to_actor.run()`` for caller-owned one-shot tasks,
|
||||||
|
``Portal.run()`` for daemon RPC results or ``Portal.open_context()``
|
||||||
|
for linked task dialogs.
|
||||||
|
|
@ -94,6 +94,18 @@ async def sleep_forever():
|
||||||
await trio.sleep_forever()
|
await trio.sleep_forever()
|
||||||
|
|
||||||
|
|
||||||
|
@tractor.context
|
||||||
|
async def sleep_forever_ctx(
|
||||||
|
ctx: tractor.Context,
|
||||||
|
) -> None:
|
||||||
|
'''
|
||||||
|
Signal task startup before sleeping until context cancellation.
|
||||||
|
|
||||||
|
'''
|
||||||
|
await ctx.started()
|
||||||
|
await sleep_forever()
|
||||||
|
|
||||||
|
|
||||||
async def do_nuthin():
|
async def do_nuthin():
|
||||||
# just nick the scheduler
|
# just nick the scheduler
|
||||||
await trio.sleep(0)
|
await trio.sleep(0)
|
||||||
|
|
@ -204,9 +216,39 @@ def test_multierror(
|
||||||
with pytest.raises((
|
with pytest.raises((
|
||||||
BaseExceptionGroup,
|
BaseExceptionGroup,
|
||||||
tractor.RemoteActorError,
|
tractor.RemoteActorError,
|
||||||
)):
|
)) as excinfo:
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
||||||
|
exc = excinfo.value
|
||||||
|
if isinstance(exc, tractor.RemoteActorError):
|
||||||
|
assert exc.boxed_type is AssertionError
|
||||||
|
return
|
||||||
|
|
||||||
|
def iter_group_leaves(
|
||||||
|
group: BaseExceptionGroup,
|
||||||
|
):
|
||||||
|
for subexc in group.exceptions:
|
||||||
|
if isinstance(subexc, BaseExceptionGroup):
|
||||||
|
yield from iter_group_leaves(subexc)
|
||||||
|
else:
|
||||||
|
yield subexc
|
||||||
|
|
||||||
|
assertion_errors: list[tractor.RemoteActorError] = []
|
||||||
|
cancellations: list[BaseException] = []
|
||||||
|
for leaf in iter_group_leaves(exc):
|
||||||
|
if isinstance(leaf, tractor.ContextCancelled):
|
||||||
|
cancellations.append(leaf)
|
||||||
|
elif isinstance(leaf, trio.Cancelled):
|
||||||
|
cancellations.append(leaf)
|
||||||
|
else:
|
||||||
|
assert isinstance(leaf, tractor.RemoteActorError)
|
||||||
|
assert leaf.boxed_type is AssertionError
|
||||||
|
assertion_errors.append(leaf)
|
||||||
|
|
||||||
|
assert len(assertion_errors) in (1, 2)
|
||||||
|
if not cancellations:
|
||||||
|
assert len(assertion_errors) == 2
|
||||||
|
|
||||||
|
|
||||||
async def do_nothing():
|
async def do_nothing():
|
||||||
pass
|
pass
|
||||||
|
|
@ -394,7 +436,7 @@ async def test_some_cancels_all(
|
||||||
|
|
||||||
except tractor.RemoteActorError as err:
|
except tractor.RemoteActorError as err:
|
||||||
assert err.boxed_type == err_type
|
assert err.boxed_type == err_type
|
||||||
# we only expect this first error to propogate
|
# we only expect this first error to propagate
|
||||||
# (all other daemons are cancelled before they
|
# (all other daemons are cancelled before they
|
||||||
# can be scheduled)
|
# can be scheduled)
|
||||||
num_actors = 1
|
num_actors = 1
|
||||||
|
|
@ -403,7 +445,7 @@ async def test_some_cancels_all(
|
||||||
else:
|
else:
|
||||||
if expect_error:
|
if expect_error:
|
||||||
pytest.fail(
|
pytest.fail(
|
||||||
"Deamon call should fail at checkpoint?")
|
"Daemon call should fail at checkpoint?")
|
||||||
|
|
||||||
# should error here with a `RemoteActorError` or a beg of them
|
# should error here with a `RemoteActorError` or a beg of them
|
||||||
|
|
||||||
|
|
@ -827,15 +869,25 @@ def test_cancel_via_SIGINT_other_task(
|
||||||
async with tractor.open_nursery(
|
async with tractor.open_nursery(
|
||||||
registry_addrs=[reg_addr],
|
registry_addrs=[reg_addr],
|
||||||
) as an:
|
) as an:
|
||||||
# just keep a set of (daemon) subactors alive for the
|
portals = [
|
||||||
# SIGINT to cancel (was 3 `run_in_actor(sleep_forever)`
|
|
||||||
# one-shots — a daemon needs no "main" task to idle).
|
|
||||||
for i in range(3):
|
|
||||||
await an.start_actor(
|
await an.start_actor(
|
||||||
f'namesucka_{i}',
|
f'namesucka_{i}',
|
||||||
|
enable_modules=[__name__],
|
||||||
)
|
)
|
||||||
task_status.started()
|
for i in range(3)
|
||||||
await trio.sleep_forever()
|
]
|
||||||
|
|
||||||
|
# Keep one linked RPC task active in every daemon before
|
||||||
|
# reporting startup, preserving the original
|
||||||
|
# `run_in_actor(sleep_forever)` cancellation target.
|
||||||
|
async with gather_contexts(
|
||||||
|
mngrs=[
|
||||||
|
portal.open_context(sleep_forever_ctx)
|
||||||
|
for portal in portals
|
||||||
|
],
|
||||||
|
):
|
||||||
|
task_status.started()
|
||||||
|
await trio.sleep_forever()
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
# should never timeout since SIGINT should cancel the current program
|
# should never timeout since SIGINT should cancel the current program
|
||||||
|
|
|
||||||
|
|
@ -667,7 +667,7 @@ def test_basic_interloop_channel_stream(
|
||||||
async with tractor.open_nursery(
|
async with tractor.open_nursery(
|
||||||
registry_addrs=[reg_addr],
|
registry_addrs=[reg_addr],
|
||||||
) as an:
|
) as an:
|
||||||
# should raise RAE diectly
|
# should raise RAE directly
|
||||||
await to_actor.run(
|
await to_actor.run(
|
||||||
partial(
|
partial(
|
||||||
stream_from_aio,
|
stream_from_aio,
|
||||||
|
|
@ -726,7 +726,7 @@ def test_trio_closes_early_causes_aio_checkpoint_raise(
|
||||||
# enable_stack_on_sig=True,
|
# enable_stack_on_sig=True,
|
||||||
registry_addrs=[reg_addr],
|
registry_addrs=[reg_addr],
|
||||||
) as an:
|
) as an:
|
||||||
# should raise RAE diectly
|
# should raise RAE directly
|
||||||
print('waiting on final infected subactor result..')
|
print('waiting on final infected subactor result..')
|
||||||
res: None = await to_actor.run(
|
res: None = await to_actor.run(
|
||||||
partial(
|
partial(
|
||||||
|
|
@ -779,7 +779,7 @@ def test_aio_exits_early_relays_AsyncioTaskExited(
|
||||||
debug_mode=debug_mode,
|
debug_mode=debug_mode,
|
||||||
# enable_stack_on_sig=True,
|
# enable_stack_on_sig=True,
|
||||||
) as an:
|
) as an:
|
||||||
# should raise RAE diectly
|
# should raise RAE directly
|
||||||
print('waiting on final infected subactor result..')
|
print('waiting on final infected subactor result..')
|
||||||
res: None = await to_actor.run(
|
res: None = await to_actor.run(
|
||||||
partial(
|
partial(
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue