blob: e0125b1aefe13d82e3c37d4a979afd3f55a10609 (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
# Ensure that an asyncio task can wait on an Event when the
# _task_queue is empty
# https://github.com/micropython/micropython/issues/16569
try:
import asyncio
except ImportError:
print("SKIP")
raise SystemExit
# This test requires checking that the asyncio scheduler
# remains active "indefinitely" when the task queue is empty.
#
# To check this, we need another independent scheduler that
# can wait for a certain amount of time. So we have to
# create one using micropython.schedule() and time.ticks_ms()
#
# Technically, this code breaks the rules, as it is clearly
# documented that Event.set() should _NOT_ be called from a
# schedule (soft IRQ) because in some cases, a race condition
# can occur, resulting in a crash. However:
# - since the risk of a race condition in that specific
# case has been analysed and excluded
# - given that there is no other simple alternative to
# write this test case,
# an exception to the rule was deemed acceptable. See
# https://github.com/micropython/micropython/pull/16772
import micropython, time
try:
micropython.schedule
except AttributeError:
print("SKIP")
raise SystemExit
evt = asyncio.Event()
def schedule_watchdog(end_ticks):
if time.ticks_diff(end_ticks, time.ticks_ms()) <= 0:
print("asyncio still pending, unlocking event")
# Caution: about to call Event.set() from a schedule
# (see the note in the comment above)
evt.set()
return
micropython.schedule(schedule_watchdog, end_ticks)
async def foo():
print("foo waiting")
schedule_watchdog(time.ticks_add(time.ticks_ms(), 100))
await evt.wait()
print("foo done")
async def main():
print("main started")
await foo()
print("main done")
asyncio.run(main())
|
