mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-31 05:58:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

gh-109700: Improve stress tests for interpreter creation (GH-109946)

Browse source 
* Ensure that destructors are called in the test that created interpreters, not after finishing it.
* Try to create/run interpreters in threads simultaneously.
* Mark tests that requires over 6GB of memory with bigmemtest.
This commit is contained in:
Serhiy Storchaka 2025-05-04 21:53:24 +03:00 • committed by GitHub
parent 95d2a81ba8
commit 61b50a98b4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 21 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

30
Lib/test/test_interpreters/test_stress.py
View file

@ -21,21 +21,29 @@ class StressTests(TestBase):
for _ in range(100): for _ in range(100):
interp = interpreters.create() interp = interpreters.create()
alive.append(interp) alive.append(interp)
del alive
support.gc_collect()
@support.requires_resource('cpu') @support.bigmemtest(size=200, memuse=32*2**20, dry_run=False)
@threading_helper.requires_working_threading() def test_create_many_threaded(self, size):
def test_create_many_threaded(self):
alive = [] alive = []
start = threading.Event()
def task(): def task():
# try to create all interpreters simultaneously
if not start.wait(support.SHORT_TIMEOUT):
raise TimeoutError
interp = interpreters.create() interp = interpreters.create()
alive.append(interp) alive.append(interp)
threads = (threading.Thread(target=task) for _ in range(200)) threads = [threading.Thread(target=task) for _ in range(size)]
with threading_helper.start_threads(threads): with threading_helper.start_threads(threads):
pass start.set()
del alive
support.gc_collect()
@support.requires_resource('cpu')
@threading_helper.requires_working_threading() @threading_helper.requires_working_threading()
def test_many_threads_running_interp_in_other_interp(self): @support.bigmemtest(size=200, memuse=34*2**20, dry_run=False)
def test_many_threads_running_interp_in_other_interp(self, size):
start = threading.Event()
interp = interpreters.create() interp = interpreters.create()
script = f"""if True: script = f"""if True:
@ -47,6 +55,9 @@ class StressTests(TestBase):
interp = interpreters.create() interp = interpreters.create()
alreadyrunning = (f'{interpreters.InterpreterError}: ' alreadyrunning = (f'{interpreters.InterpreterError}: '
'interpreter already running') 'interpreter already running')
# try to run all interpreters simultaneously
if not start.wait(support.SHORT_TIMEOUT):
raise TimeoutError
success = False success = False
while not success: while not success:
try: try:
@ -58,9 +69,10 @@ class StressTests(TestBase):
else: else:
success = True success = True
threads = (threading.Thread(target=run) for _ in range(200)) threads = [threading.Thread(target=run) for _ in range(size)]
with threading_helper.start_threads(threads): with threading_helper.start_threads(threads):
pass start.set()
support.gc_collect()
if __name__ == '__main__': if __name__ == '__main__':