TST: add test for multithreaded reductions

numpy · seberg · Jan 9, 2025 · Jan 7, 2025 · Jan 8, 2025 · Jan 8, 2025
commit b0358e0fa728917fa3dca875b7aaca8163eeb61a
diff --git a/numpy/_core/tests/test_multithreading.py b/numpy/_core/tests/test_multithreading.py
@@ -120,3 +120,16 @@ def legacy_125():
 
     task1.start()
     task2.start()
+
+def test_parallel_reduction():
+    # gh-28041
+    NUM_THREADS = 50
+
+    b = threading.Barrier(NUM_THREADS)
+
+    x = np.arange(1000)
+    def closure():
+        b.wait()
+        np.sum(x)
+
+    run_threaded(closure, NUM_THREADS, max_workers=NUM_THREADS)
diff --git a/numpy/testing/_private/utils.py b/numpy/testing/_private/utils.py
@@ -2685,9 +2685,9 @@ def _get_glibc_version():
 _glibc_older_than = lambda x: (_glibcver != '0.0' and _glibcver < x)
 
 
-def run_threaded(func, iters, pass_count=False):
+def run_threaded(func, iters, pass_count=False, max_workers=8):
     """Runs a function many times in parallel"""
-    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as tpe:
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as tpe:
         if pass_count:
             futures = [tpe.submit(func, i) for i in range(iters)]
         else: