gh-161204: Allow '-jN%' form of regrtest '-j' option

Adaptively target a given system load percentage.

Author: zware

Lib/test/libregrtest/adaptive_workers.py

@@ -0,0 +1,65 @@
+"""Load-average-dependent worker admission using the existing load tracker."""
+
+import math
+import os
+import threading
+
+
+class AdaptiveWorkerGate:
+    # Heuristic: avoid rapid adjustments against a smoothed load measurement.
+    update_interval = 5.0
+    # Heuristic, not empirically tuned: hold concurrency within 3% of the
+    # usable CPU count on either side of the target to reduce oscillation.
+    target_tolerance = 0.03
+
+    def __init__(self, target, maximum, logger):
+        self.target = target
+        self.maximum = maximum
+        self.get_load_avg = logger.get_load_avg
+        self.cpu_count = os.process_cpu_count() or 1
+        # Load is system-wide, so this only approximates affinity restrictions.
+        self.load_target = target * self.cpu_count
+        self.initial_limit = min(maximum, max(1, math.ceil(self.load_target)))
+        self.limit = self.initial_limit
+        self.active = 0
+        self.stopped = False
+        self.condition = threading.Condition()
+
+    def acquire(self):
+        with self.condition:
+            self.condition.wait_for(lambda: self.stopped or self.active < self.limit)
+            if self.stopped:
+                return False
+            self.active += 1
+            return True
+
+    def release(self):
+        with self.condition:
+            self.active -= 1
+            self.condition.notify_all()
+
+    def get_active_workers(self):
+        with self.condition:
+            return self.active
+
+    def update(self):
+        try:
+            load = self.get_load_avg()
+        except (OSError, ValueError):
+            load = None
+        with self.condition:
+            if load is None or not math.isfinite(load) or load < 0:
+                self.limit = self.initial_limit
+                self.condition.notify_all()
+                return
+            tolerance = self.target_tolerance * self.cpu_count
+            if load < self.load_target - tolerance:
+                self.limit = min(self.maximum, self.limit + 1)
+            elif load > self.load_target + tolerance:
+                self.limit = max(1, self.limit - 1)
+            self.condition.notify_all()
+
+    def stop(self):
+        with self.condition:
+            self.stopped = True
+            self.condition.notify_all()

Lib/test/libregrtest/cmdline.py

@@ -6,6 +6,19 @@
 from .utils import ALL_RESOURCES, RESOURCE_NAMES, TestFilter
 
 
+def worker_count(value: str) -> int | float:
+    if value.endswith('%'):
+        percentage = value[:-1]
+        if (not percentage.isascii() or not percentage.isdecimal()
+                or not 1 <= int(percentage) <= 100):
+            raise argparse.ArgumentTypeError('expected a percentage from 1% to 100%')
+        return int(percentage) / 100
+    try:
+        return int(value)
+    except ValueError:
+        raise argparse.ArgumentTypeError(f'invalid int value: {value!r}') from None
+
+
 USAGE = """\
 python -m test [options] [test_name1 [test_name2 ...]]
 python path/to/Lib/test/regrtest.py [options] [test_name1 [test_name2 ...]]
@@ -27,6 +40,14 @@
 EPILOG = """\
 Additional option details:
 
+`-jN%` adjusts the running worker count adaptively to approximate N% total
+system load.  At least one worker is always allowed to run, up to
+`os.process_cpu_count() * 2 + 4` (twice the limit used by `-j0`).  Running
+tests are never suspended, so the target can remain exceeded for an extended
+period of time.  At startup, `ceil(os.process_cpu_count() * (N/100))` workers
+are allowed, and failure to track load may result in reverting to this worker
+count.  Adaptive adjustments resume when load tracking recovers.
+
 `-r` randomizes test execution order. You can use `--randseed=int` to provide an
 int seed value for the randomizer. The `randseed` value will be used
 to set seeds for all random usages in tests
@@ -174,7 +195,7 @@ def __init__(self, **kwargs) -> None:
         self.verbose3 = False
         self.print_slow = False
         self.random_seed = None
-        self.use_mp = None
+        self.use_mp: int | float | None = None
         self.parallel_threads = None
         self.forever = False
         self.header = False
@@ -333,8 +354,10 @@ def _create_parser():
                        help='search for reference leaks (needs debug build, '
                             'very slow).' + more_details)
     group.add_argument('-j', '--multiprocess', metavar='PROCESSES',
-                       dest='use_mp', type=int,
-                       help='run `PROCESSES` processes at once')
+                       dest='use_mp', type=worker_count,
+                       help='run `PROCESSES` processes at once, or pass a '
+                            'percentage to adaptively target a level of load.'
+                            + more_details)
     group.add_argument('--single-process', action='store_true',
                        dest='single_process',
                        help='always run all tests sequentially in '

Lib/test/libregrtest/logger.py

@@ -21,10 +21,14 @@ def __init__(self, results: TestResults, quiet: bool, pgo: bool):
         self._quiet: bool = quiet
         self._pgo: bool = pgo
         self.get_mem_usage: Callable[[], int | None] | None = None
+        self.get_active_workers: Callable[[], int] | None = None
 
     def log(self, line: str = '') -> None:
         empty = not line
 
+        if self.get_active_workers is not None:
+            line = f'running: {self.get_active_workers()} {line}'
+
         # Add the memory usage: "mem: 1 GiB "
         if self.get_mem_usage is not None:
             mem = self.get_mem_usage()

Lib/test/libregrtest/main.py

@@ -95,12 +95,13 @@ def __init__(self, ns: Namespace, _add_python_opts: bool = False):
 
         # Workers
         self.single_process: bool = ns.single_process
+        self.adaptive_target = ns.use_mp if isinstance(ns.use_mp, float) else None
         if self.single_process or ns.use_mp is None:
             num_workers = 0   # run sequentially in a single process
-        elif ns.use_mp <= 0:
+        elif self.adaptive_target or ns.use_mp <= 0:
             num_workers = -1  # run in parallel, use the number of CPUs
         else:
-            num_workers = ns.use_mp  # run in parallel
+            num_workers = int(ns.use_mp)  # run in parallel
         self.num_workers: int = num_workers
         if ns.single_process_per_case and ns.use_mp is None:
             # Each test case runs in its own worker subprocess;
@@ -465,7 +466,7 @@ def get_state(self) -> str:
     def _run_tests_mp(self, runtests: RunTests, num_workers: int) -> None:
         from .run_workers import RunWorkers
         RunWorkers(num_workers, runtests, self.logger, self.results,
-                   self.reporter).run()
+                   self.reporter, adaptive_target=self.adaptive_target).run()
 
     def finalize_tests(self, coverage: trace.CoverageResults | None) -> None:
         if self.next_single_filename:
@@ -548,6 +549,9 @@ def _run_tests(self, selected: TestTuple, tests: TestList | None) -> int:
             # Use all CPUs + 2 extra worker processes for tests
             # that like to sleep
             self.num_workers = (os.process_cpu_count() or 1) + 2
+            if self.adaptive_target is not None:
+                # Allow more sleeping or I/O-bound tests in adaptive mode.
+                self.num_workers *= 2
 
         # For a partial run, we do not need to clutter the output.
         if (self.want_header

Lib/test/libregrtest/run_workers.py

@@ -111,6 +111,7 @@ def __init__(self, worker_id: int, runner: "RunWorkers") -> None:
         self.worker_id = worker_id
         self.runtests = runner.runtests
         self.pending = runner.pending
+        self.worker_gate = runner.worker_gate
         self.output = runner.output
         self.timeout = runner.worker_timeout
         self.log = runner.log
@@ -415,6 +416,8 @@ def run(self) -> None:
                 for test_name in case_ids:
                     if self._stopped:
                         break
+                    if self.worker_gate is not None and not self.worker_gate.acquire():
+                        raise ExitThread
                     self.start_time = time.monotonic()
                     self.test_name = test_name
                     try:
@@ -425,6 +428,8 @@ def run(self) -> None:
                         mp_result = exc.mp_result
                     finally:
                         self.test_name = _NOT_RUNNING
+                        if self.worker_gate is not None:
+                            self.worker_gate.release()
                     mp_result.result.duration = time.monotonic() - self.start_time
                     if single_process_per_case:
                         # Report the test case, not the test module
@@ -501,7 +506,7 @@ def get_running(workers: list[WorkerThread]) -> str | None:
 class RunWorkers:
     def __init__(self, num_workers: int, runtests: RunTests,
                  logger: Logger, results: TestResults,
-                 reporter: Reporter) -> None:
+                 reporter: Reporter, adaptive_target: float | None = None) -> None:
         self.num_workers = num_workers
         self.runtests = runtests
         self.logger = logger
@@ -529,6 +534,13 @@ def __init__(self, num_workers: int, runtests: RunTests,
             # these worker threads would never get anything to do.
             self.num_workers = min(self.num_workers, jobs)
 
+        self.worker_gate = None
+        if adaptive_target is not None:
+            from .adaptive_workers import AdaptiveWorkerGate
+            self.worker_gate = AdaptiveWorkerGate(
+                adaptive_target, max(1, self.num_workers), logger)
+            self.next_worker_update = time.monotonic() + self.worker_gate.update_interval
+
     def start_workers(self) -> None:
         self.workers = [WorkerThread(index, self)
                         for index in range(1, self.num_workers + 1)]
@@ -545,13 +557,18 @@ def start_workers(self) -> None:
             msg += (" (timeout: %s, worker timeout: %s)"
                     % (format_duration(self.timeout),
                        format_duration(self.worker_timeout)))
+        if self.worker_gate is not None:
+            msg += (f' (Load target: {self.worker_gate.target:.0%}, adaptive workers; '
+                    f'load average {self.worker_gate.load_target:.2f})')
         self.log(msg)
         for worker in self.workers:
             worker.start()
             self.live_worker_count += 1
 
     def stop_workers(self) -> None:
         start_time = time.monotonic()
+        if self.worker_gate is not None:
+            self.worker_gate.stop()
         for worker in self.workers:
             worker.stop()
         for worker in self.workers:
@@ -568,9 +585,17 @@ def _get_result(self) -> QueueOutput | None:
                 faulthandler.dump_traceback_later(MAIN_PROCESS_TIMEOUT,
                                                   exit=True)
 
+            timeout = PROGRESS_UPDATE
+            if self.worker_gate is not None:
+                now = time.monotonic()
+                if now >= self.next_worker_update:
+                    self.worker_gate.update()
+                    self.next_worker_update = now + self.worker_gate.update_interval
+                timeout = min(timeout, self.next_worker_update - now)
+
             # wait for a thread
             try:
-                result = self.output.get(timeout=PROGRESS_UPDATE)
+                result = self.output.get(timeout=timeout)
                 if isinstance(result, WorkerThreadExited):
                     self.live_worker_count -= 1
                     continue
@@ -647,6 +672,8 @@ def run(self) -> None:
         fail_fast = self.runtests.fail_fast
         fail_env_changed = self.runtests.fail_env_changed
         self.logger.get_mem_usage = self.get_mem_usage
+        if self.worker_gate is not None:
+            self.logger.get_active_workers = self.worker_gate.get_active_workers
 
         self.start_workers()
 
@@ -672,3 +699,4 @@ def run(self) -> None:
             self.pending.stop()
             self.stop_workers()
             self.logger.get_mem_usage = None
+            self.logger.get_active_workers = None

Lib/test/test_regrtest.py

@@ -52,6 +52,7 @@
     r'[0-9]+:[0-9]+:[0-9]+ '
     r'(?:load avg: [0-9]+\.[0-9]{2} )?'
     r'(?:mem: +[0-9]+\.[0-9] (?:MiB|GiB) )?'
+    r'(?:running: [0-9]+ )?'
 )
 RESULT_REGEX = (
     'passed',
@@ -380,9 +381,50 @@ def test_multiprocess(self):
             with self.subTest(opt=opt):
                 ns = self.parse_args([opt, '2'])
                 self.assertEqual(ns.use_mp, 2)
+                self.assertIsInstance(ns.use_mp, int)
                 self.checkError([opt], 'expected one argument')
                 self.checkError([opt, 'foo'], 'invalid int value')
 
+    def test_percentage_jobs(self):
+        for args in (['-j90%'], ['--multiprocess=90%'], ['-j', '90%']):
+            with self.subTest(args=args):
+                ns = self.parse_args(args)
+                self.assertEqual(ns.use_mp, 0.9)
+                self.assertIsInstance(ns.use_mp, float)
+                regrtest = main.Regrtest(ns)
+                self.assertEqual(regrtest.adaptive_target, 0.9)
+                self.assertEqual(regrtest.num_workers, -1)
+        for value, expected in (('1%', 0.01), ('100%', 1.0)):
+            ns = self.parse_args(['-j', value])
+            self.assertEqual(ns.use_mp, expected)
+            self.assertIsInstance(ns.use_mp, float)
+        for value in ('0%', '101%', '-1%', '1.5%', 'nan%', '%'):
+            with self.subTest(value=value):
+                self.checkError([f'-j{value}'], 'expected a percentage')
+
+    def test_percentage_jobs_without_load_average(self):
+        with unittest.mock.patch.object(os, 'getloadavg', create=True):
+            del os.getloadavg
+            self.assertEqual(self.parse_args(['-j90%']).use_mp, 0.9)
+            ns = self.parse_args(['-j90%', '--single-process'])
+            self.assertIsNone(ns.use_mp)
+
+    def test_percentage_jobs_worker_ceiling(self):
+        class StopRun(Exception):
+            pass
+
+        for cpu_count in (1, 16, None):
+            automatic = (cpu_count or 1) + 2
+            for option, expected in (('-j90%', 2 * automatic),
+                                     ('-j0', automatic), ('-j4', 4)):
+                with self.subTest(cpu_count=cpu_count, option=option):
+                    regrtest = self.create_regrtest([option, '--header'])
+                    with (unittest.mock.patch('os.process_cpu_count', return_value=cpu_count),
+                          unittest.mock.patch.object(main, 'display_header', side_effect=StopRun)):
+                        with self.assertRaises(StopRun):
+                            regrtest._run_tests((), [])
+                    self.assertEqual(regrtest.num_workers, expected)
+
     def test_coverage_sequential(self):
         for opt in '-T', '--coverage':
             with self.subTest(opt=opt):
@@ -1072,6 +1114,13 @@ def test_test3(self):
         self.check_executed_tests(output, tests,
                                   stats=3 * len(tests))
 
+    def test_percentage_jobs(self):
+        tests = [self.create_test(f'cpu{i}') for i in range(3)]
+        output = self.run_tests('-j90%', *tests)
+        self.check_executed_tests(output, tests, stats=len(tests), parallel=True)
+        self.assertIn('Load target: 90%, adaptive workers', output)
+        self.assertRegex(output, r'running: \d+ \[\s*\d+/3\]')
+
     def test_skip(self):
         code = textwrap.dedent("""
             import unittest