diff --git a/Lib/test/libregrtest/adaptive_workers.py b/Lib/test/libregrtest/adaptive_workers.py new file mode 100644 index 00000000000000..64c5e8d545adff --- /dev/null +++ b/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() diff --git a/Lib/test/libregrtest/cmdline.py b/Lib/test/libregrtest/cmdline.py index 720b073e460127..b7abf503df0ad9 100644 --- a/Lib/test/libregrtest/cmdline.py +++ b/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 ' diff --git a/Lib/test/libregrtest/logger.py b/Lib/test/libregrtest/logger.py index 7c1d9e528010c3..f685e96ce60bfb 100644 --- a/Lib/test/libregrtest/logger.py +++ b/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() diff --git a/Lib/test/libregrtest/main.py b/Lib/test/libregrtest/main.py index 032cb94223a7e6..d5066c3d35452d 100644 --- a/Lib/test/libregrtest/main.py +++ b/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 diff --git a/Lib/test/libregrtest/run_workers.py b/Lib/test/libregrtest/run_workers.py index ad02d8dba7c23e..bc5d3132917848 100644 --- a/Lib/test/libregrtest/run_workers.py +++ b/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,6 +557,9 @@ 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() @@ -552,6 +567,8 @@ def start_workers(self) -> None: 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 diff --git a/Lib/test/test_regrtest.py b/Lib/test/test_regrtest.py index e8211526ae847c..29a844dc10a862 100644 --- a/Lib/test/test_regrtest.py +++ b/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.dict(os.__dict__): + os.__dict__.pop('getloadavg', None) + 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 diff --git a/Lib/test/test_regrtest_adaptive_workers.py b/Lib/test/test_regrtest_adaptive_workers.py new file mode 100644 index 00000000000000..4a2ccf9af2bd07 --- /dev/null +++ b/Lib/test/test_regrtest_adaptive_workers.py @@ -0,0 +1,275 @@ +"""Deterministic tests for percentage-job admission and load averages.""" + +import contextlib +import io +import queue +import threading +import unittest +from unittest.mock import Mock, patch + +from test.support import threading_helper +from test.libregrtest.adaptive_workers import AdaptiveWorkerGate +from test.libregrtest.logger import Logger +from test.libregrtest.results import TestResults +from test.libregrtest.run_workers import RunWorkers + + +class AdaptiveWorkerGateTests(unittest.TestCase): + def make_gate(self): + logger = Mock() + logger.get_load_avg.return_value = 0 + with patch('os.process_cpu_count', return_value=4): + gate = AdaptiveWorkerGate(0.25, 3, logger) + logger.get_load_avg.reset_mock() + return gate, logger.get_load_avg + + def update(self, gate, get_load_avg, value): + get_load_avg.return_value = value * 4 + gate.update() + + def test_adjustment(self): + gate, get_load_avg = self.make_gate() + for expected in (2, 3, 3): + self.update(gate, get_load_avg, 0.2) + self.assertEqual(gate.limit, expected) + for value in (0.24, 0.26): + self.update(gate, get_load_avg, value) + self.assertEqual(gate.limit, 3) + for expected in (2, 1, 1): + self.update(gate, get_load_avg, 1.0) + self.assertEqual(gate.limit, expected) + + def test_each_update_samples_load(self): + gate, get_load_avg = self.make_gate() + gate.update() + self.assertEqual(gate.limit, 2) + gate.update() + self.assertEqual(gate.limit, 3) + self.assertEqual(get_load_avg.call_count, 2) + + def test_custom_target_tolerance(self): + gate, get_load_avg = self.make_gate() + gate.target_tolerance = 0.125 + gate.limit = 2 + for value in (gate.target - gate.target_tolerance, + gate.target, gate.target + gate.target_tolerance): + self.update(gate, get_load_avg, value) + self.assertEqual(gate.limit, 2) + self.update(gate, get_load_avg, 0.0625) + self.assertEqual(gate.limit, 3) + self.update(gate, get_load_avg, 0.4375) + self.assertEqual(gate.limit, 2) + + def test_active_workers(self): + gate, _ = self.make_gate() + self.assertEqual(gate.get_active_workers(), 0) + self.assertTrue(gate.acquire()) + self.assertEqual(gate.get_active_workers(), 1) + gate.release() + self.assertEqual(gate.get_active_workers(), 0) + + @threading_helper.requires_working_threading() + def test_waiter_shutdown(self): + gate, _ = self.make_gate() + self.assertTrue(gate.acquire()) + acquired = [] + thread = threading.Thread(target=lambda: acquired.append(gate.acquire())) + thread.start() + try: + gate.stop() + thread.join(5) + self.assertFalse(thread.is_alive()) + self.assertEqual(acquired, [False]) + finally: + gate.stop() + thread.join(5) + gate.release() + + @threading_helper.requires_working_threading() + def test_release_wakes_waiter(self): + gate, _ = self.make_gate() + self.assertTrue(gate.acquire()) + acquired = threading.Event() + + def worker(): + if gate.acquire(): + acquired.set() + gate.release() + + thread = threading.Thread(target=worker) + thread.start() + try: + self.assertFalse(acquired.is_set()) + gate.release() + self.assertTrue(acquired.wait(5)) + finally: + gate.stop() + thread.join(5) + self.assertEqual(gate.get_active_workers(), 0) + + def test_lower_limit_does_not_interrupt_active_workers(self): + gate, get_load_avg = self.make_gate() + gate.limit = 2 + self.assertTrue(gate.acquire()) + self.assertTrue(gate.acquire()) + self.update(gate, get_load_avg, 1.0) + self.assertEqual(gate.limit, 1) + self.assertEqual(gate.get_active_workers(), 2) + gate.release() + gate.release() + + @threading_helper.requires_working_threading() + def test_increase_wakes_waiter(self): + gate, get_load_avg = self.make_gate() + self.assertTrue(gate.acquire()) + acquired = threading.Event() + + def worker(): + if gate.acquire(): + acquired.set() + gate.release() + + thread = threading.Thread(target=worker) + thread.start() + try: + self.update(gate, get_load_avg, 0.2) + self.assertTrue(acquired.wait(5)) + finally: + gate.stop() + thread.join(5) + gate.release() + + + +class WorkerUpdateIntervalTests(unittest.TestCase): + def make_runner(self): + gate = Mock(update_interval=5.0) + runtests = Mock(timeout=None, pgo=True) + runtests.get_jobs.return_value = 2 + runtests.iter_case_groups.return_value = iter(()) + with (patch('test.libregrtest.adaptive_workers.AdaptiveWorkerGate', return_value=gate), + patch('test.libregrtest.run_workers.time.monotonic', return_value=0)): + runner = RunWorkers(2, runtests, Mock(), Mock(), Mock(), adaptive_target=0.8) + runner.live_worker_count = 1 + runner.output = Mock() + return runner, gate + + def test_frequent_results_preserve_deadline(self): + runner, gate = self.make_runner() + result = (False, Mock()) + runner.output.get.return_value = result + for now, expected_updates in ((0, 0), (1, 0), (4.9, 0), + (5, 1), (5.1, 1), (9.9, 1), (10, 2)): + with self.subTest(now=now): + with patch('test.libregrtest.run_workers.time.monotonic', return_value=now): + self.assertIs(runner._get_result(), result) + self.assertEqual(gate.update.call_count, expected_updates) + + def test_idle_wait_ends_at_update_deadline(self): + runner, gate = self.make_runner() + now = 4.0 + result = (False, Mock()) + + def get_result(*, timeout): + nonlocal now + if now == 4.0: + self.assertEqual(timeout, 1.0) + now += timeout + raise queue.Empty + self.assertEqual(timeout, 5.0) + return result + + runner.output.get.side_effect = get_result + with patch('test.libregrtest.run_workers.time.monotonic', side_effect=lambda: now): + self.assertIs(runner._get_result(), result) + gate.update.assert_called_once_with() + + def test_late_update_does_not_catch_up_in_a_burst(self): + runner, gate = self.make_runner() + runner.output.get.return_value = (False, Mock()) + with patch('test.libregrtest.run_workers.time.monotonic', return_value=30): + runner._get_result() + runner._get_result() + gate.update.assert_called_once_with() + self.assertEqual(runner.next_worker_update, 35) + + +class AdaptiveLoggerTests(unittest.TestCase): + def test_active_workers_output(self): + logger = Logger(TestResults(), quiet=False, pgo=False) + with patch.object(logger, 'get_load_avg', return_value=None): + for active in (0, 2): + with self.subTest(active=active): + logger.get_active_workers = lambda: active + with contextlib.redirect_stdout(io.StringIO()) as output: + logger.log('test_example passed') + self.assertIn(f'running: {active} test_example passed', + output.getvalue()) + logger.get_active_workers = None + with contextlib.redirect_stdout(io.StringIO()) as output: + logger.log('test_example passed') + self.assertNotIn('running:', output.getvalue()) + + def test_memory_output(self): + logger = Logger(TestResults(), quiet=False, pgo=False) + with patch.object(logger, 'get_load_avg', return_value=None): + for usage, expected in ((16 * 1024**2, 'mem: 16.0 MiB'), + (2 * 1024**3, 'mem: 2.0 GiB')): + with self.subTest(usage=usage): + logger.get_mem_usage = lambda: usage + with contextlib.redirect_stdout(io.StringIO()) as output: + logger.log('test_example passed') + self.assertIn(expected, output.getvalue()) + + +class LoadAverageTests(unittest.TestCase): + def logger(self, values): + logger = Mock() + logger.get_load_avg.side_effect = values + return logger + + def test_process_cpu_count_sets_target(self): + with (patch('os.cpu_count', return_value=16), + patch('os.process_cpu_count', return_value=4)): + gate = AdaptiveWorkerGate(0.8, 36, self.logger([0])) + self.assertEqual(gate.load_target, 3.2) + self.assertEqual(gate.limit, 4) + + def test_initial_concurrency(self): + for cpus, target, maximum, expected in ( + (16, 0.8, 36, 13), (4, 0.5, 12, 2), + (1, 0.01, 6, 1), (16, 1.0, 3, 3), (None, 0.8, 6, 1)): + with self.subTest(cpus=cpus, target=target, maximum=maximum): + with patch('os.process_cpu_count', return_value=cpus): + gate = AdaptiveWorkerGate(target, maximum, self.logger([0])) + self.assertEqual(gate.limit, expected) + + def test_tracking_failure_reverts_to_initial_count(self): + for failure in (None, -1, float('nan'), float('inf'), + OSError('no load'), ValueError('bad load')): + for initial_load in (True, False): + with self.subTest(failure=failure, initial_load=initial_load): + values = [failure, 0] if initial_load else [0, failure, 0] + logger = self.logger(values) + with patch('os.process_cpu_count', return_value=4): + gate = AdaptiveWorkerGate(0.5, 8, logger) + if not initial_load: + gate.update() + self.assertEqual(gate.limit, 3) + gate.update() + self.assertEqual(gate.limit, 2) + calls = logger.get_load_avg.call_count + gate.update() + self.assertEqual(logger.get_load_avg.call_count, calls + 1) + self.assertEqual(gate.limit, 3) + + def test_unknown_cpu_count(self): + with patch('os.process_cpu_count', return_value=None): + gate = AdaptiveWorkerGate(0.8, 8, self.logger([0])) + self.assertEqual(gate.load_target, 0.8) + self.assertEqual(gate.limit, 1) + + + +if __name__ == '__main__': + unittest.main() diff --git a/Misc/NEWS.d/next/Tests/2026-10-23-10-00-00.gh-issue-161204.sKIjSG.rst b/Misc/NEWS.d/next/Tests/2026-10-23-10-00-00.gh-issue-161204.sKIjSG.rst new file mode 100644 index 00000000000000..b68ba14cd351df --- /dev/null +++ b/Misc/NEWS.d/next/Tests/2026-10-23-10-00-00.gh-issue-161204.sKIjSG.rst @@ -0,0 +1,2 @@ +Add a percentage form of regrtest's ``-j`` option to adjust worker concurrency +toward a system load-average target using the existing load tracker.