Skip to content

Instantly share code, notes, and snippets.

@vxgmichel
Created April 15, 2020 10:18
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save vxgmichel/7ec803ebcd494a3f77365c163bf5b488 to your computer and use it in GitHub Desktop.
Save vxgmichel/7ec803ebcd494a3f77365c163bf5b488 to your computer and use it in GitHub Desktop.
Run a trio example mixing cancellation and exceptions
"""
Run a trio example mixing cancellation and exceptions.
Shows that the behavior can be hard to predict since it depends on factors like:
- whether a nursery is involved
- whether a checkpoint is performed after the raise
Produces the following output:
=========== cancel_and_raise ===========
> RuntimeError caught inside
> No exception outside
===== cancel_and_raise_in_nursery ======
> MultiError raised inside
> RuntimeError caught outside
== cancel_raise_and_await_in_nursery ===
> MultiError raised inside
> No exception outside
== cancel_raise_in_nursery_and_await ===
> Cancelled raised inside
> No exception outside
"""
import trio
async def main(coro):
try:
with trio.CancelScope() as cancel_scope:
try:
await coro(cancel_scope)
except RuntimeError:
print("> RuntimeError caught inside")
pass
except trio.MultiError:
print("> MultiError raised inside")
raise
except trio.Cancelled:
print("> Cancelled raised inside")
raise
except RuntimeError:
print("> RuntimeError caught outside")
else:
print("> No exception outside")
async def cancel_and_raise(cancel_scope):
cancel_scope.cancel()
raise RuntimeError
async def cancel_and_raise_in_nursery(cancel_scope):
async with trio.open_nursery():
cancel_scope.cancel()
raise RuntimeError
async def cancel_raise_and_await_in_nursery(cancel_scope):
async with trio.open_nursery():
try:
cancel_scope.cancel()
raise RuntimeError
finally:
await trio.sleep(0)
async def cancel_raise_in_nursery_and_await(cancel_scope):
try:
async with trio.open_nursery():
cancel_scope.cancel()
raise RuntimeError
finally:
await trio.sleep(0)
coros = [
cancel_and_raise,
cancel_and_raise_in_nursery,
cancel_raise_and_await_in_nursery,
cancel_raise_in_nursery_and_await]
for coro in coros:
print(f" {coro.__name__} ".center(40, "="))
trio.run(main, coro)
print()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment