Compare commits
6 Commits
23240c31e3
...
5fc64107e5
Author | SHA1 | Date |
---|---|---|
|
5fc64107e5 | |
|
6348c83d28 | |
|
8b8390e83c | |
|
f5c6fc2f02 | |
|
444b9bfc22 | |
|
79e70a9b08 |
|
@ -546,40 +546,123 @@ def test_cancel_via_SIGINT_other_task(
|
||||||
|
|
||||||
async def spin_for(period=3):
|
async def spin_for(period=3):
|
||||||
"Sync sleep."
|
"Sync sleep."
|
||||||
|
print(f'sync sleeping in sub-sub for {period}\n')
|
||||||
time.sleep(period)
|
time.sleep(period)
|
||||||
|
|
||||||
|
|
||||||
async def spawn():
|
async def spawn_sub_with_sync_blocking_task():
|
||||||
async with tractor.open_nursery() as tn:
|
async with tractor.open_nursery() as an:
|
||||||
await tn.run_in_actor(
|
print('starting sync blocking subactor..\n')
|
||||||
|
await an.run_in_actor(
|
||||||
spin_for,
|
spin_for,
|
||||||
name='sleeper',
|
name='sleeper',
|
||||||
)
|
)
|
||||||
|
print('exiting first subactor layer..\n')
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
'man_cancel_outer',
|
||||||
|
[
|
||||||
|
False, # passes if delay != 2
|
||||||
|
|
||||||
|
# always causes an unexpected eg-w-embedded-assert-err?
|
||||||
|
pytest.param(True,
|
||||||
|
marks=pytest.mark.xfail(
|
||||||
|
reason=(
|
||||||
|
'always causes an unexpected eg-w-embedded-assert-err?'
|
||||||
|
)
|
||||||
|
),
|
||||||
|
),
|
||||||
|
],
|
||||||
|
)
|
||||||
@no_windows
|
@no_windows
|
||||||
def test_cancel_while_childs_child_in_sync_sleep(
|
def test_cancel_while_childs_child_in_sync_sleep(
|
||||||
loglevel,
|
loglevel: str,
|
||||||
start_method,
|
start_method: str,
|
||||||
spawn_backend,
|
spawn_backend: str,
|
||||||
|
debug_mode: bool,
|
||||||
|
reg_addr: tuple,
|
||||||
|
man_cancel_outer: bool,
|
||||||
):
|
):
|
||||||
"""Verify that a child cancelled while executing sync code is torn
|
'''
|
||||||
|
Verify that a child cancelled while executing sync code is torn
|
||||||
down even when that cancellation is triggered by the parent
|
down even when that cancellation is triggered by the parent
|
||||||
2 nurseries "up".
|
2 nurseries "up".
|
||||||
"""
|
|
||||||
|
Though the grandchild should stay blocking its actor runtime, its
|
||||||
|
parent should issue a "zombie reaper" to hard kill it after
|
||||||
|
sufficient timeout.
|
||||||
|
|
||||||
|
'''
|
||||||
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...")
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
with trio.fail_after(2):
|
#
|
||||||
|
# XXX BIG TODO NOTE XXX
|
||||||
|
#
|
||||||
|
# it seems there's a strange race that can happen
|
||||||
|
# where where the fail-after will trigger outer scope
|
||||||
|
# .cancel() which then causes the inner scope to raise,
|
||||||
|
#
|
||||||
|
# BaseExceptionGroup('Exceptions from Trio nursery', [
|
||||||
|
# BaseExceptionGroup('Exceptions from Trio nursery',
|
||||||
|
# [
|
||||||
|
# Cancelled(),
|
||||||
|
# Cancelled(),
|
||||||
|
# ]
|
||||||
|
# ),
|
||||||
|
# AssertionError('assert 0')
|
||||||
|
# ])
|
||||||
|
#
|
||||||
|
# WHY THIS DOESN'T MAKE SENSE:
|
||||||
|
# ---------------------------
|
||||||
|
# - it should raise too-slow-error when too slow..
|
||||||
|
# * verified that using simple-cs and manually cancelling
|
||||||
|
# you get same outcome -> indicates that the fail-after
|
||||||
|
# can have its TooSlowError overriden!
|
||||||
|
# |_ to check this it's easy, simplly decrease the timeout
|
||||||
|
# as per the var below.
|
||||||
|
#
|
||||||
|
# - when using the manual simple-cs the outcome is different
|
||||||
|
# DESPITE the `assert 0` which means regardless of the
|
||||||
|
# inner scope effectively failing in the same way, the
|
||||||
|
# bubbling up **is NOT the same**.
|
||||||
|
#
|
||||||
|
# delays trigger diff outcomes..
|
||||||
|
# ---------------------------
|
||||||
|
# as seen by uncommenting various lines below there is from
|
||||||
|
# my POV an unexpected outcome due to the delay=2 case.
|
||||||
|
#
|
||||||
|
# delay = 1 # no AssertionError in eg, TooSlowError raised.
|
||||||
|
# delay = 2 # is AssertionError in eg AND no TooSlowError !?
|
||||||
|
delay = 4 # is AssertionError in eg AND no _cs cancellation.
|
||||||
|
|
||||||
|
with trio.fail_after(delay) as _cs:
|
||||||
|
# with trio.CancelScope() as cs:
|
||||||
|
# ^XXX^ can be used instead to see same outcome.
|
||||||
|
|
||||||
async with (
|
async with (
|
||||||
tractor.open_nursery() as an
|
# tractor.trionics.collapse_eg(), # doesn't help
|
||||||
|
tractor.open_nursery(
|
||||||
|
hide_tb=False,
|
||||||
|
debug_mode=debug_mode,
|
||||||
|
registry_addrs=[reg_addr],
|
||||||
|
) as an,
|
||||||
):
|
):
|
||||||
await an.run_in_actor(
|
await an.run_in_actor(
|
||||||
spawn,
|
spawn_sub_with_sync_blocking_task,
|
||||||
name='spawn',
|
name='sync_blocking_sub',
|
||||||
)
|
)
|
||||||
await trio.sleep(1)
|
await trio.sleep(1)
|
||||||
|
|
||||||
|
if man_cancel_outer:
|
||||||
|
print('Cancelling manually in root')
|
||||||
|
_cs.cancel()
|
||||||
|
|
||||||
|
# trigger exc-srced taskc down
|
||||||
|
# the actor tree.
|
||||||
|
print('RAISING IN ROOT')
|
||||||
assert 0
|
assert 0
|
||||||
|
|
||||||
with pytest.raises(AssertionError):
|
with pytest.raises(AssertionError):
|
||||||
|
|
|
@ -13,26 +13,24 @@ MESSAGE = 'tractoring at full speed'
|
||||||
def test_empty_mngrs_input_raises() -> None:
|
def test_empty_mngrs_input_raises() -> None:
|
||||||
|
|
||||||
async def main():
|
async def main():
|
||||||
with trio.fail_after(1):
|
with trio.fail_after(3):
|
||||||
async with (
|
async with (
|
||||||
open_actor_cluster(
|
open_actor_cluster(
|
||||||
modules=[__name__],
|
modules=[__name__],
|
||||||
|
|
||||||
# NOTE: ensure we can passthrough runtime opts
|
# NOTE: ensure we can passthrough runtime opts
|
||||||
loglevel='info',
|
loglevel='cancel',
|
||||||
# debug_mode=True,
|
debug_mode=False,
|
||||||
|
|
||||||
) as portals,
|
) as portals,
|
||||||
|
|
||||||
gather_contexts(
|
gather_contexts(mngrs=()),
|
||||||
# NOTE: it's the use of inline-generator syntax
|
|
||||||
# here that causes the empty input.
|
|
||||||
mngrs=(
|
|
||||||
p.open_context(worker) for p in portals.values()
|
|
||||||
),
|
|
||||||
),
|
|
||||||
):
|
):
|
||||||
assert 0
|
# should fail before this?
|
||||||
|
assert portals
|
||||||
|
|
||||||
|
# test should fail if we mk it here!
|
||||||
|
assert 0, 'Should have raised val-err !?'
|
||||||
|
|
||||||
with pytest.raises(ValueError):
|
with pytest.raises(ValueError):
|
||||||
trio.run(main)
|
trio.run(main)
|
||||||
|
|
|
@ -55,10 +55,17 @@ async def open_actor_cluster(
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
'Number of names is {len(names)} but count it {count}')
|
'Number of names is {len(names)} but count it {count}')
|
||||||
|
|
||||||
async with tractor.open_nursery(
|
async with (
|
||||||
**runtime_kwargs,
|
# tractor.trionics.collapse_eg(),
|
||||||
) as an:
|
tractor.open_nursery(
|
||||||
async with trio.open_nursery() as n:
|
**runtime_kwargs,
|
||||||
|
) as an
|
||||||
|
):
|
||||||
|
async with (
|
||||||
|
# tractor.trionics.collapse_eg(),
|
||||||
|
trio.open_nursery() as tn,
|
||||||
|
tractor.trionics.maybe_raise_from_masking_exc()
|
||||||
|
):
|
||||||
uid = tractor.current_actor().uid
|
uid = tractor.current_actor().uid
|
||||||
|
|
||||||
async def _start(name: str) -> None:
|
async def _start(name: str) -> None:
|
||||||
|
@ -69,9 +76,8 @@ async def open_actor_cluster(
|
||||||
)
|
)
|
||||||
|
|
||||||
for name in names:
|
for name in names:
|
||||||
n.start_soon(_start, name)
|
tn.start_soon(_start, name)
|
||||||
|
|
||||||
assert len(portals) == count
|
assert len(portals) == count
|
||||||
yield portals
|
yield portals
|
||||||
|
|
||||||
await an.cancel(hard_kill=hard_kill)
|
await an.cancel(hard_kill=hard_kill)
|
||||||
|
|
|
@ -174,7 +174,6 @@ class Actor:
|
||||||
msg_buffer_size: int = 2**6
|
msg_buffer_size: int = 2**6
|
||||||
|
|
||||||
# nursery placeholders filled in by `async_main()` after fork
|
# nursery placeholders filled in by `async_main()` after fork
|
||||||
_root_n: Nursery|None = None
|
|
||||||
_service_n: Nursery|None = None
|
_service_n: Nursery|None = None
|
||||||
|
|
||||||
_ipc_server: _server.IPCServer|None = None
|
_ipc_server: _server.IPCServer|None = None
|
||||||
|
@ -1479,8 +1478,8 @@ async def async_main(
|
||||||
collapse_eg(),
|
collapse_eg(),
|
||||||
trio.open_nursery() as root_tn,
|
trio.open_nursery() as root_tn,
|
||||||
):
|
):
|
||||||
actor._root_n = root_tn
|
# actor._root_n = root_tn
|
||||||
assert actor._root_n
|
# assert actor._root_n
|
||||||
|
|
||||||
ipc_server: _server.IPCServer
|
ipc_server: _server.IPCServer
|
||||||
async with (
|
async with (
|
||||||
|
|
|
@ -78,7 +78,6 @@ def collapse_exception_group(
|
||||||
def get_collapsed_eg(
|
def get_collapsed_eg(
|
||||||
beg: BaseExceptionGroup,
|
beg: BaseExceptionGroup,
|
||||||
|
|
||||||
bp: bool = False,
|
|
||||||
) -> BaseException|None:
|
) -> BaseException|None:
|
||||||
'''
|
'''
|
||||||
If the input beg can collapse to a single sub-exception which is
|
If the input beg can collapse to a single sub-exception which is
|
||||||
|
@ -92,7 +91,6 @@ def get_collapsed_eg(
|
||||||
return maybe_exc
|
return maybe_exc
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@acm
|
@acm
|
||||||
async def collapse_eg(
|
async def collapse_eg(
|
||||||
hide_tb: bool = True,
|
hide_tb: bool = True,
|
||||||
|
@ -102,6 +100,8 @@ async def collapse_eg(
|
||||||
# trio.Cancelled,
|
# trio.Cancelled,
|
||||||
},
|
},
|
||||||
add_notes: bool = True,
|
add_notes: bool = True,
|
||||||
|
|
||||||
|
bp: bool = False,
|
||||||
):
|
):
|
||||||
'''
|
'''
|
||||||
If `BaseExceptionGroup` raised in the body scope is
|
If `BaseExceptionGroup` raised in the body scope is
|
||||||
|
@ -115,6 +115,20 @@ async def collapse_eg(
|
||||||
yield
|
yield
|
||||||
except BaseExceptionGroup as _beg:
|
except BaseExceptionGroup as _beg:
|
||||||
beg = _beg
|
beg = _beg
|
||||||
|
|
||||||
|
if (
|
||||||
|
bp
|
||||||
|
and
|
||||||
|
len(beg.exceptions) > 1
|
||||||
|
):
|
||||||
|
import tractor
|
||||||
|
if tractor.current_actor(
|
||||||
|
err_on_no_runtime=False,
|
||||||
|
):
|
||||||
|
await tractor.pause(shield=True)
|
||||||
|
else:
|
||||||
|
breakpoint()
|
||||||
|
|
||||||
if (
|
if (
|
||||||
(exc := get_collapsed_eg(beg))
|
(exc := get_collapsed_eg(beg))
|
||||||
and
|
and
|
||||||
|
|
Loading…
Reference in New Issue