From bd09a57bf18dc8b426cc8f8cdd3f873fc795c7c7 Mon Sep 17 00:00:00 2001 From: Bill Hlavacek Date: Fri, 21 Aug 2026 20:02:11 -0600 Subject: [PATCH] Warn when a fit uses far fewer processors than were reserved (#621) How many simulations a fit runs at once is decided by its settings, mainly population_size, and not by how many processors the user reserved. When the two do not match, the extra processors can sit idle for the whole run and nothing said so. A user could reserve several machines and quietly use a fraction of them. After the fit submits its first set of jobs, it now compares how many jobs are running with how many workers connected to the cluster. It logs both numbers so a finished run can be looked at afterwards, and prints a warning when the two differ by a large margin in either direction. The warning names both numbers and points at population_size. Some methods run one generation at a time and wait for the whole generation to finish before starting the next, so some idle time toward the end of each generation is expected with them. Differential evolution, CMA-ES, and scatter search are marked as such, and their message says so to save the user from looking for a fault that is not there. Only cluster runs are reported. A local run's worker count is exactly what the user asked for, so there is nothing to compare it against. Reading the worker count from dask never stops a fit if it fails. --- pybnf/algorithms/base.py | 74 +++++++++++ pybnf/algorithms/optimizers/cmaes.py | 5 + .../optimizers/differential_evolution.py | 5 + pybnf/algorithms/optimizers/scatter_search.py | 5 + tests/test_run_loop.py | 117 +++++++++++++++++- 5 files changed, 204 insertions(+), 2 deletions(-) diff --git a/pybnf/algorithms/base.py b/pybnf/algorithms/base.py index 6972ba2da..a49bc3135 100644 --- a/pybnf/algorithms/base.py +++ b/pybnf/algorithms/base.py @@ -105,6 +105,15 @@ class Algorithm(ABC): # flag pattern as _is_simplex, so run() never references a leaf subclass. requires_master_scoring = False + # Overridable flag, set True by the generational optimizers (de, cmaes, ss). + # These propose a whole generation of parameter sets, wait for every simulation + # in it to finish, then propose the next generation. Toward the end of each + # generation only a few simulations are still running, so some workers sit idle by + # design. _report_parallelism() says so, to keep that expected idle time from being + # mistaken for a fit that is using fewer processors than it should (#621). The same + # base-class flag pattern as _is_simplex, so run() never names a leaf subclass. + waits_for_full_generation = False + #: The fit's total wall-clock budget (``wall_time_fit``, #529/ADR-0093), or None for #: an unbounded fit. Set by ``pybnf.main()`` on the algorithm it is about to run -- #: and passed on to a refiner / reused across bootstrap replicates -- so one deadline @@ -1286,12 +1295,77 @@ def run(self, client, resume=None, debug=False): # call is exactly the historical one. pool_kwargs = {'timeout': self.budget.remaining()} if self.budget is not None else {} pool = core.as_completed(futures, with_results=True, raise_errors=False, **pool_kwargs) + self._report_parallelism(client, len(futures)) self.completed_simulations = self._drain_job_pool(client, pool, pending, backup_every, debug) logger.info("Cancelling %d pending jobs" % len(pending)) client.cancel(list(pending.keys())) self._finalize_run() + def _report_parallelism(self, client, jobs_in_flight): + """Log how many jobs the fit starts with against how many workers connected, and + warn when the two differ by a large margin (#621). + + How many simulations a fit runs at once follows its settings, mainly + population_size, not how many processors were reserved. When many more workers + connect than there are jobs to run, the extra workers sit idle for the whole run + and nothing else would say so, so a user can reserve several machines and quietly + use a fraction of them. The opposite, many more jobs than workers, means work + queues up and the larger population buys no extra speed. Either way both numbers go + in the log so a finished run can be looked at afterwards. + + Only cluster runs are reported. A local run's worker count is exactly what the user + asked for through parallel_count, so there is nothing to compare it against. The + worker count comes from dask; anything that goes wrong reading it is logged and + never stops a fit. + """ + # A local run drives a LocalCluster that the Client owns, so client.cluster is set. + # A cluster run connects to a scheduler by file or address and has no such object. + if getattr(client, 'cluster', None) is not None: + return + try: + n_workers = len(client.scheduler_info().get('workers', {})) + except Exception: + logger.exception('Could not read the number of connected workers from dask, ' + 'so the parallelism report is skipped') + return + if n_workers <= 0: + return + + logger.info('Parallelism: the fit starts with %d job(s) running and %d worker(s) ' + 'connected.' % (jobs_in_flight, n_workers)) + + # A generational fit drains each generation to almost nothing before starting the + # next, so some idle time is expected with one and should not be read as a fault. + note = '' + if self.waits_for_full_generation: + note = (' This fit runs one generation at a time and waits for all of it to ' + 'finish before starting the next, so some idle time toward the end of ' + 'each generation is expected.') + + # A factor of two in either direction is the "large margin" that draws a warning. + if n_workers >= 2 * jobs_in_flight: + msg = ('The fit starts with only %d job(s) running but %d worker(s) connected, ' + 'so about %d worker(s) will sit idle. How many jobs run at once is set ' + 'by the fitting settings, mainly population_size, not by how many ' + 'processors were reserved. Consider raising population_size or reserving ' + 'fewer processors.%s' + % (jobs_in_flight, n_workers, n_workers - jobs_in_flight, note)) + logger.warning(msg) + print1('Warning: ' + msg) + elif jobs_in_flight >= 2 * n_workers: + msg = ('The fit starts with %d job(s) running but only %d worker(s) connected, ' + 'so jobs will queue and the extra jobs buy no extra speed. How many jobs ' + 'run at once is set by the fitting settings, mainly population_size. ' + 'Consider lowering population_size or reserving more processors.%s' + % (jobs_in_flight, n_workers, note)) + logger.warning(msg) + print1('Warning: ' + msg) + elif note: + # The counts are close, but a generational fit still idles toward the end of + # each generation, so put that on the record for this run. + logger.info(note.strip()) + def _finalize_run(self): """The end-of-fit path: stop reason, final parameter sets, best-fit artifacts, teardown. diff --git a/pybnf/algorithms/optimizers/cmaes.py b/pybnf/algorithms/optimizers/cmaes.py index d81d4bf80..e855472d2 100644 --- a/pybnf/algorithms/optimizers/cmaes.py +++ b/pybnf/algorithms/optimizers/cmaes.py @@ -138,6 +138,11 @@ class CMAESConfig(PyBNFConfigModel): class CMAESAlgorithm(StartPointOptimizer): """CMA-ES as a picklable, generation-synchronized reactor state machine.""" + # This fit samples a whole generation, waits for all of it to finish, then updates the + # search distribution and samples the next, so some idle workers toward the end of each + # generation are expected (#621). + waits_for_full_generation = True + #: Refiner start-point key (see StartPointOptimizer / pybnf._refine_best_fit). START_POINT_KEY = 'cmaes_start_point' diff --git a/pybnf/algorithms/optimizers/differential_evolution.py b/pybnf/algorithms/optimizers/differential_evolution.py index 4d9ab6110..e6bb46163 100644 --- a/pybnf/algorithms/optimizers/differential_evolution.py +++ b/pybnf/algorithms/optimizers/differential_evolution.py @@ -203,6 +203,11 @@ class DifferentialEvolution(MultiStartOptimizer, DifferentialEvolutionBase): """ + # This fit runs one generation at a time and waits for the whole generation to finish + # before proposing the next, so some idle workers toward the end of each generation are + # expected (#621). The asynchronous variant below does not wait, so it leaves this False. + waits_for_full_generation = True + def __init__(self, config): """ Initializes algorithm based on the config object. diff --git a/pybnf/algorithms/optimizers/scatter_search.py b/pybnf/algorithms/optimizers/scatter_search.py index aa77b2ad0..ae3120c3d 100644 --- a/pybnf/algorithms/optimizers/scatter_search.py +++ b/pybnf/algorithms/optimizers/scatter_search.py @@ -48,6 +48,11 @@ class ScatterSearch(MultiStartOptimizer, Algorithm): """ + # This fit runs a whole round of combinations, waits for all of it to finish, then + # builds the next round, so some idle workers toward the end of each round are + # expected (#621). + waits_for_full_generation = True + def __init__(self, config): # variables, popsize, maxiters, saveevery): super().__init__(config) diff --git a/tests/test_run_loop.py b/tests/test_run_loop.py index 7820ba2b8..6c93a359d 100644 --- a/tests/test_run_loop.py +++ b/tests/test_run_loop.py @@ -54,11 +54,28 @@ def result(self): class _FakeClient: """Synchronous stand-in for a distributed.Client. Runs submitted callables - inline and records cancellations.""" + inline and records cancellations. - def __init__(self): + ``_report_parallelism`` reads two things off a real client: ``client.cluster`` + (set only for a local LocalCluster run) and ``client.scheduler_info()`` (the + connected workers). By default this fake looks like a *local* run -- ``cluster`` + is a sentinel object -- so the parallelism report returns immediately and the + orchestration tests are unaffected. Pass ``n_workers`` to make it look like a + *cluster* run with that many workers connected (``cluster`` becomes None), which + is what the parallelism-report tests use.""" + + def __init__(self, n_workers=None): self.submitted = [] # list of (fn, args) self.cancelled = [] + if n_workers is None: + self.cluster = object() # a local run owns its LocalCluster + self._n_workers = 0 + else: + self.cluster = None # a cluster run connects to an outside scheduler + self._n_workers = n_workers + + def scheduler_info(self): + return {'workers': {'tcp://w%d' % i: {} for i in range(self._n_workers)}} def scatter(self, objs, broadcast=False): return [_FakeFuture(o) for o in objs] @@ -355,6 +372,102 @@ def _scored(name, score): return res +class TestReportParallelism: + """``_report_parallelism`` compares how many jobs a fit starts with against how many + workers connected, logs both numbers on a cluster run, and warns when they differ by a + large margin (#621). Driven directly with the fake client, which models a local run by + default and a cluster run when given a worker count.""" + + def _algo(self, tmp_path, generational=False): + algo = _make_algorithm(tmp_path, [[_pset('a', 1.0)]]) + algo.waits_for_full_generation = generational + return algo + + def test_local_run_is_not_reported(self, tmp_path, caplog): + """A local run's worker count is exactly what the user asked for, so there is + nothing to compare and nothing is logged or warned.""" + algo = self._algo(tmp_path) + with caplog.at_level(logging.INFO, logger='pybnf.algorithms'): + algo._report_parallelism(_FakeClient(), 4) + assert 'Parallelism' not in caplog.text + + def test_cluster_run_logs_both_numbers(self, tmp_path, caplog): + """A well-matched cluster run logs both numbers and warns about neither.""" + algo = self._algo(tmp_path) + with caplog.at_level(logging.INFO, logger='pybnf.algorithms'): + algo._report_parallelism(_FakeClient(n_workers=4), 4) + assert 'starts with 4 job(s) running and 4 worker(s) connected' in caplog.text + assert not [r for r in caplog.records if r.levelno >= logging.WARNING] + + def test_idle_workers_warn_and_name_population_size(self, tmp_path, caplog, capsys): + """Many more workers than jobs warns that workers will sit idle, names both numbers + and population_size, and prints the warning to the console.""" + algo = self._algo(tmp_path) + with caplog.at_level(logging.WARNING, logger='pybnf.algorithms'): + algo._report_parallelism(_FakeClient(n_workers=8), 2) + assert 'about 6 worker(s) will sit idle' in caplog.text + assert 'population_size' in caplog.text + assert 'Warning:' in capsys.readouterr().out + + def test_more_jobs_than_workers_warns_about_queueing(self, tmp_path, caplog): + """Many more jobs than workers warns that jobs will queue.""" + algo = self._algo(tmp_path) + with caplog.at_level(logging.WARNING, logger='pybnf.algorithms'): + algo._report_parallelism(_FakeClient(n_workers=2), 8) + assert 'jobs will queue' in caplog.text + assert 'population_size' in caplog.text + + def test_generational_note_added_to_idle_warning(self, tmp_path, caplog): + """A generational fit's idle warning also says the idle time is expected.""" + algo = self._algo(tmp_path, generational=True) + with caplog.at_level(logging.WARNING, logger='pybnf.algorithms'): + algo._report_parallelism(_FakeClient(n_workers=8), 2) + assert 'one generation at a time' in caplog.text + + def test_generational_note_logged_even_when_counts_match(self, tmp_path, caplog): + """With counts well matched there is no warning, but a generational fit still idles + toward the end of each generation, so that note is logged for the record.""" + algo = self._algo(tmp_path, generational=True) + with caplog.at_level(logging.INFO, logger='pybnf.algorithms'): + algo._report_parallelism(_FakeClient(n_workers=4), 4) + assert 'one generation at a time' in caplog.text + assert not [r for r in caplog.records if r.levelno >= logging.WARNING] + + def test_zero_workers_is_not_reported(self, tmp_path, caplog): + """No connected workers means there is nothing to compare against, so nothing is + logged. (A cluster with no workers is caught earlier, at cluster start-up.)""" + algo = self._algo(tmp_path) + with caplog.at_level(logging.INFO, logger='pybnf.algorithms'): + algo._report_parallelism(_FakeClient(n_workers=0), 4) + assert 'Parallelism' not in caplog.text + + def test_unreadable_worker_count_is_not_fatal(self, tmp_path, caplog): + """If reading the worker count from dask fails, the fit is not stopped; the failure + is logged and the report is skipped.""" + class _BrokenClient: + cluster = None + + def scheduler_info(self): + raise RuntimeError('scheduler unreachable') + + algo = self._algo(tmp_path) + with caplog.at_level(logging.INFO, logger='pybnf.algorithms'): + algo._report_parallelism(_BrokenClient(), 4) # must not raise + assert 'parallelism report is skipped' in caplog.text + + +def test_cluster_run_reports_parallelism_end_to_end(tmp_path, monkeypatch, caplog): + """run() calls the parallelism report after submitting the initial jobs: a cluster + client with more workers than the one initial job draws the idle-workers warning.""" + monkeypatch.setattr(algorithms.core, 'as_completed', _FakeAsCompleted) + gens = [[_pset('iter0run0', 10.0)]] + algo = _make_algorithm(tmp_path, gens) + with caplog.at_level(logging.INFO, logger='pybnf.algorithms'): + algo.run(_FakeClient(n_workers=6)) + assert 'starts with 1 job(s) running and 6 worker(s) connected' in caplog.text + assert 'will sit idle' in caplog.text + + class TestRecordResultAndDecide: def test_success_records_and_returns_next_psets(self):