mirror of
				https://github.com/python/cpython.git
				synced 2025-11-04 07:31:38 +00:00 
			
		
		
		
	gh-124402: Speed up test_free_threading and test_super (#124491)
* Reduce the number of iterations and the number of threads so a
  whole test file takes less than a minute.
* Refactor test_racing_iter_extend() to remove two levels of
  indentation.
* test_monitoring() uses a sleep of 100 ms instead of 1 second.
(cherry picked from commit 0387c34f7c)
		
	
			
		
			
				
	
	
		
			76 lines
		
	
	
	
		
			1.8 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			76 lines
		
	
	
	
		
			1.8 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
import unittest
 | 
						|
 | 
						|
from threading import Thread
 | 
						|
from unittest import TestCase
 | 
						|
 | 
						|
from test.support import threading_helper
 | 
						|
 | 
						|
 | 
						|
NTHREAD = 10
 | 
						|
OBJECT_COUNT = 5_000
 | 
						|
 | 
						|
 | 
						|
class C:
 | 
						|
    def __init__(self, v):
 | 
						|
        self.v = v
 | 
						|
 | 
						|
 | 
						|
@threading_helper.requires_working_threading()
 | 
						|
class TestList(TestCase):
 | 
						|
    def test_racing_iter_append(self):
 | 
						|
        l = []
 | 
						|
 | 
						|
        def writer_func():
 | 
						|
            for i in range(OBJECT_COUNT):
 | 
						|
                l.append(C(i + OBJECT_COUNT))
 | 
						|
 | 
						|
        def reader_func():
 | 
						|
            while True:
 | 
						|
                count = len(l)
 | 
						|
                for i, x in enumerate(l):
 | 
						|
                    self.assertEqual(x.v, i + OBJECT_COUNT)
 | 
						|
                if count == OBJECT_COUNT:
 | 
						|
                    break
 | 
						|
 | 
						|
        writer = Thread(target=writer_func)
 | 
						|
        readers = []
 | 
						|
        for x in range(NTHREAD):
 | 
						|
            reader = Thread(target=reader_func)
 | 
						|
            readers.append(reader)
 | 
						|
            reader.start()
 | 
						|
 | 
						|
        writer.start()
 | 
						|
        writer.join()
 | 
						|
        for reader in readers:
 | 
						|
            reader.join()
 | 
						|
 | 
						|
    def test_racing_iter_extend(self):
 | 
						|
        l = []
 | 
						|
 | 
						|
        def writer_func():
 | 
						|
            for i in range(OBJECT_COUNT):
 | 
						|
                l.extend([C(i + OBJECT_COUNT)])
 | 
						|
 | 
						|
        def reader_func():
 | 
						|
            while True:
 | 
						|
                count = len(l)
 | 
						|
                for i, x in enumerate(l):
 | 
						|
                    self.assertEqual(x.v, i + OBJECT_COUNT)
 | 
						|
                if count == OBJECT_COUNT:
 | 
						|
                    break
 | 
						|
 | 
						|
        writer = Thread(target=writer_func)
 | 
						|
        readers = []
 | 
						|
        for x in range(NTHREAD):
 | 
						|
            reader = Thread(target=reader_func)
 | 
						|
            readers.append(reader)
 | 
						|
            reader.start()
 | 
						|
 | 
						|
        writer.start()
 | 
						|
        writer.join()
 | 
						|
        for reader in readers:
 | 
						|
            reader.join()
 | 
						|
 | 
						|
 | 
						|
if __name__ == "__main__":
 | 
						|
    unittest.main()
 |