diff --git a/cpp/monoprop/detail/partition/CpuTopology.cpp b/cpp/monoprop/detail/partition/CpuTopology.cpp index dd019d8e..6c0387f3 100644 --- a/cpp/monoprop/detail/partition/CpuTopology.cpp +++ b/cpp/monoprop/detail/partition/CpuTopology.cpp @@ -15,7 +15,10 @@ #include "monoprop/detail/partition/CpuTopology.h" #include +#include #include +#include +#include #include #include @@ -88,7 +91,7 @@ namespace topo_detail { auto placement_order(const std::vector &cores, size_t n, size_t group_index, size_t group_count) -> std::vector { - if (cores.empty() || group_count * n > cores.size()) { + if (cores.empty() || group_count == 0 || group_count * n > cores.size()) { return {}; } @@ -218,15 +221,89 @@ 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; + } + std::fill_n(out, nwords, uint64_t{0}); + const auto topo = get_topology(); + if (!topo) { + return false; + } + const hwloc_cpuset_t allowed = effective_allowed_cpuset(topo); + if (!allowed) { + return false; + } + // Refused rather than truncated: a truncated mask could compare disjoint against a peer it overlaps. + const int last = hwloc_bitmap_last(allowed); + const bool representable = last >= 0 && static_cast(last) < nwords * 64; + if (representable) { + for (int pu = hwloc_bitmap_first(allowed); pu >= 0; pu = hwloc_bitmap_next(allowed, pu)) { + out[static_cast(pu) / 64] |= uint64_t{1} << (static_cast(pu) % 64); + } + } + hwloc_bitmap_free(allowed); + 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; + } + // An all-zero mask is disjoint from everything, so empty is rejected before the pairwise test. + for (size_t r = 0; r < n; ++r) { + bool any = false; + for (size_t w = 0; w < words && !any; ++w) { + any = masks[(r * words) + w] != 0; + } + if (!any) { + return false; + } + } + for (size_t a = 0; a < n; ++a) { + for (size_t b = a + 1; b < n; ++b) { + for (size_t w = 0; w < words; ++w) { + if ((masks[(a * words) + w] & masks[(b * words) + w]) != 0) { + return false; // two peers share a CPU: not private + } + } + } + } + return true; +} + /* ── partition_cpusets ─────────────────────────────────────────────────────── */ -auto partition_cpusets(size_t n, size_t group_index, size_t group_count) -> std::vector { +auto partition_cpusets(size_t n, size_t group_index, size_t group_count, bool mask_is_private) -> std::vector { if (!config::get().partition_pinning) { return {}; } const auto cores = enumerate_physical_cores(); + + // A private mask IS this rank's share: the launcher already separated co-located ranks. + if (mask_is_private) { + group_index = 0; + group_count = 1; + } const auto order = topo_detail::placement_order(cores, n, group_index, group_count); + if (order.empty()) { + 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::fflush(stderr); + }); + } + std::vector sets(order.size()); for (size_t i = 0; i < order.size(); ++i) { sets[i] = CpuSet{order[i]}; diff --git a/cpp/monoprop/detail/partition/CpuTopology.h b/cpp/monoprop/detail/partition/CpuTopology.h index 1b60359e..1928d2d1 100644 --- a/cpp/monoprop/detail/partition/CpuTopology.h +++ b/cpp/monoprop/detail/partition/CpuTopology.h @@ -24,7 +24,9 @@ #pragma once +#include #include +#include #include #include "monoprop/detail/EnvConfig.h" @@ -88,6 +90,20 @@ auto placement_order(const std::vector &cores, size_t n, size_t gr */ auto enumerate_physical_cores() -> std::vector; +//! Affinity-mask exchange width, in 64-bit words. A mask needing more is "cannot classify", never private. +inline constexpr size_t kAffinityMaskWords = 64; + +static_assert(kAffinityMaskWords > 0 && kAffinityMaskWords <= static_cast(INT_MAX), + "the affinity-mask width is an MPI_Allgather element count, which is an int"); + +//! This process's allowed cpuset as @p nwords 64-bit words; false with @p out zeroed when it does not fit. +auto affinity_mask_words(uint64_t *out, size_t nwords) -> bool; + +/*! @brief Whether the @p n masks of @p words words laid end to end in @p masks are pairwise disjoint. + * False for @p n < 2 and for any empty mask: all-zero is disjoint from everything, and shared is the safe error. + */ +[[nodiscard]] auto masks_are_pairwise_disjoint(const uint64_t *masks, size_t n, size_t words) -> bool; + /*! * @brief Build placement tokens for one MPI rank's partitions. * @@ -98,10 +114,15 @@ auto enumerate_physical_cores() -> std::vector; * @param n Number of partitions to place. * @param group_index This rank's 0-based index among the co-located ranks on the host. * @param group_count Total number of co-located ranks on the host. + * @param mask_is_private True only when the co-located ranks' affinity masks have been measured + * pairwise DISJOINT (PartitionGroup::classify_node_masks_), so this mask is our share. * @returns Vector of @p n CpuSet tokens, or empty when @c monoprop_PARTITION_PINNING is disabled, - * hwloc is unavailable, or the host cannot provide @p group_count × @p n distinct cores. + * hwloc is unavailable, or fewer than @p group_count x @p n cores are visible (@p n when private). + * + * @note Under @p mask_is_private the group split is skipped: our share is already this rank's alone. */ -auto partition_cpusets(size_t n, size_t group_index = 0, size_t group_count = 1) -> std::vector; +auto partition_cpusets(size_t n, size_t group_index = 0, size_t group_count = 1, bool mask_is_private = false) + -> std::vector; /*! * @brief Bind the calling thread to the PU identified by @p set. diff --git a/cpp/monoprop/detail/partition/PartitionGroup.h b/cpp/monoprop/detail/partition/PartitionGroup.h index fc3f64fd..169a0020 100644 --- a/cpp/monoprop/detail/partition/PartitionGroup.h +++ b/cpp/monoprop/detail/partition/PartitionGroup.h @@ -14,8 +14,10 @@ #pragma once +#include #include #include +#include #include #include #include @@ -59,7 +61,7 @@ class PartitionGroup { errs_(static_cast(n_partitions)) { make_transport_(); discover_node_peers_(); - cpusets_ = topo_partition_cpusets(n_, node_rank_, node_size_); + cpusets_ = topo_partition_cpusets(n_, node_rank_, node_size_, node_mask_private_); start_masters_(); // The masters are already running, so a ctor throw must not escape: ~PartitionGroup would never run, // and destroying joinable threads during unwinding calls std::terminate. @@ -79,10 +81,11 @@ class PartitionGroup { parent_(src.parent_), node_rank_(src.node_rank_), node_size_(src.node_size_), + node_mask_private_(src.node_mask_private_), partitions_(static_cast(src.n_)), errs_(static_cast(src.n_)) { make_transport_(); - cpusets_ = topo_partition_cpusets(n_, node_rank_, node_size_); + cpusets_ = topo_partition_cpusets(n_, node_rank_, node_size_, node_mask_private_); start_masters_(); try { // see the primary ctor: a throw past live masters would std::terminate run_on_all([&](int r) { @@ -146,11 +149,12 @@ class PartitionGroup { } // Free-function wrapper so the header compiles on non-Linux (where partition_cpusets returns {}). - static auto topo_partition_cpusets(int n, int group_index, int group_count) + static auto topo_partition_cpusets(int n, int group_index, int group_count, bool mask_is_private) -> std::vector { return monoprop::detail::partition::partition_cpusets(static_cast(n), static_cast(group_index), - static_cast(group_count)); + static_cast(group_count), + mask_is_private); } // Under an MPI parent, find how many ranks share this host and which we are, so each co-located rank @@ -162,11 +166,31 @@ class PartitionGroup { MPI_Comm_split_type(parent_.mpi, MPI_COMM_TYPE_SHARED, 0, MPI_INFO_NULL, &node); MPI_Comm_rank(node, &node_rank_); MPI_Comm_size(node, &node_size_); + classify_node_masks_(node); MPI_Comm_free(&node); } #endif } +#ifdef monoprop_ENABLE_MPI + // A rank seeing 16 of 128 CPUs is equally "my own 16" and "eight of us share these 16": only the masks tell. + auto classify_node_masks_(MPI_Comm node) -> void { + node_mask_private_ = false; + if (node_size_ <= 1) { + return; // nobody to collide with; the normal split already handles group_count == 1 + } + constexpr size_t kMaskWords = monoprop::detail::partition::kAffinityMaskWords; + std::array mine{}; + // Refusal zeroes `mine` and an all-zero row is never private, so no reduction of the verdict is needed. + monoprop::detail::partition::affinity_mask_words(mine.data(), kMaskWords); + std::vector all(kMaskWords * static_cast(node_size_), 0); + MPI_Allgather(mine.data(), kMaskWords, MPI_UINT64_T, all.data(), kMaskWords, MPI_UINT64_T, node); + node_mask_private_ = monoprop::detail::partition::masks_are_pairwise_disjoint(all.data(), + static_cast(node_size_), + kMaskWords); + } +#endif + auto make_transport_() -> void { #ifdef monoprop_ENABLE_MPI if (parent_.kind == mpi::Comm::Kind::Mpi && mpi::size(parent_) > 1) { @@ -253,6 +277,7 @@ class PartitionGroup { mpi::Comm parent_; // enclosing communicator (size R) — decides the transport int node_rank_ = 0; // this rank's index among the ranks sharing the host int node_size_ = 1; // how many parent ranks share the host (1 unless MPI R>1) + bool node_mask_private_ = false; // set by classify_node_masks_; copied, never re-derived, by the copy ctor std::unique_ptr shm_; // set iff R == 1 #ifdef monoprop_ENABLE_MPI std::unique_ptr hyb_; // set iff R > 1 diff --git a/cpp/tests/cpu_topology_tests.cpp b/cpp/tests/cpu_topology_tests.cpp index 5cda9980..93dd165c 100644 --- a/cpp/tests/cpu_topology_tests.cpp +++ b/cpp/tests/cpu_topology_tests.cpp @@ -25,6 +25,7 @@ #include #include +#include #include #include @@ -32,6 +33,7 @@ #include #endif +#include "monoprop/detail/EnvConfig.h" // config::get().partition_pinning -- the one licensed empty placement #include "monoprop/detail/partition/CpuTopology.h" namespace partition = monoprop::detail::partition; @@ -47,6 +49,19 @@ struct AffinityGuard { #endif }; +namespace { + +// An empty placement is licensed by pinning being off and by nothing else; "placed nothing" is the bug. +auto empty_placement_is_licensed() -> bool { + if (monoprop::config::get().partition_pinning) { + return false; + } + BOOST_TEST_MESSAGE("monoprop_PARTITION_PINNING is off; partition_cpusets places nothing"); + return true; +} + +} // namespace + /* ── Live smoke tests ─────────────────────────────────────────────────────── */ BOOST_AUTO_TEST_CASE(cpu_topology_enumerate_and_place) { @@ -62,7 +77,7 @@ BOOST_AUTO_TEST_CASE(cpu_topology_enumerate_and_place) { // guard restores affinity on scope exit } // When topology discovery succeeds, a non-empty core list must produce a non-empty placement. - if (!cores.empty()) { + if (!cores.empty() && !(one.empty() && empty_placement_is_licensed())) { BOOST_CHECK_EQUAL(one.size(), 1u); } @@ -93,9 +108,33 @@ BOOST_AUTO_TEST_CASE(cpu_topology_place_co_located_ranks) { // invariant (one rank's busy-polling collectives cannot starve the other's barrier spins). BOOST_CHECK(rank0.front().pu != rank1.front().pu); - // Oversubscription: 2 ranks × cores.size() partitions > total physical cores. - const auto past_end = partition::partition_cpusets(/*n=*/cores.size(), /*group_index=*/1, /*group_count=*/2); - BOOST_CHECK(past_end.empty()); + // Both arms passed explicitly so neither depends on the host: refuse the shared one, fill the private. + const auto shared_mask = partition::partition_cpusets(/*n=*/cores.size(), + /*group_index=*/1, + /*group_count=*/2, + /*mask_is_private=*/false); + BOOST_CHECK(shared_mask.empty()); + + const auto private_mask = partition::partition_cpusets(/*n=*/cores.size(), + /*group_index=*/1, + /*group_count=*/2, + /*mask_is_private=*/true); + if (private_mask.empty() && empty_placement_is_licensed()) { + return; + } + BOOST_REQUIRE_EQUAL(private_mask.size(), cores.size()); + std::set placed; + for (const auto &set : private_mask) { + placed.insert(set.pu); + } + BOOST_CHECK_EQUAL(placed.size(), cores.size()); + std::set visible; + for (const auto &core : cores) { + visible.insert(core.cpu); + } + for (const int pu : placed) { + BOOST_CHECK(visible.count(pu) == 1); + } } /* ── Policy unit tests (deterministic, no hwloc or live hardware) ─────────── */ @@ -173,3 +212,211 @@ BOOST_AUTO_TEST_CASE(cpu_topology_policy_uneven_domains) { BOOST_CHECK_EQUAL(order[1], 2); BOOST_CHECK_EQUAL(order[2], 4); } + +/* ── The cgroup-placement classification ──────────────────────────────────── */ + +BOOST_AUTO_TEST_CASE(cpu_topology_policy_per_rank_slice_starves_without_collapse) { + // One rank's slice under `srun --cpu-bind=cores`: 2 cores of a 16-core host, one L3 domain. + const std::vector slice = {{6, 0}, {7, 0}}; + + BOOST_CHECK(placement_order(slice, 2, /*group_index=*/3, /*group_count=*/8).empty()); + + // Collapsed to a single group -- what mask_is_private does -- the same slice places fully. + const auto collapsed = placement_order(slice, 2, /*group_index=*/0, /*group_count=*/1); + BOOST_REQUIRE_EQUAL(collapsed.size(), 2u); + BOOST_CHECK_EQUAL(collapsed[0], 6); + BOOST_CHECK_EQUAL(collapsed[1], 7); +} + +BOOST_AUTO_TEST_CASE(cpu_topology_policy_private_mask_collapses_even_when_the_split_would_fit) { + // 2 ranks x 2 partitions fits these 4 cores, so the collapse is not a fallback: it moves rank 1's cores. + const std::vector cores = {{0, 0}, {2, 1}, {4, 0}, {6, 1}}; + + const auto split = placement_order(cores, 2, /*group_index=*/1, /*group_count=*/2); + BOOST_REQUIRE_EQUAL(split.size(), 2u); + BOOST_CHECK_EQUAL(split[0], 2); + BOOST_CHECK_EQUAL(split[1], 6); + + // What mask_is_private now passes: the head of this rank's own interleave over both domains. + const auto collapsed = placement_order(cores, 2, /*group_index=*/0, /*group_count=*/1); + BOOST_REQUIRE_EQUAL(collapsed.size(), 2u); + BOOST_CHECK_EQUAL(collapsed[0], 0); + BOOST_CHECK_EQUAL(collapsed[1], 2); +} + +namespace { + +// The flat [n * words] array MPI_Allgather leaves behind, built from per-rank PU-index lists. +auto packed_masks(const std::vector> &pus, size_t words) -> std::vector { + std::vector out(pus.size() * words, 0); + for (size_t r = 0; r < pus.size(); ++r) { + for (const size_t pu : pus[r]) { + out[(r * words) + (pu / 64)] |= uint64_t{1} << (pu % 64); + } + } + return out; +} + +} // namespace + +BOOST_AUTO_TEST_CASE(cpu_topology_masks_disjoint_vs_identical) { + constexpr size_t kWords = partition::kAffinityMaskWords; + + const auto disjoint = packed_masks({{0, 1}, {2, 3}}, kWords); + BOOST_CHECK(partition::masks_are_pairwise_disjoint(disjoint.data(), 2, kWords)); + + const auto identical = packed_masks({{0, 1}, {0, 1}}, kWords); + BOOST_CHECK(!partition::masks_are_pairwise_disjoint(identical.data(), 2, kWords)); + + // Partial overlap: "not private" is conservative, since collapsing points every rank at the same cores. + const auto partial = packed_masks({{0, 1}, {1, 2}}, kWords); + BOOST_CHECK(!partition::masks_are_pairwise_disjoint(partial.data(), 2, kWords)); + + // An unreadable mask arrives empty and must not be read as "disjoint from everything". + const auto with_empty = packed_masks({{0, 1}, {}}, kWords); + BOOST_CHECK(!partition::masks_are_pairwise_disjoint(with_empty.data(), 2, kWords)); + + const auto lone = packed_masks({{0, 1}}, kWords); + BOOST_CHECK(!partition::masks_are_pairwise_disjoint(lone.data(), 1, kWords)); + + const auto four_ok = packed_masks({{0}, {1}, {2}, {3}}, kWords); + BOOST_CHECK(partition::masks_are_pairwise_disjoint(four_ok.data(), 4, kWords)); + const auto four_bad = packed_masks({{0}, {1}, {2}, {1}}, kWords); + BOOST_CHECK(!partition::masks_are_pairwise_disjoint(four_bad.data(), 4, kWords)); +} + +BOOST_AUTO_TEST_CASE(cpu_topology_masks_span_word_boundaries) { + constexpr size_t kWords = partition::kAffinityMaskWords; + + // A per-word comparison that forgot to loop would answer from word 0 alone. + const auto low_high = packed_masks({{5}, {200}}, kWords); + BOOST_CHECK(partition::masks_are_pairwise_disjoint(low_high.data(), 2, kWords)); + + const auto both_high = packed_masks({{200}, {200}}, kWords); + BOOST_CHECK(!partition::masks_are_pairwise_disjoint(both_high.data(), 2, kWords)); + + const auto late_overlap = packed_masks({{1, 3000}, {2, 3000}}, kWords); + BOOST_CHECK(!partition::masks_are_pairwise_disjoint(late_overlap.data(), 2, kWords)); +} + +BOOST_AUTO_TEST_CASE(cpu_topology_affinity_mask_covers_enumerated_cores) { + // The only check that the mask EXCHANGED and the cores PLACED come from one view of the machine. + const auto cores = partition::enumerate_physical_cores(); + if (cores.empty()) { + return; // hwloc loaded no topology at all: there is no second view to agree with. + } + std::vector mine(partition::kAffinityMaskWords, 0); + if (!partition::affinity_mask_words(mine.data(), mine.size())) { + // Not a skip: refusal keys on the HIGHEST allowed PU, and core.cpu is each core's LOWEST sibling. + std::vector wide(partition::kAffinityMaskWords * 64, 0); + BOOST_REQUIRE(partition::affinity_mask_words(wide.data(), wide.size())); + size_t highest = 0; + for (size_t w = wide.size(); w-- > 0;) { + if (wide[w] != 0) { + highest = (w * 64) + static_cast(63 - __builtin_clzll(wide[w])); + break; + } + } + BOOST_CHECK_GE(highest, partition::kAffinityMaskWords * 64); + return; + } + size_t set_bits = 0; + for (const uint64_t w : mine) { + set_bits += static_cast(__builtin_popcountll(w)); + } + BOOST_CHECK(set_bits > 0u); + for (const auto &core : cores) { + const auto pu = static_cast(core.cpu); + BOOST_CHECK((mine[pu / 64] >> (pu % 64)) & 1U); + } +} + +#if defined(__linux__) + +namespace { + +// Confine to the first `k` cores, asserting the narrowing took hold; false only if the kernel refused. +auto confine_to_first(const std::vector &full, size_t k) -> bool { + cpu_set_t mask; + CPU_ZERO(&mask); + for (size_t i = 0; i < k; ++i) { + CPU_SET(full[i].cpu, &mask); + } + if (sched_setaffinity(0, sizeof(mask), &mask) != 0) { + return false; + } + BOOST_REQUIRE_EQUAL(partition::enumerate_physical_cores().size(), k); + return true; +} + +} // namespace + +BOOST_AUTO_TEST_CASE(cpu_topology_per_rank_mask_still_places) { + const auto full = partition::enumerate_physical_cores(); + if (full.empty()) { + return; // hwloc loaded no topology: there is no mask to confine to. + } + // ONE core exercises the collapse, so a single-core runner does not turn this into a free pass. + const size_t k = std::min(full.size(), 2); + + const AffinityGuard guard; + if (!confine_to_first(full, k)) { + return; // the kernel refused the affinity call; nothing below is reachable + } + + // What `srun --cpu-bind=cores` produces: our whole share, told there are eight sibling ranks. + const auto sets = + partition::partition_cpusets(/*n=*/k, /*group_index=*/3, /*group_count=*/8, /*mask_is_private=*/true); + if (sets.empty() && empty_placement_is_licensed()) { + return; + } + BOOST_REQUIRE_EQUAL(sets.size(), k); + for (const auto &set : sets) { + // Never pin outside the mask the launcher gave us. + bool inside = false; + for (size_t i = 0; i < k; ++i) { + inside = inside || set.pu == full[i].cpu; + } + BOOST_CHECK(inside); + } + if (k == 2) { + BOOST_CHECK(sets[0].pu != sets[1].pu); + } +} + +BOOST_AUTO_TEST_CASE(cpu_topology_shared_mask_keeps_co_located_ranks_disjoint) { + const auto full = partition::enumerate_physical_cores(); + if (full.size() < 2) { + return; // two ranks cannot hold disjoint cores when there is only one + } + // Two cores, one per rank: a two-physical-core CI runner cannot supply the four the port asked for. + const size_t per_rank = full.size() >= 4 ? 2 : 1; + const size_t shared = 2 * per_rank; + + const AffinityGuard guard; + if (!confine_to_first(full, shared)) { + return; // the kernel refused the affinity call; nothing below is reachable + } + + const auto rank0 = partition::partition_cpusets(/*n=*/per_rank, + /*group_index=*/0, + /*group_count=*/2, + /*mask_is_private=*/false); + const auto rank1 = partition::partition_cpusets(/*n=*/per_rank, + /*group_index=*/1, + /*group_count=*/2, + /*mask_is_private=*/false); + if (rank0.empty() && rank1.empty() && empty_placement_is_licensed()) { + return; + } + BOOST_REQUIRE_EQUAL(rank0.size(), per_rank); + BOOST_REQUIRE_EQUAL(rank1.size(), per_rank); + for (const auto &a : rank0) { + for (const auto &b : rank1) { + // Two ranks on one core starve each other: busy-polling collectives against barrier spins. + BOOST_CHECK(a.pu != b.pu); + } + } +} + +#endif // __linux__