tractor/examples/debugging/root_timeout_while_child_cr...

47 lines
1.1 KiB
Python
Raw Normal View History

import trio
import tractor
async def key_error() -> None:
'''
Raise a ``KeyError``.
'''
return {}['doggy']
async def main() -> None:
'''
Root is fail-after-cancelled while blocking and child RPC fails
simultaneously.
'''
async with (
tractor.open_nursery(
debug_mode=True,
# loglevel='debug' # ?XXX required?
) as an,
trio.open_nursery() as tn,
):
# spawn the actor..
portal: tractor.Portal = await an.start_actor(
'key_error',
enable_modules=[__name__],
)
print(
f'Child is up @ {portal.chan.aid.reprol()}'
)
# ..then schedule its erroring task in the bg while the
# root blocks below.
tn.start_soon(portal.run, key_error)
# XXX: originally a bug caused by this is where root would
# enter the debugger and clobber the tty used by the repl
# even though child should have it locked.
with trio.fail_after(1):
await trio.Event().wait()
if __name__ == '__main__':
trio.run(main)