import structio async def test_silent(i, j): print(f"[test] Parent is alive, exiting after {i:.2f} seconds") k = structio.clock() with structio.skip_after(i) as scope: print(f"[test] Sleeping for {j} seconds") await structio.sleep(j) print(f"[test] Finished in {structio.clock() - k:.2f} seconds (timed out: {scope.timed_out})") async def test_loud(i, j): print(f"[test] Parent is alive, exiting after {i:.2f} seconds") k = structio.clock() try: with structio.with_timeout(i) as scope: print(f"[test] Sleeping for {j} seconds") await structio.sleep(j) except structio.TimedOut: print("[test] Timed out!") print(f"[test] Finished in {structio.clock() - k:.2f} seconds") async def deadlock(): await structio.Event().wait() async def test_deadlock(i): print(f"[test] Parent is alive, will exit in {i} seconds") t = structio.clock() with structio.skip_after(i): print("[test] Entering deadlock") await deadlock() print(f"[test] Done in {structio.clock() - t:.2f} seconds") async def test_nested(i): print(f"[test] Parent is alive, will exit in {i} seconds") t = structio.clock() with structio.skip_after(i): print("[test] Entered first scope") with structio.skip_after(i * 2): # Even though this scope's timeout is # larger than its parent, structio will # still cancel it when its containing # scope expires print("[test] Entered second scope") await deadlock() print(f"[test] Done in {structio.clock() - t:.2f} seconds") structio.run(test_silent, 3, 5) structio.run(test_loud, 3, 5) structio.run(test_deadlock, 5) structio.run(test_nested, 5)