circuitpython/tests/extmod/asyncio_event_fair.py

50 lines
892 B
Python

# Test fairness of Event.set()
# That tasks which continuously wait on events don't take over the scheduler
try:
import asyncio
except ImportError:
print("SKIP")
raise SystemExit
# CIRCUITPY-CHANGE: CircuitPython provides __await__()
async def foo():
return 42
try:
foo().__await__
except AttributeError:
print("SKIP")
raise SystemExit
async def task1(id):
for i in range(4):
print("sleep", id)
await asyncio.sleep(0)
async def task2(id, ev):
for i in range(4):
ev.set()
ev.clear()
print("wait", id)
await ev.wait()
async def main():
ev = asyncio.Event()
tasks = [
asyncio.create_task(task1(0)),
asyncio.create_task(task2(2, ev)),
asyncio.create_task(task1(1)),
asyncio.create_task(task2(3, ev)),
]
await tasks[1]
ev.set()
asyncio.run(main())