Warn when a fit uses far fewer processors than were reserved (#621)

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.

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'
 

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.

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)

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):