py/scheduler: Add assert that scheduler is locked when unlocking.

And add a test that shows how this can happen when multiple threads are
accessing the scheduler, which fails if atomic sections are not used.

1 files changed, 49 insertions, 0 deletions

diff --git a/tests/thread/stress_schedule.py b/tests/thread/stress_schedule.py
new file mode 100644
index 000000000..c5a402b3a
--- /dev/null
+++ b/tests/thread/stress_schedule.py
@@ -0,0 +1,49 @@
+# This test ensures that the scheduler doesn't trigger any assertions
+# while dealing with concurrent access from multiple threads.
+
+import _thread
+import utime
+import micropython
+import gc
+
+try:
+    micropython.schedule
+except AttributeError:
+    print("SKIP")
+    raise SystemExit
+
+gc.disable()
+
+n = 0  # How many times the task successfully ran.
+
+
+def task(x):
+    global n
+    n += 1
+
+
+def thread():
+    while True:
+        try:
+            micropython.schedule(task, None)
+        except RuntimeError:
+            # Queue full, back off.
+            utime.sleep_ms(10)
+
+
+for i in range(8):
+    _thread.start_new_thread(thread, ())
+
+_NUM_TASKS = const(10000)
+_TIMEOUT_MS = const(10000)
+
+# Wait up to 10 seconds for 10000 tasks to be scheduled.
+t = utime.ticks_ms()
+while n < _NUM_TASKS and utime.ticks_diff(utime.ticks_ms(), t) < _TIMEOUT_MS:
+    pass
+
+if n < _NUM_TASKS:
+    # Not all the tasks were scheduled, likely the scheduler stopped working.
+    print(n)
+else:
+    print("PASS")