import thread, time from pypy.module.thread.test.support import GenericTestThread import pytest class AppTestThread(GenericTestThread): def test_thread_error(self): import _thread assert _thread.error is RuntimeError # uh, since CPython 3.3 def test_start_new_thread(self): import _thread feedback = [] please_start = [] def f(x, y, z): self.waitfor(lambda: please_start) feedback.append(42) _thread.start_new_thread(f, (1, 2), {'z': 3}) assert feedback == [] # still empty please_start.append(1) # trigger self.waitfor(lambda: feedback) assert feedback == [42] def test_thread_count(self): import _thread, time feedback = [] please_start = [] def f(): feedback.append(42) self.waitfor(lambda: please_start) assert _thread._count() == 0 _thread.start_new_thread(f, ()) self.waitfor(lambda: feedback) assert _thread._count() == 1 please_start.append(1) # trigger # XXX joining a thread seems difficult at applevel. def test_start_new_thread_args(self): import _thread def f(): pass test_args = [ (f, [], {}), (f, (), []), ("", (), {}), ] for args in test_args: try: _thread.start_new_thread(*args) assert False except TypeError: pass def test_get_ident(self): import _thread ident = _thread.get_ident() feedback = [] def f(): feedback.append(_thread.get_ident()) ident2 = _thread.start_new_thread(f, ()) assert ident2 != ident assert ident == _thread.get_ident() self.waitfor(lambda: feedback) assert feedback[0] != ident def test_get_native_id(self): import _thread ident = _thread.get_native_id() feedback = [] def f(): feedback.append(_thread.get_native_id()) ident2 = _thread.start_new_thread(f, ()) assert ident2 != ident assert ident == _thread.get_native_id() self.waitfor(lambda: feedback) assert feedback[0] != ident def test_sys_getframe(self): # this checks that each thread gets its own ExecutionContext. def main(): import _thread, sys def dump_frames(feedback): f = sys._getframe() for i in range(3): if f is None: feedback.append(None) else: feedback.append(f.f_code.co_name) self.busywait(0.04) assert f is sys._getframe(i) f = f.f_back def dummyfn(feedback): dump_frames(feedback) feedback = [] dummyfn(feedback) assert feedback == ['dump_frames', 'dummyfn', 'main'] feedbacks = [] for i in range(3): feedback = [] _thread.start_new_thread(dummyfn, (feedback,)) feedbacks.append(feedback) expected = 3*[['dump_frames', 'dummyfn', None]] # without 'main' self.waitfor(lambda: feedbacks == expected) assert feedbacks == expected main() def test_thread_exit(self): import _thread, sys, io # preemptively import trackback to speed up the call to fn3, # which relies on the space.appexec in error.py:write_unraisable import traceback def fn1(): _thread.exit() def fn2(): raise SystemExit def fn3(): raise ValueError("hello world") prev = sys.stderr try: sys.stderr = io.StringIO() _thread.start_new_thread(fn1, ()) _thread.start_new_thread(fn2, ()) self.busywait(0.2) # time for the threads to finish assert sys.stderr.getvalue() == '' sys.stderr = io.StringIO() _thread.start_new_thread(fn3, ()) self.waitfor(lambda: "ValueError" in sys.stderr.getvalue()) result = sys.stderr.getvalue().splitlines() #assert result[0].startswith("Unhandled exception in thread ") assert result[1].startswith("Traceback ") assert result[-1] == "ValueError: hello world" finally: sys.stderr = prev def test_perthread_excinfo(self): import _thread, sys done = [] def fn1(n): success = False try: caught = False try: try: {}[n] except KeyError: self.busywait(0.05) caught = True raise except KeyError: self.busywait(0.05) assert caught etype, evalue, etb = sys.exc_info() assert etype is KeyError assert evalue.args[0] == n success = True finally: done.append(success) for i in range(20): _thread.start_new_thread(fn1, (i,)) self.waitfor(lambda: len(done) == 20) assert done == 20*[True] # see stderr for failures in the threads def test_no_corruption(self): import _thread lst = [] done_marker = [] def f(x, done): for j in range(40): lst.insert(0, x+j) # all threads trying to modify the same list done.append(True) for i in range(0, 120, 40): done = [] _thread.start_new_thread(f, (i, done)) done_marker.append(done) for done in done_marker: self.waitfor(lambda: done, delay=3) assert done # see stderr for failures in threads assert sorted(lst) == list(range(120)) def test_stack_size(self): import _thread _thread.stack_size(0) res = _thread.stack_size(0) assert res == 0 res = _thread.stack_size(1024*1024) assert res == 0 res = _thread.stack_size(2*1024*1024) assert res == 1024*1024 res = _thread.stack_size(0) assert res == 2*1024*1024 def test_interrupt_main(self): import _thread, time import signal def f(): for x in range(40): if waiting: _thread.interrupt_main() return time.sleep(0.1) def busy_wait(): waiting.append(None) for x in range(50): time.sleep(0.1) waiting.pop() # This is normally called by app_main.py signal.signal(signal.SIGINT, signal.default_int_handler) for i in range(100): print() print("loop", i) waiting = [] _thread.start_new_thread(f, ()) raises(KeyboardInterrupt, busy_wait) def test_interrupt_main_wrong_arg(self): import _thread with raises(ValueError): _thread.interrupt_main(-1) def test_interrupt_non_main(self): import _thread as thread, time, sys if sys.platform == "win32": skip("hangs on windows") import __pypy__ import signal def f(): try: childstarted.append(thread.get_ident()) self.waitfor(lambda: False) except ValueError: childstarted.pop() else: thread.interrupt_main() # to crash the test for i in range(20): print(i) # first wait for the child to start childstarted = [] thread.start_new_thread(f, ()) self.waitfor(lambda: bool(childstarted)) # then interrupt it from the main thread __pypy__.thread._raise_in_thread(childstarted[0], ValueError) # then wait for the exception to arrive in the child thread # which empties childstarted again def wait(): return not childstarted self.waitfor(wait) def test_interrupt_self(self): import _thread as thread, __pypy__ ident = thread.get_ident() for i in range(20): print(i) with raises(ValueError): __pypy__.thread._raise_in_thread(ident, ValueError) self.waitfor(lambda self: False) @pytest.mark.skip("too slow") class _AppTestThread(GenericTestThread): ''' This test is very slow, do not run it by default. ''' def setup_class(cls): GenericTestThread.setup_class.im_func(cls) # if we cannot start more than, say, 1000 threads on this OS, then # we can check that we get the proper error at app-level space = cls.space lock = thread.allocate_lock() lock.acquire() def f(): lock.acquire() lock.release() start = thread._count() try: try: for i in range(1000): thread.start_new_thread(f, ()) finally: lock.release() except (thread.error, MemoryError): cls.w_can_start_many_threads = space.wrap(False) else: cls.w_can_start_many_threads = space.wrap(True) # wait a bit to allow all threads to finish now remaining = thread._count() retries = 0 while remaining > start: retries += 1 if retries == 200: raise Exception("the test's threads don't stop!") time.sleep(0.2) remaining = thread._count() def test_many_threads(self): import time, sys if sys.version_info[0] < 3: import thread as _thread else: import _thread if self.can_start_many_threads or sys.platform == 'win32': skip("this OS supports too many threads to check (> 1000)") lock = _thread.allocate_lock() lock.acquire() count = [0] def f(): count[0] += 1 lock.acquire() lock.release() count[0] -= 1 try: try: for i in range(1000): _thread.start_new_thread(f, ()) finally: lock.release() # wait a bit to allow most threads to finish now while count[0] > 10: print(count[0]) # <- releases the GIL print("ok.") except (_thread.error, MemoryError): pass else: raise Exception("could unexpectedly start 1000 threads") # safety: check that we can start a new thread here _thread.start_new_thread(lambda: None, ())