diff --git a/pybnf/algorithms/base.py b/pybnf/algorithms/base.py index 6972ba2d..a49bc313 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 d81d4bf8..e855472d 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 4d9ab611..e6bb4616 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 aa77b2ad..ae3120c3 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 7820ba2b..6c93a359 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):