Created
April 15, 2020 10:18
-
-
Save vxgmichel/7ec803ebcd494a3f77365c163bf5b488 to your computer and use it in GitHub Desktop.
Run a trio example mixing cancellation and exceptions
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
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