parallel_testsuite: Switch to pool.imap_unordered. NFC (#25751)

This allows us the print that status line back on the main thread as the
results arrive there. It also avoids some of the shared state.

I have a followup change that uses just a single line for output line
other modern test runners do.

Author: sbc100

test/parallel_testsuite.py

@@ -44,7 +44,8 @@ def cap_max_workers_in_pool(max_workers, is_browser):
   return max_workers
 
 
-def run_test(test, allowed_failures_counter, lock, progress_counter, num_tests, buffer):
+def run_test(args):
+  test, allowed_failures_counter, lock, buffer = args
   # If we have exceeded the number of allowed failures during the test run,
   # abort executing further tests immediately.
   if allowed_failures_counter and allowed_failures_counter.value < 0:
@@ -57,38 +58,6 @@ def test_failed():
 
   start_time = time.perf_counter()
 
-  def compute_progress():
-    if not lock:
-      return ''
-    with lock:
-      val = f'[{int(progress_counter.value * 100 / num_tests)}%] '
-      progress_counter.value += 1
-    return with_color(CYAN, val)
-
-  def printResult(res):
-    elapsed = time.perf_counter() - start_time
-    progress = compute_progress()
-    if res.test_result == 'success':
-      msg = f'ok ({elapsed:.2f}s)'
-      errlog(f'{progress}{res.test} ... {with_color(GREEN, msg)}')
-    elif res.test_result == 'errored':
-      msg = f'{res.test} ... ERROR'
-      errlog(f'{progress}{with_color(RED, msg)}')
-    elif res.test_result == 'failed':
-      msg = f'{res.test} ... FAIL'
-      errlog(f'{progress}{with_color(RED, msg)}')
-    elif res.test_result == 'skipped':
-      msg = f"skipped '{res.buffered_result.reason}'"
-      errlog(f"{progress}{res.test} ... {with_color(CYAN, msg)}")
-    elif res.test_result == 'unexpected success':
-      msg = f'unexpected success ({elapsed:.2f}s)'
-      errlog(f'{progress}{res.test} ... {with_color(RED, msg)}')
-    elif res.test_result == 'expected failure':
-      msg = f'expected failure ({elapsed:.2f}s)'
-      errlog(f'{progress}{res.test} ... {with_color(RED, msg)}')
-    else:
-      assert False
-
   olddir = os.getcwd()
   result = BufferedParallelTestResult()
   result.start_time = start_time
@@ -110,7 +79,7 @@ def printResult(res):
     result.addError(test, e)
     test_failed()
   finally:
-    printResult(result)
+    result.elapsed = time.perf_counter() - start_time
 
   # Before attempting to delete the tmp dir make sure the current
   # working directory is not within it.
@@ -149,11 +118,44 @@ def __init__(self, max_cores, options):
     self.max_cores = max_cores
     self.max_failures = options.max_failures
     self.failing_and_slow_first = options.failing_and_slow_first
+    self.progress_counter = 0
 
   def addTest(self, test):
     super().addTest(test)
     test.is_parallel = True
 
+  def printOneResult(self, res):
+    percent = int(self.progress_counter * 100 / self.num_tests)
+    progress = f'[{percent:2d}%] '
+    self.progress_counter += 1
+
+    if res.test_result == 'success':
+      msg = 'ok'
+      color = GREEN
+    elif res.test_result == 'errored':
+      msg = 'ERROR'
+      color = RED
+    elif res.test_result == 'failed':
+      msg = 'FAIL'
+      color = RED
+    elif res.test_result == 'skipped':
+      reason = res.skipped[0][1]
+      msg = f"skipped '{reason}'"
+      color = CYAN
+    elif res.test_result == 'unexpected success':
+      msg = 'unexpected success'
+      color = RED
+    elif res.test_result == 'expected failure':
+      color = RED
+      msg = 'expected failure'
+    else:
+      assert False, f'unhandled test result {res.test_result}'
+
+    if res.test_result != 'skipped':
+      msg += f' ({res.elapsed:.2f}s)'
+
+    errlog(f'{with_color(CYAN, progress)}{res.test} ... {with_color(color, msg)}')
+
   def run(self, result):
     # The 'spawn' method is used on windows and it can be useful to set this on
     # all platforms when debugging multiprocessing issues.  Without this we
@@ -163,6 +165,7 @@ def run(self, result):
     # multiprocessing.set_start_method('spawn')
 
     tests = self.get_sorted_tests()
+    self.num_tests = len(tests)
     contains_browser_test = any(test.is_browser_test() for test in tests)
     use_cores = cap_max_workers_in_pool(min(self.max_cores, len(tests), num_cores()), contains_browser_test)
     errlog(f'Using {use_cores} parallel test processes')
@@ -178,15 +181,23 @@ def run(self, result):
         if python_multiprocessing_structures_are_buggy():
           # When multiprocessing shared structures are buggy we don't support failfast
           # or the progress bar.
-          allowed_failures_counter = progress_counter = lock = None
+          allowed_failures_counter = lock = None
           if self.max_failures < 2**31 - 1:
             errlog('The version of python being used is not compatible with --failfast and --max-failures options. See https://github.com/python/cpython/issues/71936')
             sys.exit(1)
         else:
           allowed_failures_counter = manager.Value('i', self.max_failures)
-          progress_counter = manager.Value('i', 0)
           lock = manager.Lock()
-        results = pool.starmap(run_test, ((t, allowed_failures_counter, lock, progress_counter, len(tests), result.buffer) for t in tests), chunksize=1)
+
+        results = []
+        args = ((t, allowed_failures_counter, lock, result.buffer) for t in tests)
+        for res in pool.imap_unordered(run_test, args, chunksize=1):
+          # results may be be None if # of allowed errors was exceeded
+          # and the harness aborted.
+          if res:
+            self.printOneResult(res)
+            results.append(res)
+
         # Send a task to each worker to tear down the browser and server. This
         # relies on the implementation detail in the worker pool that all workers
         # are cycled through once.
@@ -195,9 +206,6 @@ def run(self, result):
         if num_tear_downs != use_cores:
           errlog(f'Expected {use_cores} teardowns, got {num_tear_downs}')
 
-    # Filter out the None results which can occur if # of allowed errors was exceeded and the harness aborted.
-    results = [r for r in results if r is not None]
-
     if self.failing_and_slow_first:
       previous_test_run_results = common.load_previous_test_run_results()
       for r in results: