1import threading 2import unittest 3 4from test import support 5from test.support import import_helper 6from test.support import threading_helper 7# Raise SkipTest if subinterpreters not supported. 8import_helper.import_module('_interpreters') 9from test.support import interpreters 10from .utils import TestBase 11 12 13class StressTests(TestBase): 14 15 # In these tests we generally want a lot of interpreters, 16 # but not so many that any test takes too long. 17 18 @support.requires_resource('cpu') 19 def test_create_many_sequential(self): 20 alive = [] 21 for _ in range(100): 22 interp = interpreters.create() 23 alive.append(interp) 24 25 @support.requires_resource('cpu') 26 @threading_helper.requires_working_threading() 27 def test_create_many_threaded(self): 28 alive = [] 29 def task(): 30 interp = interpreters.create() 31 alive.append(interp) 32 threads = (threading.Thread(target=task) for _ in range(200)) 33 with threading_helper.start_threads(threads): 34 pass 35 36 @support.requires_resource('cpu') 37 @threading_helper.requires_working_threading() 38 def test_many_threads_running_interp_in_other_interp(self): 39 interp = interpreters.create() 40 41 script = f"""if True: 42 import _interpreters 43 _interpreters.run_string({interp.id}, '1') 44 """ 45 46 def run(): 47 interp = interpreters.create() 48 alreadyrunning = (f'{interpreters.InterpreterError}: ' 49 'interpreter already running') 50 success = False 51 while not success: 52 try: 53 interp.exec(script) 54 except interpreters.ExecutionFailed as exc: 55 if exc.excinfo.msg != 'interpreter already running': 56 raise # re-raise 57 assert exc.excinfo.type.__name__ == 'InterpreterError' 58 else: 59 success = True 60 61 threads = (threading.Thread(target=run) for _ in range(200)) 62 with threading_helper.start_threads(threads): 63 pass 64 65 66if __name__ == '__main__': 67 # Test needs to be a package, so we can do relative imports. 68 unittest.main() 69