structio/tests/self_cancel.py

29 lines
915 B
Python

import structio
async def sleeper(n):
print(f"[sleeper] Going to sleep for {n} seconds!")
i = structio.clock()
try:
await structio.sleep(n)
except structio.Cancelled:
print(f"[sleeper] Oh no, I've been cancelled! (was gonna sleep {structio.clock() - i:.2f} more seconds)")
raise
print("[sleeper] Woke up!")
async def main(n, o, p):
print(f"[main] Parent is alive, spawning {n} children sleeping {o} seconds each")
t = structio.clock()
async with structio.create_pool() as pool:
for i in range(o):
pool.spawn(sleeper, n)
print(f"[main] Children spawned, sleeping {p} seconds before cancelling")
await structio.sleep(p)
# Note that cancellations propagate to all inner task scopes!
pool.scope.cancel()
print(f"[main] Parent exited in {structio.clock() - t:.2f} seconds")
structio.run(main, 5, 2, 2)