From 7648deb9c9e49d34afcb5ade1f51062945480ed7 Mon Sep 17 00:00:00 2001 From: Dave Woodruff Date: Fri, 24 Jul 2026 10:48:29 -0700 Subject: [PATCH] Flex ranks: two-level layout exchange (#726) and coherence read-outcome diagnostic (#742) Replace the unequal-rank path's interim fullcomm.allgather buffer-layout exchange with a scalable two-level scheme (allgather within each cylinder, allgather across one anchor rank per cylinder, broadcast within each cylinder), chosen over local-compute because extensions register send fields dynamically. Localized to how strata_buffer_layouts is populated at startup; the multi-source reader is untouched, and the equal-rank path keeps its strata_comm allgather verbatim. Add an always-on per-field read-outcome counter to the unequal-rank multi-source reader (new_accepted / not_new / rejected_incoherent / rejected_cross_reader / accepted_mixed), so an infrequently-reporting bounds cylinder can be diagnosed as a coherence problem vs. a slow upstream sender. Each cylinder prints a per-field summary at finalization (aggregated across its ranks, rank-0-gated, only for fields that did multi-source reads); an opt-in periodic line (coherence_diagnostics_period in the spcomm options) supports live debugging; counters are exposed as SPCommunicator.coherence_counters. Tests: new test_flex_layout_exchange.py pins the two-level exchange against fullcomm.allgather across several cylinder partitions (wired into CI and run_coverage.bash); the deterministic coherence-policy stub now also pins the counter buckets; test_flexible_rank_duals.py asserts the counters partition the total and that the strict DUALS field never accepts a blended read. Co-Authored-By: Claude Fable 5 --- .github/workflows/test_pr_and_main.yml | 1 + doc/designs/flexible_rank_assignments.md | 95 ++++++----- mpisppy/cylinders/spcommunicator.py | 174 +++++++++++++++++++- mpisppy/cylinders/spwindow.py | 15 +- mpisppy/spin_the_wheel.py | 4 + mpisppy/tests/test_flex_coherence_policy.py | 53 ++++++ mpisppy/tests/test_flex_layout_exchange.py | 77 +++++++++ mpisppy/tests/test_flexible_rank_duals.py | 39 ++++- run_coverage.bash | 3 + 9 files changed, 409 insertions(+), 52 deletions(-) create mode 100644 mpisppy/tests/test_flex_layout_exchange.py diff --git a/.github/workflows/test_pr_and_main.yml b/.github/workflows/test_pr_and_main.yml index dcef8da32..9a15a741c 100644 --- a/.github/workflows/test_pr_and_main.yml +++ b/.github/workflows/test_pr_and_main.yml @@ -1042,6 +1042,7 @@ jobs: run: | cd mpisppy/tests mpiexec -np 6 coverage run $COV_ARGS -m mpi4py test_spwindow_multisource.py + mpiexec -np 6 coverage run $COV_ARGS -m mpi4py test_flex_layout_exchange.py mpiexec -np 6 coverage run $COV_ARGS -m mpi4py test_flexible_rank_cylinders.py mpiexec -np 6 coverage run $COV_ARGS -m mpi4py test_flexible_rank_duals.py mpiexec -np 6 coverage run $COV_ARGS -m mpi4py test_flexible_rank_xhat.py diff --git a/doc/designs/flexible_rank_assignments.md b/doc/designs/flexible_rank_assignments.md index a507190a1..185cc8211 100644 --- a/doc/designs/flexible_rank_assignments.md +++ b/doc/designs/flexible_rank_assignments.md @@ -377,24 +377,25 @@ window creation. Cons: over one anchor rank per cylinder — rather than a single `fullcomm.allgather`, which scales worse at N=thousands. - *What the communication-layer cut actually ships, and the release - gate.* The first cut uses a single `fullcomm.allgather` for the - unequal-rank layout exchange. This is deliberate but interim: it is - effectively zero new code (the existing `SPWindow` exchange run on - `fullcomm` instead of `strata_comm`), it is a one-time *startup* cost - on a cold path — not the RMA hot path — and at development/test scale - (a handful of ranks) it is free. It is **not** the end state. - Because total rank counts in the thousands are a real operating - regime here, the O(N) startup allgather and its O(N)-per-rank layout - storage must be replaced by the two-level (or local-compute) scheme - **before flexible ranks is documented or recommended for production - use** (it can land on `main` before then, since it is inert until a - non-default ratio). That replacement is its own focused change — it - touches only how - `strata_buffer_layouts` is populated at startup, not the multi-source - reader — so it is tracked as a release-gate item rather than folded - into the feature phases, letting it be reviewed and scale-tested on - its own. See §Gate reliance on the feature with an MPI CI matrix. + *What the unequal-rank path actually ships.* The first cut used a + single `fullcomm.allgather` (the existing `SPWindow` exchange run on + `fullcomm` instead of `strata_comm`) — deliberately interim, because + total rank counts in the thousands are a real operating regime and an + O(N) startup allgather must not be the production exchange. That + interim exchange has since been replaced (Pyomo/mpi-sppy#726) by the + **two-level scheme** (`two_level_layout_exchange` in + `spcommunicator.py`): an allgather *within* each cylinder, an + allgather across one anchor rank per cylinder (the base rank, on a + temporary `fullcomm.Split` comm of just the anchors — a handful of + ranks), and a broadcast of the assembled result within each cylinder. + The two-level exchange was chosen over local-compute because field + registration is dynamic (extensions register extra send fields), so + layouts cannot be reconstructed from static information alone. Every + rank still *stores* all N layouts — each is a small dict of 3-int + tuples, and a reader legitimately needs the layout of any peer rank + its overlap maps touch — only the exchange pattern changed. The + change is localized to how `strata_buffer_layouts` is populated at + startup; the multi-source reader is untouched. *Lock granularity.* `MPI_Win_lock(rank=target, ...)` is per- target-rank in the MPI spec, not per-window — a writer's exclusive @@ -578,6 +579,28 @@ two code paths in the multi-source reader. No cylinder-wide iteration counter (it would add synchronization the async design avoids and is unnecessary given the per-field analysis). +#### Read-outcome diagnostic + +An always-on, per-field counter at the multi-source reader +(Pyomo/mpi-sppy#742; `_count_coherence_read` in `spcommunicator.py`) +buckets every multi-source read (>= 2 sources) as `new_accepted` / +`not_new` / `rejected_incoherent` / `rejected_cross_reader` / +`accepted_mixed`. This lets an infrequently-reporting bounds cylinder +be diagnosed as a coherence problem (reads rejected because sources +disagree on `write_id`, or blended on a relaxed field) vs. a slow +upstream sender (no new data), and measures how often a multi-source +read actually straddles a publish — the empirical basis for the +strict-vs-relaxed choices above, especially under an asynchronous APH +sender. Cost is two integer increments per multi-source read, so +counting is unconditional; each cylinder prints a per-field summary at +finalization (`report_coherence_diagnostics`, aggregated across the +cylinder's ranks, rank-0-gated, only for fields that did multi-source +reads — so equal-rank runs print nothing), and an opt-in periodic line +(`coherence_diagnostics_period` in the spcomm options: print local +counters every N multi-source reads) supports live debugging. The +counters are exposed programmatically as +`SPCommunicator.coherence_counters`. + ### Impact on Existing Components @@ -697,12 +720,11 @@ the first pass bundled into "Phase 0" has already landed separately. **Phase 2: Communication layer** (additive; reached only when a ratio differs from 1.0) -- Add the `fullcomm.allgather` layout exchange for the unequal-rank - path (Option D's addressing), *alongside* the existing - `strata_comm`-based exchange, which the equal-rank path keeps using. - This is the interim exchange; the scalable replacement is a release - gate, not a feature phase (see the Option D layout-exchange note and - §Gate reliance on the feature with an MPI CI matrix). +- Add the layout exchange for the unequal-rank path (Option D's + addressing), *alongside* the existing `strata_comm`-based exchange, + which the equal-rank path keeps using. (The first cut's interim + `fullcomm.allgather` has since been replaced by the scalable + two-level exchange — see the Option D layout-exchange note.) - Implement multi-source `get_receive_buffer()` using overlap maps, as a path taken only under non-default ratios; the single-source reader is unchanged for the equal-rank case. @@ -905,13 +927,13 @@ two MPI implementations (e.g. OpenMPI and MPICH) and more than one mpi4py / MPI version, since that path is where the RMA-portability risk lives. -The same "finish before recommending it" list carries the **scalable -layout exchange**: the interim `fullcomm.allgather` (see the Option D -layout-exchange note) must be replaced by the two-level or local-compute -scheme before the feature is documented or recommended for production -use, because total rank counts in the thousands are a real operating -regime here. Both are prerequisites for recommending the feature, not -for landing the intervening phases on `main`. +The **scalable layout exchange** used to sit on this same "finish +before recommending it" list; it is now done — the interim +`fullcomm.allgather` was replaced by the two-level exchange +(Pyomo/mpi-sppy#726; see the Option D layout-exchange note) — leaving +the MPI-implementation matrix above as the remaining prerequisite for +recommending the feature (not for landing the intervening phases on +`main`). ### Possible future work (out of scope) @@ -926,17 +948,6 @@ scenario, saving storage and bandwidth on those fields. This is *only* valid for the Category-2 fields — never for the Category-1 per-scenario fields — and is explicitly **not** required for flexible ranks. -**Coherence read-outcome diagnostic** (Pyomo/mpi-sppy#742). An -always-on, per-field counter at the multi-source reader breaking each -read into `not_new` / `new_accepted` / `rejected_incoherent` / -`accepted_mixed`. This lets an infrequently-reporting bounds cylinder be -diagnosed as a coherence problem (reads rejected because sources disagree -on `write_id`) vs. a slow upstream sender (no new data), and measures how -often a multi-source read straddles a publish — the empirical basis for -the strict-vs-relaxed choices above, especially under an asynchronous APH -sender. - - ### Backward Compatibility When all rank ratios are 1.0 (the default), the system behaves diff --git a/mpisppy/cylinders/spcommunicator.py b/mpisppy/cylinders/spcommunicator.py index 1a45e1645..df466bb92 100644 --- a/mpisppy/cylinders/spcommunicator.py +++ b/mpisppy/cylinders/spcommunicator.py @@ -111,6 +111,54 @@ def reduce_source_write_ids(source_ids, strict: bool) -> int: return source_ids[0] if len(set(source_ids)) == 1 else -1 return min(source_ids) + +def two_level_layout_exchange(my_layout, fullcomm, cylinder_comm, cylinder_bases): + """Exchange every rank's buffer layout on the unequal-rank path, where the + window lives on ``fullcomm``, without a flat ``fullcomm.allgather`` (an + O(N) startup collective that must not be the exchange at total rank counts + in the thousands). Three scalable steps instead: + + 1. allgather within each cylinder (size = that cylinder's rank count); + 2. allgather across one anchor rank per cylinder (the base rank, + ``cylinder_comm`` rank 0) on a temporary comm of just the anchors + (size = the number of cylinders, a small constant); + 3. broadcast the assembled result within each cylinder. + + Every rank still *stores* all N layouts (each is a small dict of + 3-int tuples), because a reader legitimately needs the layout of any + peer rank its overlap maps touch; only the exchange pattern changes. + + Args: + my_layout: this rank's ``SPWindow.buffer_layout``. + fullcomm: the window comm spanning all ranks. + cylinder_comm: this rank's within-cylinder comm, rank-ordered by + global rank (so its rank 0 is the cylinder's base rank). + cylinder_bases: each cylinder's base (lowest) global rank, in + cylinder order. + + Returns: + list: every rank's layout, indexed by global rank on ``fullcomm``. + """ + cylinder_layouts = cylinder_comm.allgather(my_layout) + # anchor comm: cylinder base ranks only; Split is collective on fullcomm + is_anchor = cylinder_comm.Get_rank() == 0 + anchor_comm = fullcomm.Split( + color=0 if is_anchor else MPI.UNDEFINED, key=fullcomm.Get_rank() + ) + if is_anchor: + # anchors are ordered by global rank, i.e. in cylinder order + per_cylinder_layouts = anchor_comm.allgather(cylinder_layouts) + anchor_comm.Free() + else: + per_cylinder_layouts = None + per_cylinder_layouts = cylinder_comm.bcast(per_cylinder_layouts, root=0) + + layouts = [None] * fullcomm.Get_size() + for base, one_cylinder in zip(cylinder_bases, per_cylinder_layouts): + layouts[base : base + len(one_cylinder)] = one_cylinder + assert None not in layouts + return layouts + def communicator_array(data_length: int): """ Allocate an MPI memory region with a padded length (multiple of 8 doubles = 64B), @@ -347,6 +395,18 @@ def __init__(self, spbase_object, fullcomm, strata_comm, cylinder_comm, communic self.overlap_maps = {} # -> list[OverlapSegment] (global ranks) self._overlap_source_ranks = {} # -> sorted distinct source global ranks + # Per-field read-outcome counters for the unequal-rank multi-source + # reader (see _count_coherence_read); always accumulated (two integer + # increments per multi-source read), reported at finalization by + # report_coherence_diagnostics. Empty on the equal-rank path and for + # single-source reads, which cannot straddle a publish. + self.coherence_counters = {} + # opt-in periodic per-field line for live debugging: print local + # counters every N multi-source reads (0 = off) + self._coherence_report_period = int( + self.options.get("coherence_diagnostics_period", 0) + ) + # setup FieldLengths which calculates # the length of each buffer type based # on the problem data @@ -541,9 +601,13 @@ def make_windows(self) -> None: window_spec = self._build_window_spec() # Equal-rank: window on strata_comm (rank i of every cylinder), # addressed by strata_rank. Unequal-rank: window on fullcomm, - # addressed by global rank via overlap maps (strata_comm is None). + # addressed by global rank via overlap maps (strata_comm is None); + # the layout exchange then must not be a flat allgather on fullcomm, + # so pass the two-level exchange (see two_level_layout_exchange). window_comm = self.fullcomm if self._flex_ranks else self.strata_comm - self.window = SPWindow(window_spec, window_comm) + layout_exchanger = self._flex_layout_exchange if self._flex_ranks else None + self.window = SPWindow(window_spec, window_comm, + layout_exchanger=layout_exchanger) self._create_field_rank_mappings() self.register_receive_fields() @@ -698,6 +762,13 @@ def get_receive_buffer(self, # self._flex_ranks is True; the equal-rank path above never calls them. # ------------------------------------------------------------------ + def _flex_layout_exchange(self, my_layout): + """The unequal-rank window's layout exchange: two-level instead of a + flat allgather on fullcomm (see two_level_layout_exchange).""" + return two_level_layout_exchange( + my_layout, self.fullcomm, self.cylinder_comm, self._cylinder_bases + ) + def _items_per_scen_for_field(self, field: Field): """Number of field items each scenario contributes, indexed by global scenario index, for a per-scenario field (used to build overlap maps). @@ -893,15 +964,32 @@ def _flex_get_multi_source(self, buf, field, peer_cylinder, synchronize): source_snapshots[r] = snapshot source_ids.append(int(snapshot[logical_len - 1])) - new_id = reduce_source_write_ids( - source_ids, strict=field in _STRICT_COHERENCE_FIELDS - ) + strict = field in _STRICT_COHERENCE_FIELDS + new_id = reduce_source_write_ids(source_ids, strict=strict) + + # Read-outcome diagnostic: count genuinely multi-source reads (>= 2 + # sources; a single source cannot straddle a publish). The outcome + # buckets let a user tell a coherence problem (reads rejected or + # blended) from a slow upstream sender (nothing new to read) when a + # consumer appears to report infrequently. + counters = None + if len(source_ids) >= 2: + counters = self._count_coherence_read(field) + mixed = len(set(source_ids)) > 1 if not self._write_ids_agree(new_id, synchronize): + if counters is not None: + # a strict local-source mismatch is the fundamental coherence + # miss; otherwise this rank's sources agreed and the collective + # cross-reader check rejected the read + counters["rejected_incoherent" if strict and mixed + else "rejected_cross_reader"] += 1 buf._is_new = False return False if new_id > last_id: + if counters is not None: + counters["accepted_mixed" if mixed else "new_accepted"] += 1 # assemble the accepted data into buf, then commit via the shared # _mark_new (which stamps the id slot the assembly does not touch) data_view = buf.value_array() @@ -910,9 +998,85 @@ def _flex_get_multi_source(self, buf, field, peer_cylinder, synchronize): data_view[seg.local_offset : seg.local_offset + seg.count] = \ snapshot[seg.remote_offset : seg.remote_offset + seg.count] return self._mark_new(buf, new_id) + if counters is not None: + # strict + mixed lands here when every reader rank computed the + # sentinel -1, so cross-reader agreement held but the id cannot + # advance -- still a coherence rejection, not a slow sender + counters["rejected_incoherent" if strict and mixed + else "not_new"] += 1 buf._is_new = False return False + def _count_coherence_read(self, field: Field) -> dict: + """Count one multi-source read of ``field`` and return its outcome + counters (created on first use) for the caller to bucket: + + * ``new_accepted`` -- sources agreed on an advanced write_id; used. + * ``not_new`` -- coherent, but the write_id did not advance (the + sender has not published since the last accepted read). + * ``rejected_incoherent`` -- a strict field's sources disagreed, so + the read was rejected and will be retried (the fundamental + coherence miss: the read straddled a publish). + * ``rejected_cross_reader`` -- this rank's sources agreed, but the + collective cross-reader write_id check rejected the read (some + other rank of this cylinder saw a different id). + * ``accepted_mixed`` -- a relaxed field's sources disagreed and the + blended assembly was used anyway. + + The buckets partition ``total``. The coherence miss rate is + ``(rejected_incoherent + accepted_mixed) / total``; if ``not_new`` + dominates instead, the upstream sender is just slow. + """ + counters = self.coherence_counters.setdefault(field, { + "total": 0, + "new_accepted": 0, + "not_new": 0, + "rejected_incoherent": 0, + "rejected_cross_reader": 0, + "accepted_mixed": 0, + }) + counters["total"] += 1 + if self._coherence_report_period > 0 and self.cylinder_rank == 0 \ + and counters["total"] % self._coherence_report_period == 0: + # live-debugging line: this rank's counts only (the current + # read's outcome bucket is not yet incremented) + print(f"coherence diagnostic [{self.__class__.__name__}] " + f"{field.name}: " + + ", ".join(f"{k}={v}" for k, v in counters.items()), + flush=True) + return counters + + def report_coherence_diagnostics(self): + """Print a per-field summary of the multi-source read outcomes + accumulated in ``coherence_counters`` (see _count_coherence_read for + the buckets and their diagnosis). Collective on ``cylinder_comm``: + every rank of the cylinder must call it (different ranks can have + different multi-source fields, or none, so the counters are gathered + rather than reduced); rank 0 prints. Inert -- no output, one gather -- + at equal ranks or when no multi-source reads happened. + """ + if not self._flex_ranks: + return + all_counters = self.cylinder_comm.gather(self.coherence_counters, root=0) + if self.cylinder_rank != 0: + return + totals = {} + for rank_counters in all_counters: + for field, counters in rank_counters.items(): + aggregate = totals.setdefault(field, dict.fromkeys(counters, 0)) + for outcome, count in counters.items(): + aggregate[outcome] += count + for field in sorted(totals): + counters = totals[field] + if counters["total"] == 0: + continue + misses = counters["rejected_incoherent"] + counters["accepted_mixed"] + print(f"coherence diagnostic [{self.__class__.__name__}] " + f"{field.name}: " + + ", ".join(f"{k}={v}" for k, v in counters.items()) + + f", miss rate={misses / counters['total']:.2%}", + flush=True) + def receive_nonant_bounds(self): """ receive the bounds on the nonanticipative variables based on Field.NONANT_LOWER_BOUNDS and Field.NONANT_UPPER_BOUNDS. Updates the diff --git a/mpisppy/cylinders/spwindow.py b/mpisppy/cylinders/spwindow.py index 49f34563c..812e425f1 100644 --- a/mpisppy/cylinders/spwindow.py +++ b/mpisppy/cylinders/spwindow.py @@ -126,12 +126,20 @@ def padded_len_n_doubles(logical_len: int) -> int: class SPWindow: - def __init__(self, my_fields: dict, strata_comm: MPI.Comm, field_order=None): + def __init__(self, my_fields: dict, strata_comm: MPI.Comm, field_order=None, + layout_exchanger=None): """ Design A (padded transfers): - Layout tuple is (offset, logical_len, padded_len) and offset advances by padded_len. - put/get always transfer padded_len doubles. - ID slot is at (offset + logical_len - 1). + + ``layout_exchanger``, when given, is a callable mapping this rank's + ``buffer_layout`` to the list of every window rank's layout (indexed by + rank on ``strata_comm``); it replaces the default + ``strata_comm.allgather``. The unequal-rank path passes a two-level + exchange here because its window comm is ``fullcomm``, where a flat + allgather does not scale to total rank counts in the thousands. """ self.strata_comm = strata_comm self.strata_rank = strata_comm.Get_rank() @@ -184,7 +192,10 @@ def __init__(self, my_fields: dict, strata_comm: MPI.Comm, field_order=None): self.buff[off + logical_len - 1] = 0.0 # Gather layouts across ranks - self.strata_buffer_layouts = strata_comm.allgather(self.buffer_layout) + if layout_exchanger is None: + self.strata_buffer_layouts = strata_comm.allgather(self.buffer_layout) + else: + self.strata_buffer_layouts = layout_exchanger(self.buffer_layout) def free(self): if self.window is not None: diff --git a/mpisppy/spin_the_wheel.py b/mpisppy/spin_the_wheel.py index f90a22211..6ec719896 100644 --- a/mpisppy/spin_the_wheel.py +++ b/mpisppy/spin_the_wheel.py @@ -179,6 +179,10 @@ def run(self, comm_world=None): # Anything that's left to do spcomm.finalize() + # Unequal-rank runs: per-field multi-source read-outcome summary + # (collective on cylinder_comm; inert at equal ranks) + spcomm.report_coherence_diagnostics() + # to ensure the messages below are True cylinder_comm.Barrier() global_toc(f"Hub algorithm {opt_class.__name__} complete, waiting for spoke finalization", comm_world.rank == 0) diff --git a/mpisppy/tests/test_flex_coherence_policy.py b/mpisppy/tests/test_flex_coherence_policy.py index 21c7842c5..29d55a6ed 100644 --- a/mpisppy/tests/test_flex_coherence_policy.py +++ b/mpisppy/tests/test_flex_coherence_policy.py @@ -170,6 +170,9 @@ def _make_reader(): (Field.DUALS, peer): _SEGMENTS, (Field.NONANTS_VALS, peer): _SEGMENTS, } + # read-outcome diagnostic state (set by __init__ in the real class) + sp.coherence_counters = {} + sp._coherence_report_period = 0 return sp, peer @@ -234,5 +237,55 @@ def test_relaxed_accepts_mixed_at_the_floor(self): np.testing.assert_allclose(buf.value_array(), _ASSEMBLED) +class TestCoherenceCounters(unittest.TestCase): + """The read-outcome diagnostic, driven deterministically through the same + stub: each read outcome must land in exactly one counter bucket, so the + buckets partition the total and the miss rate is computable after the run. + (synchronize is False throughout, so rejected_cross_reader stays 0; that + bucket needs the collective check the MPI integration test exercises.)""" + + def _read(self, sp, buf, field, peer): + return sp._flex_get_multi_source(buf, field, peer, synchronize=False) + + def _counters(self, sp, field): + return sp.coherence_counters[field] + + def test_outcomes_bucketed_and_partition_total(self): + sp, peer = _make_reader() + buf = RecvArray(_DATA_LEN * 2) + + # strict + mixed -> rejected_incoherent (the fundamental miss) + sp.window.set_write_ids(5, 4) + self._read(sp, buf, Field.DUALS, peer) + # coherent + advanced -> new_accepted + sp.window.set_write_ids(6, 6) + self._read(sp, buf, Field.DUALS, peer) + # coherent, id unchanged -> not_new (a slow sender, not a miss) + self._read(sp, buf, Field.DUALS, peer) + + counters = self._counters(sp, Field.DUALS) + self.assertEqual(counters["rejected_incoherent"], 1) + self.assertEqual(counters["new_accepted"], 1) + self.assertEqual(counters["not_new"], 1) + self.assertEqual(counters["accepted_mixed"], 0) + self.assertEqual(counters["rejected_cross_reader"], 0) + self.assertEqual(counters["total"], 3) + + def test_relaxed_mixed_counts_as_accepted_mixed(self): + sp, peer = _make_reader() + buf = RecvArray(_DATA_LEN * 2) + + sp.window.set_write_ids(7, 5) # mixed; relaxed accepts at the floor + self._read(sp, buf, Field.NONANTS_VALS, peer) + # mixed again but the floor has not advanced -> not_new, not a miss + self._read(sp, buf, Field.NONANTS_VALS, peer) + + counters = self._counters(sp, Field.NONANTS_VALS) + self.assertEqual(counters["accepted_mixed"], 1) + self.assertEqual(counters["not_new"], 1) + self.assertEqual(counters["rejected_incoherent"], 0) + self.assertEqual(counters["total"], 2) + + if __name__ == "__main__": unittest.main() diff --git a/mpisppy/tests/test_flex_layout_exchange.py b/mpisppy/tests/test_flex_layout_exchange.py new file mode 100644 index 000000000..561ce94fc --- /dev/null +++ b/mpisppy/tests/test_flex_layout_exchange.py @@ -0,0 +1,77 @@ +############################################################################### +# mpi-sppy: MPI-based Stochastic Programming in PYthon +# +# Copyright (c) 2024, Lawrence Livermore National Security, LLC, Alliance for +# Sustainable Energy, LLC, The Regents of the University of California, et al. +# All rights reserved. Please see the files COPYRIGHT.md and LICENSE.md for +# full copyright and license information. +############################################################################### +"""Direct test of the unequal-rank two-level buffer-layout exchange. + +``two_level_layout_exchange`` replaces the flat ``fullcomm.allgather`` +layout exchange on the unequal-rank path (an O(N) startup collective that +must not be the exchange at total rank counts in the thousands) with an +allgather within each cylinder, an allgather across one anchor rank per +cylinder, and a within-cylinder broadcast. Its contract is that the +result is *identical* to what ``fullcomm.allgather`` would have produced +-- a list of every rank's layout indexed by global rank -- so this test +pins the two against each other, with rank-unique payloads (misrouting +any rank's layout breaks equality) across several cylinder partitions, +including single-rank cylinders at either end and a single-cylinder run. + +mpiexec -np 6 python -m mpi4py -m pytest mpisppy/tests/test_flex_layout_exchange.py +""" + +import unittest + +from mpi4py import MPI + +from mpisppy.cylinders.spcommunicator import two_level_layout_exchange +from mpisppy.utils.rank_apportionment import cylinder_bases, rank_to_cylinder + +comm = MPI.COMM_WORLD + + +@unittest.skipUnless(comm.size == 6, "needs exactly 6 MPI ranks") +class TestTwoLevelLayoutExchange(unittest.TestCase): + + PARTITIONS = ( + [6], + [1, 5], + [5, 1], + [4, 2], + [2, 4], + [3, 3], + [2, 2, 2], + [1, 2, 3], + [3, 2, 1], + [1, 1, 1, 1, 1, 1], + ) + + def _exchange_matches_allgather(self, rank_counts): + global_rank = comm.Get_rank() + cylinder_index, _ = rank_to_cylinder(global_rank, rank_counts) + # same split spin_the_wheel uses on the unequal-rank path + cylinder_comm = comm.Split(color=cylinder_index, key=global_rank) + try: + # a rank-unique picklable payload standing in for the layout dict + my_layout = { + "global_rank": global_rank, + "cylinder": cylinder_index, + "data": list(range(global_rank + 1)), + } + result = two_level_layout_exchange( + my_layout, comm, cylinder_comm, cylinder_bases(rank_counts) + ) + self.assertEqual(result, comm.allgather(my_layout)) + finally: + cylinder_comm.Free() + + def test_matches_allgather_across_partitions(self): + for rank_counts in self.PARTITIONS: + with self.subTest(rank_counts=rank_counts): + self._exchange_matches_allgather(rank_counts) + + +if __name__ == "__main__": + unittest.main() diff --git a/mpisppy/tests/test_flexible_rank_duals.py b/mpisppy/tests/test_flexible_rank_duals.py index fbeb8ac8b..955c0f502 100644 --- a/mpisppy/tests/test_flexible_rank_duals.py +++ b/mpisppy/tests/test_flexible_rank_duals.py @@ -85,7 +85,14 @@ def _run(num_scens, max_iterations, hub_ratio, spoke_ratio): payload = wheel.BestOuterBound else: payload = None - return comm.bcast(payload, root=0) + bound = comm.bcast(payload, root=0) + # every rank's coherence read-outcome counters (field name -> buckets), + # allgathered so each rank can assert on the whole picture + local_counters = { + field.name: dict(counters) + for field, counters in wheel.spcomm.coherence_counters.items() + } + return bound, comm.allgather(local_counters) @unittest.skipUnless(comm.size == 6, "needs exactly 6 MPI ranks") @@ -96,8 +103,9 @@ class TestFlexibleRankDuals(unittest.TestCase): MAX_ITERS = 50 def test_strict_duals_both_directions_agree_and_bound_valid(self): - ob_42 = _run(self.NUM_SCENS, self.MAX_ITERS, 1.0, 0.5) # 4-rank hub, 2-rank spoke - ob_24 = _run(self.NUM_SCENS, self.MAX_ITERS, 1.0, 2.0) # 2-rank hub, 4-rank spoke + # 4-rank hub, 2-rank spoke / 2-rank hub, 4-rank spoke + ob_42, counters_42 = _run(self.NUM_SCENS, self.MAX_ITERS, 1.0, 0.5) + ob_24, counters_24 = _run(self.NUM_SCENS, self.MAX_ITERS, 1.0, 2.0) # Finite bound (a broken read would give inf or error out). self.assertTrue(abs(ob_42) < float("inf")) @@ -115,6 +123,31 @@ def test_strict_duals_both_directions_agree_and_bound_valid(self): self.assertLessEqual(ob_42, EF_OPT + 1e-4 * abs(EF_OPT)) self.assertLessEqual(ob_24, EF_OPT + 1e-4 * abs(EF_OPT)) + # Coherence read-outcome diagnostic. Only reads with >= 2 sources are + # counted, so presence depends on the split: in 4+2 each 5-scenario + # spoke rank must straddle the 4-rank hub's <=3-scenario slices + # (guaranteed multi-source), while in 2+4 the spoke's slices may nest + # inside the hub's halves and leave every read single-source. + for all_counters in (counters_42, counters_24): + for rank_counters in all_counters: + for counters in rank_counters.values(): + # the outcome buckets partition the total + self.assertGreater(counters["total"], 0) + self.assertEqual( + counters["total"], + counters["new_accepted"] + counters["not_new"] + + counters["rejected_incoherent"] + + counters["rejected_cross_reader"] + + counters["accepted_mixed"], + ) + # DUALS is strict: a blended assembly must never be accepted + if "DUALS" in rank_counters: + self.assertEqual(rank_counters["DUALS"]["accepted_mixed"], 0) + duals_42 = [c["DUALS"] for c in counters_42 if "DUALS" in c] + self.assertGreater(len(duals_42), 0) + # the 4+2 bound can only have come from accepted multi-source reads + self.assertGreater(sum(c["new_accepted"] for c in duals_42), 0) + if __name__ == "__main__": unittest.main() diff --git a/run_coverage.bash b/run_coverage.bash index 2f3453302..6dd9d00dd 100755 --- a/run_coverage.bash +++ b/run_coverage.bash @@ -265,6 +265,9 @@ run_phase "test_dualcg_with_cylinders (mpiexec -np 2)" \ run_phase "test_spwindow_multisource (mpiexec -np 6)" \ mpiexec -np 6 $OVERSUBSCRIBE coverage run --rcfile="$PROJ_DIR/.coveragerc" -m mpi4py mpisppy/tests/test_spwindow_multisource.py +run_phase "test_flex_layout_exchange (mpiexec -np 6)" \ + mpiexec -np 6 $OVERSUBSCRIBE coverage run --rcfile="$PROJ_DIR/.coveragerc" -m mpi4py mpisppy/tests/test_flex_layout_exchange.py + run_phase "test_flexible_rank_cylinders (mpiexec -np 6)" \ mpiexec -np 6 $OVERSUBSCRIBE coverage run --rcfile="$PROJ_DIR/.coveragerc" -m mpi4py mpisppy/tests/test_flexible_rank_cylinders.py