b070765427
The native emitter will not release/bounce the GIL when running code, so if it runs tight loops then no other threads get a chance to run (if the GIL is enabled). So for the thread tests, explicitly include a call to `time.sleep(0)` (or equivalent) to bounce the GIL and give other threads a chance to run. For some tests (eg `thread_coop.py`) the whole point of the test is to test that the GIL is correctly bounced. So for those cases force the use of the bytecode emitter for the busy functions. Signed-off-by: Damien George <damien@micropython.org>
45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
# Test that having multiple threads block on stdin doesn't cause any issues.
|
|
#
|
|
# The test doesn't expect any input on stdin.
|
|
#
|
|
# This is a regression test for https://github.com/micropython/micropython/issues/15230
|
|
# on rp2, but doubles as a general property to test across all ports.
|
|
import sys
|
|
import time
|
|
import _thread
|
|
|
|
try:
|
|
import select
|
|
except ImportError:
|
|
print("SKIP")
|
|
raise SystemExit
|
|
|
|
|
|
class StdinWaiter:
|
|
def __init__(self):
|
|
self._done = False
|
|
|
|
def wait_stdin(self, timeout_ms):
|
|
poller = select.poll()
|
|
poller.register(sys.stdin, select.POLLIN)
|
|
poller.poll(timeout_ms)
|
|
# Ignoring the poll result as we don't expect any input
|
|
self._done = True
|
|
|
|
def is_done(self):
|
|
return self._done
|
|
|
|
|
|
thread_waiter = StdinWaiter()
|
|
_thread.start_new_thread(thread_waiter.wait_stdin, (1000,))
|
|
StdinWaiter().wait_stdin(1000)
|
|
|
|
# Spinning here is mostly not necessary but there is some inconsistency waking
|
|
# the two threads, especially on CPython CI runners where the thread may not
|
|
# have run yet. The actual delay is <20ms but spinning here instead of
|
|
# sleep(0.1) means the test can run on MP builds without float support.
|
|
while not thread_waiter.is_done():
|
|
time.sleep(0)
|
|
|
|
# The background thread should have completed its wait by now.
|
|
print(thread_waiter.is_done())
|