Enhance the test case as test_itertools.py

Author: hyongtao-code

Lib/test/test_free_threading/test_set.py

@@ -148,102 +148,60 @@ def read_set():
         for t in threads:
             t.join()
 
+    @threading_helper.reap_threads
     def test_length_hint_used_race(self):
-        s = set(range(2000))
-        it = iter(s)
-
-        NUM_LOOPS = 50_000
-        barrier = Barrier(2)
+        NUM_ITERS = 10
+        NUM_THREADS = 10
+        NUM_LOOPS = 2_000
 
-        def reader():
-            barrier.wait()
-            for _ in range(NUM_LOOPS):
-                it.__length_hint__()
+        for _ in range(NUM_ITERS):
+            s = set(range(2000))
+            it = iter(s)
 
-        def writer():
-            barrier.wait()
-            i = 0
-            for _ in range(NUM_LOOPS):
-                s.add(i)
-                s.discard(i - 1)
-                i += 1
+            def worker():
+                for i in range(NUM_LOOPS):
+                    it.__length_hint__()
+                    s.add(i)
+                    s.discard(i - 1)
 
-        threads = [
-            Thread(target=reader),
-            Thread(target=writer),
-        ]
-        for t in threads:
-            t.start()
-        for t in threads:
-            t.join()
+            threading_helper.run_concurrently(worker, nthreads=NUM_THREADS)
 
+    @threading_helper.reap_threads
     def test_length_hint_exhaust_race(self):
-        NUM_LOOPS = 10_000
-        INNER_HINTS = 20
-        barrier = Barrier(2)
-        box = {"it": None}
+        NUM_ITERS = 50
+        NUM_THREADS = 10
 
-        def exhauster():
-            for _ in range(NUM_LOOPS):
-                s = set(range(256))
-                box["it"] = iter(s)
-                barrier.wait()          # start together
-                try:
-                    while True:
-                        next(box["it"])
-                except StopIteration:
-                    pass
-                barrier.wait()          # end iteration
-
-        def reader():
-            for _ in range(NUM_LOOPS):
-                barrier.wait()
-                it = box["it"]
-                for _ in range(INNER_HINTS):
+        for _ in range(NUM_ITERS):
+            s = set(range(256))
+            it = iter(s)
+
+            def worker():
+                while True:
                     it.__length_hint__()
-                barrier.wait()
+                    try:
+                        next(it)
+                    except StopIteration:
+                        break
 
-        threads = [
-            Thread(target=reader),
-            Thread(target=exhauster),
-        ]
-        for t in threads:
-            t.start()
-        for t in threads:
-            t.join()
+            threading_helper.run_concurrently(worker, nthreads=NUM_THREADS)
 
+    @threading_helper.reap_threads
     def test_iternext_concurrent_exhaust_race(self):
-        NUM_LOOPS = 20_000
-        barrier = Barrier(3)
-        box = {"it": None}
+        NUM_ITERS = 200
+        NUM_THREADS = 10
 
-        def advancer():
-            for _ in range(NUM_LOOPS):
-                barrier.wait()
-                it = box["it"]
+        for _ in range(NUM_ITERS):
+            s = set(range(64))
+            it = iter(s)
+
+            def worker():
                 while True:
                     try:
                         next(it)
                     except StopIteration:
                         break
-                barrier.wait()
 
-        def producer():
-            for _ in range(NUM_LOOPS):
-                s = set(range(64))
-                box["it"] = iter(s)
-                barrier.wait()
-                barrier.wait()
-
-        threads = [
-            Thread(target=advancer),
-            Thread(target=advancer),
-            Thread(target=producer),
-        ]
-        for t in threads:
-            t.start()
-        for t in threads:
-            t.join()
+            threading_helper.run_concurrently(worker, nthreads=NUM_THREADS)
 
 
 @threading_helper.requires_working_threading()