diff --git a/cpp/monoprop/detail/partition/CpuTopology.cpp b/cpp/monoprop/detail/partition/CpuTopology.cpp index 81f0be4d..9438ab1f 100644 --- a/cpp/monoprop/detail/partition/CpuTopology.cpp +++ b/cpp/monoprop/detail/partition/CpuTopology.cpp @@ -20,7 +20,6 @@ #include #include #include -#include #include #include #include @@ -32,8 +31,6 @@ namespace monoprop::detail::partition { namespace { -/* ── Process-lifetime hwloc topology ──────────────────────────────────────── */ - // hwloc_topology_t is safe for concurrent read-only access after hwloc_topology_load(). struct TopologyHolder { hwloc_topology_t topo = nullptr; @@ -70,8 +67,6 @@ auto get_topology() -> hwloc_topology_t { return holder.topo; } -/* ── Effective allowed cpuset for the calling thread ──────────────────────── */ - // Queries the current thread's affinity to respect any launcher-imposed restriction (cgroup, MPI // process binding) narrower than the topology's own allowed cpuset. Falls back to the topology // allowed cpuset when the cpubind query is unsupported on this platform. Caller must free the bitmap. @@ -88,9 +83,33 @@ auto effective_allowed_cpuset(hwloc_topology_t topo) -> hwloc_cpuset_t { return hwloc_bitmap_dup(hwloc_topology_get_allowed_cpuset(topo)); } +// One process has one placement, so the last verdict is the whole state. Locked rather than atomic: +// the five fields are read together and a torn mix of two placements would explain neither. +struct PlacementRecord { + std::mutex mu; + PlacementReport report; +}; + +auto placement_record() -> PlacementRecord & { + static PlacementRecord rec; + return rec; +} + } // anonymous namespace -/* ── topo_detail::placement_order ─────────────────────────────────────────── */ +auto placement_report() -> PlacementReport { + auto &rec = placement_record(); + const std::lock_guard lock(rec.mu); + return rec.report; +} + +auto format_unpinned_line(const PlacementReport &report) -> std::string { + return std::format("monoprop: partition pinning requested but not possible " + "({} cores visible, {} groups x {} partitions); threads run unpinned.\n", + report.cores_visible, + report.groups, + report.partitions); +} namespace topo_detail { @@ -155,8 +174,6 @@ auto placement_order(const std::vector &cores, size_t n, size_t gr } // namespace topo_detail -/* ── enumerate_physical_cores ──────────────────────────────────────────────── */ - auto enumerate_physical_cores() -> std::vector { auto *const topo = get_topology(); if (!topo) { @@ -226,8 +243,6 @@ auto enumerate_physical_cores() -> std::vector { return cores; } -/* ── affinity_mask_words ───────────────────────────────────────────────────── */ - auto affinity_mask_words(uint64_t *out, size_t nwords) -> bool { if (out == nullptr || nwords == 0) { return false; @@ -253,8 +268,6 @@ auto affinity_mask_words(uint64_t *out, size_t nwords) -> bool { return representable; } -/* ── masks_are_pairwise_disjoint ───────────────────────────────────────────── */ - auto masks_are_pairwise_disjoint(const uint64_t *masks, size_t n, size_t words) -> bool { if (masks == nullptr || words == 0 || n < 2) { return false; @@ -281,8 +294,6 @@ auto masks_are_pairwise_disjoint(const uint64_t *masks, size_t n, size_t words) return true; } -/* ── summarize_masks ──────────────────────────────────────────────────────── */ - // hwloc indexes a bitmap in unsigned long units, so a 64-bit word must be one of them. static_assert(sizeof(unsigned long) == sizeof(uint64_t), "the mask word is not an hwloc bitmap unit"); @@ -355,8 +366,6 @@ auto place_line_is_new(std::string_view line) -> bool { return true; } -/* ── partition_cpusets ─────────────────────────────────────────────────────── */ - auto partition_cpusets(size_t n, size_t group_index, size_t group_count, NodeMask mask) -> std::vector { const auto cores = enumerate_physical_cores(); @@ -367,15 +376,23 @@ auto partition_cpusets(size_t n, size_t group_index, size_t group_count, NodeMas } const auto order = topo_detail::placement_order(cores, n, group_index, group_count); - if (order.empty()) { + // `groups` is the count actually used, so a PerRank collapse reads back as the 1 group it became. + PlacementReport report{.pinned = !order.empty(), + .cores_visible = cores.size(), + .groups = group_count, + .partitions = n}; + { + auto &rec = placement_record(); + const std::lock_guard lock(rec.mu); + report.decisions = rec.report.decisions + 1; + rec.report = report; + } + + if (!report.pinned) { + // Kept: this is the only channel that survives with no Python in the process at all. static std::once_flag warned; std::call_once(warned, [&] { - std::print(stderr, - "monoprop: partition pinning requested but not possible " - "({} cores visible, {} groups x {} partitions); threads run unpinned.\n", - cores.size(), - group_count, - n); + std::fputs(format_unpinned_line(report).c_str(), stderr); std::fflush(stderr); }); } @@ -387,8 +404,6 @@ auto partition_cpusets(size_t n, size_t group_index, size_t group_count, NodeMas return sets; } -/* ── pin_this_thread ───────────────────────────────────────────────────────── */ - auto pin_this_thread(const CpuSet &set) -> void { if (set.pu < 0) { return; diff --git a/cpp/monoprop/detail/partition/CpuTopology.h b/cpp/monoprop/detail/partition/CpuTopology.h index 40a0583a..4d318d4f 100644 --- a/cpp/monoprop/detail/partition/CpuTopology.h +++ b/cpp/monoprop/detail/partition/CpuTopology.h @@ -139,6 +139,30 @@ struct MaskSummary { */ [[nodiscard]] auto place_line_is_new(std::string_view line) -> bool; +/*! @brief What the last partition_cpusets() call decided, process-wide. + * + * Placement belongs to the process, not to a propagator, so one record answers "did the partition + * threads get pinned, and if not, why not" for whatever asked last. Refusal is silent by design -- + * pinning is performance-only -- and silence measured 24.6x on propagate at 256 partitions over 128 + * cores, so the outcome is recorded rather than only printed. + */ +struct PlacementReport { + bool pinned = false; //!< false ⇒ the request was refused and every partition thread runs unpinned + size_t cores_visible = 0; //!< physical cores in the calling thread's effective allowed mask + size_t groups = 0; //!< co-located ranks those cores were split between; 1 once a per-rank mask collapses it + size_t partitions = 0; //!< partitions this rank asked to place + uint64_t decisions = 0; //!< placements this process has decided, this one included; 0 ⇒ none yet +}; + +//! Snapshot of that record. Locked, so it is safe from any thread; `decisions` tells a fresh verdict from a stale one. +[[nodiscard]] auto placement_report() -> PlacementReport; + +/*! @brief The one-line "threads run unpinned" message for @p report, newline-terminated. + * Returned rather than written so the stderr print and the binding's RuntimeWarning cannot drift apart, + * and so the text is testable without an oversubscribed host. + */ +[[nodiscard]] auto format_unpinned_line(const PlacementReport &report) -> std::string; + //! Whether the launcher has already handed this rank a private slice of the node, or the node's CPUs are shared. enum class NodeMask { Shared, PerRank }; @@ -158,6 +182,7 @@ enum class NodeMask { Shared, PerRank }; * @p group_count x @p n cores are visible (@p n under PerRank). * * @note Under NodeMask::PerRank the group split is skipped: our share is already this rank's alone. + * @note Records the outcome in placement_report() whether or not it could place. */ auto partition_cpusets(size_t n, size_t group_index = 0, size_t group_count = 1, NodeMask mask = NodeMask::Shared) -> std::vector; diff --git a/cpp/tests/cpu_topology_tests.cpp b/cpp/tests/cpu_topology_tests.cpp index 782fe82c..23e1ddf2 100644 --- a/cpp/tests/cpu_topology_tests.cpp +++ b/cpp/tests/cpu_topology_tests.cpp @@ -457,3 +457,58 @@ BOOST_AUTO_TEST_CASE(cpu_topology_place_line_unknown_is_not_a_verdict) { "COMMPLACE rank=0 node_rank=0 node_size=1 masks=unknown cpus=0 node_cpus=0 " "cpu_list=none\n"); } + +/* ── The placement record ─────────────────────────────────────────────────── */ + +// partition_cpusets() answers an oversubscribed request with an empty vector and a line on C++ +// stderr, which pytest's fd capture swallows; until this record, nothing in-process could tell that +// refusal from a host where hwloc simply loaded no topology. +BOOST_AUTO_TEST_CASE(cpu_topology_placement_report_records_the_refusal) { + const auto cores = partition::enumerate_physical_cores(); + const auto before = partition::placement_report(); + + const auto too_many = partition::partition_cpusets(/*n=*/1'000'000); + // The branch under test, asserted rather than assumed: nothing below means anything if it placed. + BOOST_REQUIRE(too_many.empty()); + + const auto after = partition::placement_report(); + BOOST_CHECK(!after.pinned); + BOOST_CHECK_EQUAL(after.partitions, 1'000'000U); + BOOST_CHECK_EQUAL(after.groups, 1U); + BOOST_CHECK_EQUAL(after.cores_visible, cores.size()); + BOOST_CHECK_EQUAL(after.decisions, before.decisions + 1); +} + +// The other outcome, so a record that answered "unpinned" unconditionally fails here. Neither arm +// skips: a host with no topology is itself a verdict the record has to state. +BOOST_AUTO_TEST_CASE(cpu_topology_placement_report_records_a_placement) { + const auto cores = partition::enumerate_physical_cores(); + const auto one = partition::partition_cpusets(/*n=*/1); + const auto report = partition::placement_report(); + + BOOST_CHECK_EQUAL(report.partitions, 1U); + BOOST_CHECK_EQUAL(report.cores_visible, cores.size()); + BOOST_CHECK_EQUAL(report.pinned, !cores.empty()); + BOOST_CHECK_EQUAL(one.size(), cores.empty() ? 0U : 1U); +} + +// `groups` is the count the placement USED, so the per-rank collapse has to show as the 1 it became. +BOOST_AUTO_TEST_CASE(cpu_topology_placement_report_shows_the_collapsed_group_count) { + partition::partition_cpusets(/*n=*/1, /*group_index=*/3, /*group_count=*/8, partition::NodeMask::PerRank); + BOOST_CHECK_EQUAL(partition::placement_report().groups, 1U); + + partition::partition_cpusets(/*n=*/1, /*group_index=*/3, /*group_count=*/8, partition::NodeMask::Shared); + BOOST_CHECK_EQUAL(partition::placement_report().groups, 8U); +} + +// One text for the stderr line and for the binding's RuntimeWarning: a drift between them is two bugs. +BOOST_AUTO_TEST_CASE(cpu_topology_unpinned_line_names_every_field) { + const partition::PlacementReport report{.pinned = false, + .cores_visible = 128, + .groups = 1, + .partitions = 256, + .decisions = 4}; + BOOST_CHECK_EQUAL(partition::format_unpinned_line(report), + "monoprop: partition pinning requested but not possible " + "(128 cores visible, 1 groups x 256 partitions); threads run unpinned.\n"); +} diff --git a/src/monoprop/__init__.py b/src/monoprop/__init__.py index b36c791f..79978ff7 100644 --- a/src/monoprop/__init__.py +++ b/src/monoprop/__init__.py @@ -27,6 +27,7 @@ antihermitian_generator_correction, has_mpi, is_antihermitian, + placement_report, ) from ._version import version as __version__ from .circuit import ( @@ -68,6 +69,7 @@ "integrals_to_fermion", "is_antihermitian", "jordan_wigner_basis_change", + "placement_report", "validate_parameter_mapping", ] diff --git a/src/monoprop/bindings/bindings.cpp.in b/src/monoprop/bindings/bindings.cpp.in index 9fd6ca24..9052133b 100644 --- a/src/monoprop/bindings/bindings.cpp.in +++ b/src/monoprop/bindings/bindings.cpp.in @@ -23,6 +23,7 @@ #include "monoprop/Info.h" #include "monoprop/MPFunctions.h" #include "monoprop/detail/mpi/MPICompat.h" +#include "monoprop/detail/partition/CpuTopology.h" using namespace monoprop; using namespace nanobind::literals; @@ -93,6 +94,33 @@ auto basis_enum_2_str(Basis basis) -> std::string { throw std::invalid_argument("Unknown Basis enum value"); } } + +/* An unpinned placement costs 16-25x on propagate and says so only on C++ stderr, which pytest's + * file-descriptor capture swallows without `-s`. A RuntimeWarning reaches pytest.warns and the + * caller's warning filters, so a harness can fail closed on it. + * + * PyErr_WarnEx runs Python code and so needs the GIL, which the partition master threads never hold; + * this is called from a binding entry point on the constructing thread, which does. That same GIL + * serialises `warned_through`, which is what keeps the report of an EARLIER placement from being + * re-announced by a construction that placed nothing. + */ +auto warn_if_unpinned(int stack_level) -> bool { + static uint64_t warned_through = 0; + const auto report = monoprop::detail::partition::placement_report(); + if (report.decisions == warned_through) { + return false; + } + warned_through = report.decisions; + if (report.pinned) { + return false; + } + auto text = monoprop::detail::partition::format_unpinned_line(report); + text.pop_back(); // the newline belongs to the stderr line; a warning message carries its own + if (PyErr_WarnEx(PyExc_RuntimeWarning, text.c_str(), stack_level) < 0) { + throw nb::python_error(); // the caller's filter turned this warning into an error + } + return true; +} } // namespace monoprop::bindings::detail NB_MODULE(_core, m) { @@ -129,6 +157,26 @@ NB_MODULE(_core, m) { &monoprop::antihermitian_generator_correction, "indices"_a, "Get the generator correction for a Majorana operator (represented by indices)."); + m.def( + "placement_report", + [] { + const auto r = monoprop::detail::partition::placement_report(); + nb::dict out; + out["pinned"] = r.pinned; + out["cores_visible"] = r.cores_visible; + out["groups"] = r.groups; + out["partitions"] = r.partitions; + out["decisions"] = r.decisions; + return out; + }, + "What the last partition placement decided, process-wide: pinned, cores_visible, groups, " + "partitions, decisions. decisions == 0 means no placement has been attempted yet, and " + "pinned False means every partition thread runs on whatever mask the launcher left."); + m.def("warn_if_unpinned", + &monoprop::bindings::detail::warn_if_unpinned, + "stack_level"_a = 1, + "Raise a RuntimeWarning if the most recent placement ran unpinned, once per placement. " + "Returns whether it warned."); // clang-format off @_BINDINGS_BODY_@ // clang-format on diff --git a/src/monoprop/monomial_propagator.py b/src/monoprop/monomial_propagator.py index 63be1c0b..f98a3640 100644 --- a/src/monoprop/monomial_propagator.py +++ b/src/monoprop/monomial_propagator.py @@ -30,6 +30,7 @@ import numpy as np +from monoprop._core import warn_if_unpinned from monoprop._dispatch import dispatch from .circuit import ( @@ -129,6 +130,11 @@ def _init_simulator( comm=comm, basis=basis, ) + # Unpinned partition threads cost 16-25x and say so only on C++ stderr, which pytest's + # fd capture swallows. Emitted here, on the thread that placed them, because PyErr_WarnEx + # needs the GIL and the partition masters hold none. stack_level=3 names the caller's + # constructor rather than this line. + warn_if_unpinned(stack_level=3) @classmethod def from_circuit( diff --git a/tests/test_placement.py b/tests/test_placement.py new file mode 100644 index 00000000..9fddaf0a --- /dev/null +++ b/tests/test_placement.py @@ -0,0 +1,104 @@ +# Copyright 2026 Algorithmiq +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Partition placement is reportable from Python, and refusing to pin is not silent. + +Asking for more partitions than there are visible physical cores leaves every partition thread +unpinned, which measured 16-25x on propagate. The engine says so on C++ stderr, which pytest's +file-descriptor capture swallows without ``-s``; these cover the two channels that survive it. +""" + +from __future__ import annotations + +import os +import warnings +from contextlib import contextmanager + +import pytest + +from monoprop import MajoranaPropagator, placement_report +from monoprop.fermi import MajoranaOperator + +_REPORT_FIELDS = {"pinned", "cores_visible", "groups", "partitions", "decisions"} + + +@contextmanager +def _oversubscribed(): + """Yield a partition count guaranteed to exceed this process's visible physical cores. + + Confines the process to a single CPU where the platform allows, so the oversubscribed build + costs two threads instead of one per CPU. Where it does not, one more partition than there are + CPUs still exceeds the physical cores, since a core never has fewer than one. Neither arm + skips -- both reach the same refusal. + """ + getter = getattr(os, "sched_getaffinity", None) + setter = getattr(os, "sched_setaffinity", None) + if getter is not None and setter is not None: + saved = getter(0) + try: + setter(0, {min(saved)}) + except OSError: + pass + else: + try: + yield 2 + finally: + setter(0, saved) + return + yield (len(getter(0)) if getter is not None else (os.cpu_count() or 1)) + 1 + + +def _build(serial_comm): + return MajoranaPropagator( + MajoranaOperator({(0, 1, 2, 3): 1.0}, 2), [], cutoff=4, comm=serial_comm + ) + + +def test_placement_report_names_every_field(): + report = placement_report() + assert set(report) == _REPORT_FIELDS + assert isinstance(report["pinned"], bool) + for field in _REPORT_FIELDS - {"pinned"}: + assert isinstance(report[field], int) + + +def test_oversubscription_warns_and_reports_unpinned(monkeypatch, serial_comm): + before = placement_report()["decisions"] + with _oversubscribed() as partitions: + monkeypatch.setenv("monoprop_PARTITIONS", str(partitions)) + with pytest.warns(RuntimeWarning, match="threads run unpinned"): + _build(serial_comm) + report = placement_report() + + # The refusal itself, not merely that some RuntimeWarning was raised. + assert report["pinned"] is False + assert report["partitions"] == partitions + assert report["cores_visible"] < partitions + assert report["decisions"] == before + 1 + + +def test_a_build_that_places_nothing_does_not_re_announce(monkeypatch, serial_comm): + """One placement, one warning: a single-partition build decides nothing and must stay quiet.""" + with _oversubscribed() as partitions: + monkeypatch.setenv("monoprop_PARTITIONS", str(partitions)) + with pytest.warns(RuntimeWarning, match="threads run unpinned"): + _build(serial_comm) + decisions = placement_report()["decisions"] + + monkeypatch.setenv("monoprop_PARTITIONS", "1") + with warnings.catch_warnings(): + # A re-announcement of the stale report raises here. + warnings.simplefilter("error") + _build(serial_comm) + assert placement_report()["decisions"] == decisions