mirror of
https://github.com/micropython/micropython.git
synced 2026-01-28 14:50:21 +01:00
f918e33645
This test can now run correctly with the native emitter because the `thread()` function will run the scheduler during the call to `time.sleep_ms(1)`. Signed-off-by: Jeff Epler <jepler@unpythonic.net>
62 lines
1.3 KiB
Python
62 lines
1.3 KiB
Python
# This test ensures that the scheduler doesn't trigger any assertions
|
|
# while dealing with concurrent access from multiple threads.
|
|
|
|
import _thread
|
|
import time
|
|
import micropython
|
|
import gc
|
|
|
|
try:
|
|
micropython.schedule
|
|
except AttributeError:
|
|
print("SKIP")
|
|
raise SystemExit
|
|
|
|
gc.disable()
|
|
|
|
_NUM_TASKS = 10000
|
|
_TIMEOUT_MS = 10000
|
|
|
|
n = 0 # How many times the task successfully ran.
|
|
t = None # Start time of test, assigned here to preallocate entry in globals dict.
|
|
thread_run = True # If the thread should continue running.
|
|
|
|
|
|
def task(x):
|
|
global n
|
|
n += 1
|
|
|
|
|
|
def thread():
|
|
while thread_run:
|
|
try:
|
|
micropython.schedule(task, None)
|
|
except RuntimeError:
|
|
# Queue full, back off.
|
|
time.sleep_ms(1)
|
|
|
|
|
|
for i in range(8):
|
|
try:
|
|
_thread.start_new_thread(thread, ())
|
|
except OSError:
|
|
# System cannot create a new thead, so stop trying to create them.
|
|
break
|
|
|
|
# Wait up to 10 seconds for 10000 tasks to be scheduled.
|
|
t = time.ticks_ms()
|
|
while n < _NUM_TASKS and time.ticks_diff(time.ticks_ms(), t) < _TIMEOUT_MS:
|
|
time.sleep(0)
|
|
|
|
# Stop all threads.
|
|
thread_run = False
|
|
time.sleep_ms(20)
|
|
|
|
gc.enable()
|
|
|
|
if n < _NUM_TASKS:
|
|
# Not all the tasks were scheduled, likely the scheduler stopped working.
|
|
print(n)
|
|
else:
|
|
print("PASS")
|