structio/tests/nested_pool_outer_raises.py

61 lines
2.3 KiB
Python

import structio
from nested_pool_inner_raises import successful, failing
async def main_simple(
children_outer: list[tuple[str, int]], children_inner: list[tuple[str, int]]
):
before = structio.clock()
try:
async with structio.create_pool() as p1:
print(f"[main] Spawning children in first context ({hex(id(p1))})")
for name, delay in children_outer:
p1.spawn(failing, name, delay)
print("[main] Children spawned")
async with structio.create_pool() as p2:
print(f"[main] Spawning children in second context ({hex(id(p2))})")
for name, delay in children_inner:
p2.spawn(successful, name, delay)
print("[main] Children spawned")
except TypeError:
print("[main] TypeError caught!")
print(f"[main] Children exited in {structio.clock() - before:.2f} seconds")
async def main_nested(
children_outer: list[tuple[str, int]], children_inner: list[tuple[str, int]]
):
before = structio.clock()
try:
async with structio.create_pool() as p1:
print(f"[main] Spawning children in first context ({hex(id(p1))})")
for name, delay in children_outer:
p1.spawn(failing, name, delay)
print("[main] Children spawned")
async with structio.create_pool() as p2:
print(f"[main] Spawning children in second context ({hex(id(p2))})")
for name, delay in children_inner:
p2.spawn(successful, name, delay)
print("[main] Children spawned")
async with structio.create_pool() as p3:
print(f"[main] Spawning children in third context ({hex(id(p3))})")
for name, delay in children_inner:
p3.spawn(successful, name, delay)
print("[main] Children spawned")
except TypeError:
print("[main] TypeError caught!")
print(f"[main] Children exited in {structio.clock() - before:.2f} seconds")
if __name__ == "__main__":
structio.run(
main_simple,
[("second", 2), ("third", 3)],
[("first", 1), ("fourth", 4)],
)
structio.run(
main_nested,
[("second", 2), ("third", 3)],
[("first", 1), ("fourth", 4)],
)