From 13eb46234b01ad0e552982f1822a4f680da7da18 Mon Sep 17 00:00:00 2001 From: Aaron Miller Date: Wed, 29 Jul 2026 01:23:44 +0200 Subject: [PATCH 1/3] =?UTF-8?q?feat(mpi):=20=F0=9F=93=8A=20opt-in=20per-pa?= =?UTF-8?q?rtition=20profile=20for=20the=20partitioned=20collectives?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit monoprop_COMM_PROFILE=1 accumulates, per partition and per cache-line-isolated slot, where a collective's wall time goes: table_p0 (fills one partition runs alone) vs table_par (fills every partition runs on its own slice) vs table_move (payload memcpy) vs mpi vs barrier wait, plus the barrier's L3-domain group count. Off by default and never allocated then, so the hot path pays one null check per instrumented region. The splits are the ones that decide protocol questions. table_p0 vs table_par separates a serial protocol from a parallel one, and attributing barrier wait per partition is what makes the asymmetry visible -- the master's own wait stays small precisely when the master is the bottleneck. table_par vs table_move separates bookkeeping, which a better protocol shrinks, from data movement, which it cannot. Assisted-by: ClaudeCode:claude-opus-5 --- src/monoprop/detail/EnvConfig.h | 3 + src/monoprop/detail/mpi/CommProfile.h | 134 ++++++++++++++++++++++++++ 2 files changed, 137 insertions(+) create mode 100644 src/monoprop/detail/mpi/CommProfile.h diff --git a/src/monoprop/detail/EnvConfig.h b/src/monoprop/detail/EnvConfig.h index 5fae134a..67f4fe40 100644 --- a/src/monoprop/detail/EnvConfig.h +++ b/src/monoprop/detail/EnvConfig.h @@ -24,6 +24,7 @@ // monoprop_NUM_THREADS positive int (1..1e6), else ignored → num_threads // monoprop_PARTITION_PINNING bool, default ON; 0/false disables per-core pinning → partition_pinning // monoprop_PARTITIONS int N | "auto" | "off"; parsed where it is used (resolve_partition_count_) +// monoprop_COMM_PROFILE bool, default OFF; per-partition collective accounting to stderr → comm_profile namespace monoprop::config { @@ -57,6 +58,7 @@ inline auto parse_positive_int(const char *text) -> std::optional { struct Settings { std::optional num_threads; bool partition_pinning = true; + bool comm_profile = false; }; // Parse the environment once; the Settings are cached and shared across TUs. @@ -65,6 +67,7 @@ inline auto get() -> const Settings & { Settings s; s.num_threads = detail::parse_positive_int(std::getenv("monoprop_NUM_THREADS")); s.partition_pinning = detail::parse_flag(std::getenv("monoprop_PARTITION_PINNING"), true); + s.comm_profile = detail::parse_flag(std::getenv("monoprop_COMM_PROFILE"), false); return s; }(); return settings; diff --git a/src/monoprop/detail/mpi/CommProfile.h b/src/monoprop/detail/mpi/CommProfile.h new file mode 100644 index 00000000..1575a04a --- /dev/null +++ b/src/monoprop/detail/mpi/CommProfile.h @@ -0,0 +1,134 @@ +// 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. + +#pragma once + +#include +#include +#include +#include +#include +#include + +// Opt-in accounting for where a partitioned collective's wall time goes (monoprop_COMM_PROFILE=1). Off by +// default and never allocated then, so the hot path pays one null check per instrumented region. +// +// The split that matters is table_p0 vs table_par: the offset/count tables a HybridComm collective +// rebuilds are O(R*S^2), and whether one partition fills them alone (table_p0, the other S-1 parked in a +// barrier) or every partition fills its own slice (table_par) is the difference between a serial and a +// parallel protocol. Barrier wait is attributed per partition to make that asymmetry visible. table_move +// separates payload memcpy, which a better protocol does not shrink, from bookkeeping, which it does. + +namespace monoprop::mpi { + +class CommProfile { +public: + using Clock = std::chrono::steady_clock; + + // One cache-line-isolated accumulator per partition: every counter is written only by its own + // partition thread, so profiling adds no coherence traffic of its own to what it measures. + struct alignas(64) Slot { + uint64_t barrier_ns = 0; + uint64_t table_p0_ns = 0; // fills executed under `if (local_partition == 0)` + uint64_t table_par_ns = 0; // fills every partition runs on its own slice + uint64_t table_move_ns = 0; // payload memcpy (pack into / scatter out of staging), not bookkeeping + uint64_t mpi_ns = 0; // time inside MPI itself + uint64_t n_barriers = 0; + uint64_t n_verbs = 0; + }; + + explicit CommProfile(int n_partitions, int mpi_rank) + : slots_(static_cast(n_partitions)), + mpi_rank_(mpi_rank) {} + + // L3 domains the transport's barrier grouped its partitions into (< 2 ⇒ the flat barrier ran). + int barrier_groups = 0; + + auto slot(int partition) -> Slot & { return slots_[static_cast(partition)]; } + + // Aggregate over partitions, plus partition 0 broken out. Written to stderr at teardown, one + // block per MPI rank; a driver greps `COMMPROF` and sums/compares across ranks. + // + // noexcept because the only caller is a transport destructor, which may run while an exception + // unwinds: a diagnostic print that fails must be dropped, never escalated to a terminate. + auto dump() const noexcept -> void { + try { + dump_(); + } + catch (...) { // std::print can throw (formatting, or a write error on stderr) + } + } + +private: + auto dump_() const -> void { + Slot total; + for (const Slot &s : slots_) { + total.barrier_ns += s.barrier_ns; + total.table_p0_ns += s.table_p0_ns; + total.table_par_ns += s.table_par_ns; + total.table_move_ns += s.table_move_ns; + total.mpi_ns += s.mpi_ns; + total.n_barriers += s.n_barriers; + total.n_verbs += s.n_verbs; + } + const Slot &p0 = slots_.front(); + // Peer barrier wait is the cost of the master's serial phases seen from the other side. + const uint64_t peer_barrier_ns = total.barrier_ns - p0.barrier_ns; + std::print(stderr, + "COMMPROF rank={} partitions={} barrier_groups={} verbs={} barriers={} " + "table_p0_s={:.3f} table_par_s={:.3f} table_move_s={:.3f} mpi_s={:.3f} " + "barrier_p0_s={:.3f} barrier_peers_s={:.3f} barrier_per_sync_us={:.2f}\n", + mpi_rank_, + slots_.size(), + barrier_groups, + p0.n_verbs, + p0.n_barriers, + to_s(p0.table_p0_ns), + to_s(total.table_par_ns) / static_cast(slots_.size()), + to_s(total.table_move_ns) / static_cast(slots_.size()), + to_s(p0.mpi_ns), + to_s(p0.barrier_ns), + to_s(peer_barrier_ns) / static_cast(slots_.size() > 1 ? slots_.size() - 1 : 1), + total.n_barriers == 0 + ? 0.0 + : (static_cast(total.barrier_ns) / static_cast(total.n_barriers)) / 1000.0); + std::fflush(stderr); + } + + static auto to_s(uint64_t ns) -> double { return static_cast(ns) / 1e9; } + + std::vector slots_; + int mpi_rank_; +}; + +// Adds the scope's duration to one counter. Held by value in the instrumented region; `target` must +// outlive it (it is a field of the comm's own CommProfile). +class ScopedNs { +public: + explicit ScopedNs(uint64_t *target) : target_(target), start_(CommProfile::Clock::now()) {} + ScopedNs(const ScopedNs &) = delete; + auto operator=(const ScopedNs &) -> ScopedNs & = delete; + ~ScopedNs() { + if (target_ != nullptr) { + *target_ += static_cast( + std::chrono::duration_cast(CommProfile::Clock::now() - start_).count()); + } + } + +private: + uint64_t *target_; + CommProfile::Clock::time_point start_; +}; + +} // namespace monoprop::mpi From fef1f051fcea9cd183d717ee486b9204dbebfb95 Mon Sep 17 00:00:00 2001 From: Aaron Miller Date: Wed, 29 Jul 2026 01:24:03 +0200 Subject: [PATCH 2/3] =?UTF-8?q?perf(mpi):=20=E2=9A=A1=EF=B8=8F=20remove=20?= =?UTF-8?q?the=20serial=20O(R*S^2)=20floor=20from=20HybridComm?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A partitioned multi-node run was protocol-bound, not network-bound: every collective rebuilt O(R*S^2) integer offset tables and partition 0 filled all of them alone while the other S-1 partitions spun at a barrier. Measured with monoprop_COMM_PROFILE=1 (2 nodes, 512-site Hubbard, 4 layers, R=2, S=112), partition 0's serial fill was 22.33 s of a 26.74 s run, while MPI itself was 1.58 s -- the wire was never the problem. Four changes, wall 26.74 -> 4.46 s (6.0x), expectation value bit-identical at every step and at every rank x partition split: - Parallel prefix. The offset of block (rank b, dest t, source u) splits into a per-(b,t) base needing global knowledge, which is only O(R*S) and stays on partition 0, and a scan over u that partition t owns outright. The count matrix is relaid source-partition-major so each partition writes a contiguous run; dest-major would put S partitions on every cache line and trade a serial fill for pure false sharing. - alltoallv_reverse. The answer leg travels the query exchange's legs backwards, so its geometry is the query round's, and rebuilding it costs 3*R*S^2 entries and a barrier for nothing. It is a ratio and not an identity: the query leg carries Sink::kStride elements per record and the answer leg one, so every reused offset and per-rank count is divided by the stride (exact -- every forward count is a multiple of it). Reusing them undivided would still deliver correct data while staging and transmitting kStride times the bytes. - Empty-block veto. Each partition publishes a cache-line-padded bitmask of the ranks it sends anything to; the sizing phase then runs source-partition-outer, so a partition that sends nothing costs one load instead of R strided probes into its count array. At early layers nearly every block is empty. - Two-level barrier. PartitionBarrier fans in within an L3 domain and then across domains, so both the arrival fetch_add and the release-store invalidation cost O(S/G) coherence transactions inside one L3 slice instead of O(S) across the socket interconnect. Domains are derived from the partition cpusets rather than from the placement logic, so the two cannot drift apart, and a rank spanning one domain keeps the flat barrier (a root barrier of one is pure overhead). Tests: the pre-existing hybrid_comm cases all sent the same count to every destination per source, so a transposed (b,t,u) index passed unnoticed -- two asymmetric-count cases close that. partition_barrier_tests.cpp tests the two-level path directly, because engaging it through a real comm needs pinning and >=2 L3 domains, which no test host can be relied on to have. Both additions are mutation-verified: an inconsistent receiver index and a skipped root barrier each fail the new cases while the old ones pass. Assisted-by: ClaudeCode:claude-opus-5 --- .../detail/evolution/layer_build/Engine.h | 11 +- src/monoprop/detail/mpi/HybridComm.h | 599 ++++++++++++++---- src/monoprop/detail/mpi/MPICompat.h | 41 +- src/monoprop/detail/mpi/PartitionBarrier.h | 123 +++- src/monoprop/detail/mpi/ShmComm.h | 162 +++-- src/monoprop/detail/partition/CpuTopology.h | 29 + .../detail/partition/PartitionGroup.h | 13 +- tests/cpp/hybrid_comm_tests.cpp | 128 ++++ tests/cpp/partition_barrier_tests.cpp | 161 +++++ 9 files changed, 1055 insertions(+), 212 deletions(-) create mode 100644 tests/cpp/partition_barrier_tests.cpp diff --git a/src/monoprop/detail/evolution/layer_build/Engine.h b/src/monoprop/detail/evolution/layer_build/Engine.h index 63335a35..ce6e7dc3 100644 --- a/src/monoprop/detail/evolution/layer_build/Engine.h +++ b/src/monoprop/detail/evolution/layer_build/Engine.h @@ -381,7 +381,16 @@ struct LayerBuildEngine { auto resp = resolve_incoming(inc_q, local_op, R, is_leader_pass, matched, combined_size, sink); std::vector resp_recv = response_recv_counts(); std::vector> inc_r; - mpi::begin_alltoallv(resp, comm, /*skip_self=*/false, &resp_recv).wait_into(inc_r); + // The answers travel the query exchange's legs backwards, one per query, so the hybrid transport + // reuses that exchange's offset tables; nothing may collectively intervene between the two calls. + // Sink::kStride is the query leg's words per query, the ratio between the two legs' counts. + mpi::begin_alltoallv(resp, + comm, + /*skip_self=*/false, + &resp_recv, + /*reverse_of_previous=*/true, + /*forward_stride=*/static_cast(Sink::kStride)) + .wait_into(inc_r); process_responses(inc_r, src_idx_r, queries_r, R, my_rank, sink); } diff --git a/src/monoprop/detail/mpi/HybridComm.h b/src/monoprop/detail/mpi/HybridComm.h index 71a61a38..a01929d8 100644 --- a/src/monoprop/detail/mpi/HybridComm.h +++ b/src/monoprop/detail/mpi/HybridComm.h @@ -22,6 +22,7 @@ #include #include #include +#include #include #include #include @@ -29,6 +30,8 @@ #include +#include "monoprop/detail/EnvConfig.h" +#include "monoprop/detail/mpi/CommProfile.h" #include "monoprop/detail/mpi/PartitionBarrier.h" // Composes R MPI ranks x S in-process partitions into one flat P=R*S SPMD world. Global id is rank-major @@ -42,11 +45,13 @@ namespace monoprop::mpi { class HybridComm { public: // n_local_partitions = S, identical on every rank (the facade ctor checks that before constructing). - HybridComm(MPI_Comm parent, int n_local_partitions) + // partition_l3_domains (optional, one entry per partition) makes the barrier two-level; see + // PartitionBarrier. + HybridComm(MPI_Comm parent, int n_local_partitions, const std::vector &partition_l3_domains = {}) : parent_(parent), s_(n_local_partitions), slots_(static_cast(n_local_partitions)), - barrier_(n_local_partitions) { + barrier_(n_local_partitions, partition_l3_domains) { MPI_Comm_size(parent_, &r_); MPI_Comm_rank(parent_, &mpi_rank_); int provided = MPI_THREAD_SINGLE; @@ -66,11 +71,35 @@ class HybridComm { mpi_recv_displs_.resize(static_cast(r_)); pack_off_.resize(rss); scatter_off_.resize(rss); + const size_t rs = static_cast(r_) * static_cast(s_); + col_send_.resize(rs); + col_recv_.resize(rs); + tpre_send_.resize(rs); + tpre_recv_.resize(rs); + rev_send_counts_.resize(static_cast(r_)); + rev_send_displs_.resize(static_cast(r_)); + rev_recv_counts_.resize(static_cast(r_)); + rev_recv_displs_.resize(static_cast(r_)); + mask_words_ = pad_to_line_(static_cast((r_ + 63) / 64)); + send_mask_.assign(static_cast(s_) * mask_words_, 0); + run_stride_ = pad_to_line_(static_cast(r_)); + run_scratch_.assign(static_cast(s_) * run_stride_, 0); + if (config::get().comm_profile) { + prof_ = std::make_unique(s_, mpi_rank_); + prof_->barrier_groups = barrier_.group_count(); + } } HybridComm(const HybridComm &) = delete; auto operator=(const HybridComm &) -> HybridComm & = delete; + // Profiling is opt-in, so the common case destroys nothing and prints nothing. + ~HybridComm() { + if (prof_ != nullptr) { + prof_->dump(); + } + } + auto size() const -> int { return r_ * s_; } auto global_rank(int local_partition) const -> int { return mpi_rank_ * s_ + local_partition; } @@ -137,6 +166,32 @@ class HybridComm { }); } + // Payload-only reverse of the immediately preceding alltoallv_resolve: same legs, opposite + // direction, one element back per element received. See alltoallv_reverse_impl_ for the contract. + auto alltoallv_reverse(int local_partition, + const void *send, + const int *send_counts /*[P]*/, + const int *send_displs /*[P]*/, + void *recv, + const int *recv_counts /*[P]*/, + const int *recv_displs /*[P]*/, + size_t elem, + MPI_Datatype dt, + int forward_stride) -> void { + guard_partition0_(local_partition, "alltoallv_reverse", [&] { + alltoallv_reverse_impl_(local_partition, + send, + send_counts, + send_displs, + recv, + recv_counts, + recv_displs, + elem, + dt, + forward_stride); + }); + } + template auto allreduce_sum(int local_partition, T local_val) -> T { return guard_partition0_(local_partition, "allreduce_sum", [&] { @@ -150,34 +205,32 @@ class HybridComm { }); } - // recv_counts[g] = amount global partition g sends to this partition. 2 barriers + one S*S-int MPI_Alltoall. + // recv_counts[g] = amount global partition g sends to this partition. 2 barriers + one S*S-int + // MPI_Alltoall; the count matrix is filled row-per-partition before the first barrier. auto alltoall_counts_impl_(int local_partition, const int *send_counts /*[P]*/, int *recv_counts /*[P]*/) -> void { - const size_t u = static_cast(local_partition); - slots_[u].counts = send_counts; - sync(); + count_verb(local_partition); if (local_partition == 0) { - // Pack the S*S count matrix per dest rank, dest-partition-major (t) then source-partition-minor (su). - // Every element is written here and MPI_Alltoall fills counts_recv_ fully, so neither is pre-zeroed. - for (int b = 0; b < r_; ++b) { - for (int t = 0; t < s_; ++t) { - for (int su = 0; su < s_; ++su) { - const size_t idx = ((static_cast(b) * static_cast(s_)) + static_cast(t)) - * static_cast(s_) - + static_cast(su); - counts_send_[idx] = slots_[static_cast(su)].counts[b * s_ + t]; - } - } - } + ++tables_gen_; + } + // Each partition writes its OWN source row of the count matrix before the barrier that publishes + // it; the rows are disjoint, so this costs no extra barrier. Every element is written here and + // MPI_Alltoall fills counts_recv_ fully, so neither buffer is pre-zeroed. + { + ScopedNs timer{par_table_ns(local_partition)}; + fill_count_row_(local_partition, send_counts); + } + sync(local_partition); + if (local_partition == 0) { + ScopedNs timer{mpi_ns(local_partition)}; MPI_Alltoall(counts_send_.data(), s_ * s_, MPI_INT, counts_recv_.data(), s_ * s_, MPI_INT, parent_); } - sync(); - // Partition t extracts its row: recv from (rank a, partition su) is contiguous per source rank a. + sync(local_partition); + // Partition t extracts its column: what (rank a, partition su) sends to it. const int t = local_partition; + ScopedNs timer{par_table_ns(local_partition)}; for (int a = 0; a < r_; ++a) { for (int su = 0; su < s_; ++su) { - const size_t idx = (static_cast(a) * static_cast(s_) * static_cast(s_)) - + (static_cast(t) * static_cast(s_)) + static_cast(su); - recv_counts[a * s_ + su] = counts_recv_[idx]; + recv_counts[a * s_ + su] = counts_recv_[cnt_idx_(a, su, t)]; } } // No trailing barrier: past the last sync only counts_recv_ is read, and partition 0 cannot rewrite @@ -197,25 +250,36 @@ class HybridComm { size_t elem, MPI_Datatype dt) -> void { const size_t u = static_cast(local_partition); + count_verb(local_partition); + if (local_partition == 0) { + ++tables_gen_; + } Slot &me = slots_[u]; me.ptr = send; me.send_counts = send_counts; me.send_displs = send_displs; me.recv_counts = recv_counts; - sync(); // B1 - - // B2: partition 0 sizes/reallocates staging; must finish before any partition packs into stage_send_. - if (local_partition == 0) { - size_staging_(elem); + // No count matrix on this path, so the veto row phase A reads is filled on its own. + { + ScopedNs timer{par_table_ns(local_partition)}; + fill_send_mask_(local_partition, send_counts); } - sync(); // B2 + sync(local_partition); // B1 - // B3: each partition packs its own cross-rank blocks into stage_send_ (disjoint writes). - pack_send_(local_partition, elem); - sync(); // B3 + // Sizing is S-way parallel and carries its own two barriers (see size_staging_parallel_); on + // return every base is visible and staging is grown, so packing into stage_send_ is safe. + size_staging_parallel_(local_partition, elem); + + // Each partition packs its own cross-rank blocks into stage_send_ (disjoint writes). + { + ScopedNs timer{move_ns(local_partition)}; + pack_send_(local_partition, elem); + } + sync(local_partition); // B3 - // B4: partition 0 runs the single MPI_Alltoallv while peers park at the barrier. + // Partition 0 runs the single MPI_Alltoallv while peers park at the barrier. if (local_partition == 0) { + ScopedNs timer{mpi_ns(local_partition)}; MPI_Alltoallv(stage_send_.data(), mpi_send_counts_.data(), mpi_send_displs_.data(), @@ -226,10 +290,12 @@ class HybridComm { dt, parent_); } - sync(); // B4 + sync(local_partition); // B4 // Scatter each global source's contiguous run from stage_recv_ to recv_displs[g] (all legs, incl. - // self-rank, go through staging). Block starts come from scatter_off_: no peer slot is read past B4. + // self-rank, go through staging). Block starts come from the offset tables, so no peer slot is + // read after the last barrier. + ScopedNs timer{move_ns(local_partition)}; char *dst = static_cast(recv); const int t = local_partition; for (int a = 0; a < r_; ++a) { @@ -238,12 +304,12 @@ class HybridComm { const int cnt = recv_counts[g]; if (cnt != 0) { std::memcpy(dst + static_cast(recv_displs[g]) * elem, - stage_recv_.data() + scatter_off_[block_idx_(a, t, su)] * elem, + stage_recv_.data() + recv_block_off_(a, t, su) * elem, static_cast(cnt) * elem); } } } - // No trailing barrier (see alltoall_counts_impl_): past B4 only stage_recv_/scatter_off_/own buffers. + // No trailing barrier (see alltoall_counts_impl_): past B4 only stage_recv_/offset tables/own buffers. } // Fused count-resolve + payload alltoallv: folds the standalone count exchange into this verb's @@ -260,49 +326,57 @@ class HybridComm { size_t elem, MPI_Datatype dt) -> void { const size_t u = static_cast(local_partition); + count_verb(local_partition); + if (local_partition == 0) { + ++tables_gen_; + } Slot &me = slots_[u]; me.ptr = send; me.send_counts = send_counts; me.send_displs = send_displs; - // recv_counts is an output here — deliberately not published; the count Alltoall resolves it. - sync(); // B1 + // recv_counts is an output here — deliberately not published; the count Alltoall resolves it. The + // count row comes from the local argument, so it rides the same barrier as the slot publish. + { + ScopedNs timer{par_table_ns(local_partition)}; + fill_count_row_(local_partition, send_counts); + } + sync(local_partition); // B1: slots and the whole send-count matrix published if (local_partition == 0) { - for (int b = 0; b < r_; ++b) { - for (int t = 0; t < s_; ++t) { - for (int su = 0; su < s_; ++su) { - counts_send_[block_idx_(b, t, su)] = slots_[static_cast(su)].send_counts[b * s_ + t]; - } - } - } + ScopedNs timer{mpi_ns(local_partition)}; MPI_Alltoall(counts_send_.data(), s_ * s_, MPI_INT, counts_recv_.data(), s_ * s_, MPI_INT, parent_); - // Size staging from counts_recv_: recv of partition t from (rank, su) sits at rank*S*S + t*S + su. - size_staging_impl_(elem, [this](int t, int rank, int su) { - return counts_recv_[(static_cast(rank) * static_cast(s_) * static_cast(s_)) - + (static_cast(t) * static_cast(s_)) + static_cast(su)]; - }); } - sync(); // B2 + sync(local_partition); // B2: counts_recv_ visible to every partition + + // Size staging from counts_recv_: what partition t gets from (rank, su) is at cnt_idx_(rank, su, t). + // Carries its own two barriers and its own timers — do not wrap it in one. + size_staging_parallel_(local_partition, elem, [this](int t, int rank, int su) { + return counts_recv_[cnt_idx_(rank, su, t)]; + }); const int t = local_partition; - long long total = 0; - for (int a = 0; a < r_; ++a) { - for (int su = 0; su < s_; ++su) { - const int g = a * s_ + su; - const int c = - counts_recv_[(static_cast(a) * static_cast(s_) * static_cast(s_)) - + (static_cast(t) * static_cast(s_)) + static_cast(su)]; - recv_counts[g] = c; - recv_displs[g] = checked_int_(total); - total += c; + { + ScopedNs timer{par_table_ns(local_partition)}; + long long total = 0; + for (int a = 0; a < r_; ++a) { + for (int su = 0; su < s_; ++su) { + const int g = a * s_ + su; + const int c = counts_recv_[cnt_idx_(a, su, t)]; + recv_counts[g] = c; + recv_displs[g] = checked_int_(total); + total += c; + } } + recv.resize(static_cast(checked_int_(total))); } - recv.resize(static_cast(checked_int_(total))); - - pack_send_(local_partition, elem); - sync(); // B3 + { + ScopedNs timer{move_ns(local_partition)}; + pack_send_(local_partition, elem); + } + sync(local_partition); // B3 if (local_partition == 0) { + ScopedNs timer{mpi_ns(local_partition)}; MPI_Alltoallv(stage_send_.data(), mpi_send_counts_.data(), mpi_send_displs_.data(), @@ -313,8 +387,9 @@ class HybridComm { dt, parent_); } - sync(); // B4 + sync(local_partition); // B4 + ScopedNs scatter_timer{move_ns(local_partition)}; char *dst = reinterpret_cast(recv.data()); for (int a = 0; a < r_; ++a) { for (int su = 0; su < s_; ++su) { @@ -322,16 +397,129 @@ class HybridComm { const int cnt = recv_counts[g]; if (cnt != 0) { std::memcpy(dst + static_cast(recv_displs[g]) * elem, - stage_recv_.data() + scatter_off_[block_idx_(a, t, su)] * elem, + stage_recv_.data() + recv_block_off_(a, t, su) * elem, static_cast(cnt) * elem); } } } + if (local_partition == 0) { + reverse_ready_gen_ = tables_gen_; // this round's tables may be reversed exactly once + } // No trailing barrier: same discipline as alltoallv_impl_. } + // Reverses the immediately preceding alltoallv_resolve: every leg carries one element back per RECORD + // it delivered, so this round's geometry is that round's with the send/recv roles swapped and every + // count and offset divided by `forward_stride`, the forward leg's elements per record (a query is >= 2 + // words, its answer one value). Every forward count is a multiple of the stride, so the division is + // exact and no count exchange or sizing pass is needed -- that is the 3*R*S^2 table entries and one + // barrier saved. Reusing the offsets undivided would still route correctly but would stage and + // transmit `forward_stride` times the bytes. + // + // The reuse is legal because both ends see the same nesting: rank X ordered its query block to Y as + // (dest partition of Y major, source partition of X minor), and Y's recv view of it has that nesting + // and those counts, so Y answering in place produces the layout X expects back. + // + // CONTRACT: must be the next table-touching verb after an alltoallv_resolve on this comm, with + // send_counts/recv_counts the transpose of that round's. Checked on partition 0 via tables_gen_. + auto alltoallv_reverse_impl_(int local_partition, + const void *send, + const int *send_counts /*[P]*/, + const int *send_displs /*[P]*/, + void *recv, + const int *recv_counts /*[P]*/, + const int *recv_displs /*[P]*/, + size_t elem, + MPI_Datatype dt, + int forward_stride) -> void { + count_verb(local_partition); + if (local_partition == 0 && tables_gen_ != reverse_ready_gen_) { + throw std::runtime_error("HybridComm::alltoallv_reverse must directly follow the " + "alltoallv_resolve whose layout it reverses; another collective " + "has overwritten the offset tables since."); + } + sync(local_partition); // B1: the contract check is global before anyone reuses a table + + if (local_partition == 0) { + ScopedNs timer{p0_table_ns(local_partition)}; + // Per-rank counts/displs are the forward round's, roles swapped and scaled down by the + // stride. Kept in their own scratch so the forward tables stay intact for the scatter below. + long long send_total = 0; + long long recv_total = 0; + for (int b = 0; b < r_; ++b) { + const size_t i = static_cast(b); + rev_send_counts_[i] = mpi_recv_counts_[i] / forward_stride; + rev_recv_counts_[i] = mpi_send_counts_[i] / forward_stride; + rev_send_displs_[i] = checked_int_(send_total); + rev_recv_displs_[i] = checked_int_(recv_total); + send_total += rev_send_counts_[i]; + recv_total += rev_recv_counts_[i]; + } + grow_(stage_send_, static_cast(send_total) * elem); + grow_(stage_recv_, static_cast(recv_total) * elem); + reverse_ready_gen_ = ~0ULL; // one reverse per resolve + } + sync(local_partition); // B2: staging sized + + // Pack into the query round's RECV geometry: partition t answers (rank a, partition su) at the + // very offset that query block occupied. + { + ScopedNs timer{move_ns(local_partition)}; + const int t = local_partition; + const char *src = static_cast(send); + for (int a = 0; a < r_; ++a) { + for (int su = 0; su < s_; ++su) { + const int g = a * s_ + su; + const int cnt = send_counts[g]; + if (cnt != 0) { + const size_t off = recv_block_off_(a, t, su) / static_cast(forward_stride); + std::memcpy(stage_send_.data() + off * elem, + src + static_cast(send_displs[g]) * elem, + static_cast(cnt) * elem); + } + } + } + } + sync(local_partition); // B3 + + if (local_partition == 0) { + ScopedNs timer{mpi_ns(local_partition)}; + MPI_Alltoallv(stage_send_.data(), + rev_send_counts_.data(), // send side = the query round's recv side / stride + rev_send_displs_.data(), + dt, + stage_recv_.data(), + rev_recv_counts_.data(), // recv side = the query round's send side / stride + rev_recv_displs_.data(), + dt, + parent_); + } + sync(local_partition); // B4 + + // Scatter from the query round's SEND geometry: partition u collects the answers to the queries + // it sent, block (rank b, dest partition t) sitting where it packed that query. + ScopedNs timer{move_ns(local_partition)}; + const int u = local_partition; + char *dst = static_cast(recv); + for (int b = 0; b < r_; ++b) { + const size_t base = static_cast(b) * static_cast(s_); + for (int t = 0; t < s_; ++t) { + const int g = b * s_ + t; + const int cnt = recv_counts[g]; + if (cnt != 0) { + const size_t off = (tpre_send_[base + static_cast(t)] + pack_off_[block_idx_(b, t, u)]) + / static_cast(forward_stride); + std::memcpy(dst + static_cast(recv_displs[g]) * elem, + stage_recv_.data() + off * elem, + static_cast(cnt) * elem); + } + } + } + } + template auto allreduce_sum_impl_(int local_partition, T local_val) -> T { + count_verb(local_partition); Slot &me = slots_[static_cast(local_partition)]; if constexpr (std::is_floating_point_v) { me.f64 = static_cast(local_val); @@ -339,8 +527,9 @@ class HybridComm { else { me.u64 = static_cast(local_val); } - sync(); + sync(local_partition); if (local_partition == 0) { + ScopedNs timer{mpi_ns(local_partition)}; if constexpr (std::is_floating_point_v) { double local = 0.0; for (int s = 0; s < s_; ++s) { @@ -356,7 +545,7 @@ class HybridComm { MPI_Allreduce(&local, &red_u64_, 1, MPI_UINT64_T, MPI_SUM, parent_); } } - sync(); + sync(local_partition); T out{}; if constexpr (std::is_floating_point_v) { out = static_cast(red_f64_); @@ -371,29 +560,34 @@ class HybridComm { // In-place element-wise allreduce-sum across the flat P-world, slice-partitioned across partitions in // ascending order (bit-identical to a sequential sum). auto allreduce_sum_inplace_impl_(int local_partition, double *values, size_t len) -> void { + count_verb(local_partition); slots_[static_cast(local_partition)].vec = values; - sync(); // all inputs published + sync(local_partition); // all inputs published if (local_partition == 0) { grow_(red_vec_, len); } - sync(); // red_vec_ sized + sync(local_partition); // red_vec_ sized constexpr size_t kLine = 64 / sizeof(double); const size_t lines = (len + kLine - 1) / kLine; const size_t per = (lines + static_cast(s_) - 1) / static_cast(s_); const size_t lo = std::min(len, static_cast(local_partition) * per * kLine); const size_t hi = std::min(len, lo + per * kLine); - for (size_t k = lo; k < hi; ++k) { - double acc = 0.0; - for (int s = 0; s < s_; ++s) { - acc += slots_[static_cast(s)].vec[k]; + { + ScopedNs timer{move_ns(local_partition)}; + for (size_t k = lo; k < hi; ++k) { + double acc = 0.0; + for (int s = 0; s < s_; ++s) { + acc += slots_[static_cast(s)].vec[k]; + } + red_vec_[k] = acc; // disjoint line-rounded slices: no two partitions store to one line } - red_vec_[k] = acc; // disjoint line-rounded slices: no two partitions store to one line } - sync(); // local reduction complete + sync(local_partition); // local reduction complete if (local_partition == 0) { + ScopedNs timer{mpi_ns(local_partition)}; MPI_Allreduce(MPI_IN_PLACE, red_vec_.data(), static_cast(len), MPI_DOUBLE, MPI_SUM, parent_); } - sync(); // global result in red_vec_ + sync(local_partition); // global result in red_vec_ std::memcpy(values, red_vec_.data(), len * sizeof(double)); // No trailing barrier: red_vec_ is rewritten only inside a future verb's barriered phases. } @@ -404,7 +598,6 @@ class HybridComm { private: struct alignas(64) Slot { const void *ptr = nullptr; - const int *counts = nullptr; const int *send_counts = nullptr; const int *send_displs = nullptr; const int *recv_counts = nullptr; @@ -413,12 +606,22 @@ class HybridComm { uint64_t u64 = 0; }; - // Flat index of the (rank, dest partition, source partition) block in the R*S*S offset/count tables. + // Flat index of the (rank, dest partition, source partition) block in the R*S*S offset tables. auto block_idx_(int b, int t, int u) const -> size_t { return (static_cast(b) * static_cast(s_) + static_cast(t)) * static_cast(s_) + static_cast(u); } + // Flat index into the exchanged count matrix, SOURCE-partition-major within each rank's block, so + // partition u owns the contiguous run [cnt_idx_(b,u,0), cnt_idx_(b,u,s_)) and fills it itself: the + // O(R*S^2) transpose partition 0 used to run alone becomes R*S contiguous writes per partition with + // no false sharing. Dest-partition-major would put s_ partitions on every cache line instead. + // MPI_Alltoall is indifferent: `b` stays outermost, and both sides index through this one helper. + auto cnt_idx_(int b, int u, int t) const -> size_t { + return (static_cast(b) * static_cast(s_) + static_cast(u)) * static_cast(s_) + + static_cast(t); + } + template static auto grow_(V &v, size_t need) -> void { if (v.size() < need) { @@ -426,63 +629,154 @@ class HybridComm { } } - // Partition 0 only; recv counts come from partition t's published recv_counts. - auto size_staging_(size_t elem) -> void { - size_staging_impl_(elem, [this](int t, int rank, int su) { - return slots_[static_cast(t)].recv_counts[rank * s_ + su]; - }); + // Round a row of 8-byte entries up to whole cache lines, so one partition's row never shares a line + // with a peer's. + static auto pad_to_line_(size_t entries) -> size_t { + constexpr size_t kPerLine = 64 / 8; + return ((entries + kPerLine - 1) / kPerLine) * kPerLine; } - // recv_count(t, rank, su) yields the count partition t on this rank receives from (rank, source partition su). - template - auto size_staging_impl_(size_t elem, RecvCountFn recv_count) -> void { + // Partition u's veto row: bit b set iff u sends at least one element to rank b. `mask_words_` is the + // padded row stride; only the first ceil(R/64) words carry bits. + auto mask_row_(int u) -> uint64_t * { return send_mask_.data() + static_cast(u) * mask_words_; } + static auto mask_test_(const uint64_t *row, int b) -> bool { + return (row[static_cast(b) >> 6] & (1ULL << (static_cast(b) & 63U))) != 0; + } + auto mask_empty_(const uint64_t *row) const -> bool { + const size_t live = static_cast((r_ + 63) / 64); + for (size_t w = 0; w < live; ++w) { + if (row[w] != 0) { + return false; + } + } + return true; + } + + // Phase A's per-rank running offsets, one padded row per partition (r_ can be in the hundreds, so + // this is a member and not a stack array). + auto run_row_(int t) -> size_t * { return run_scratch_.data() + static_cast(t) * run_stride_; } + + // Partition u summarizes its own send counts into its veto row, from its own argument and before the + // publishing barrier, so it costs no barrier and reads no peer. + auto fill_send_mask_(int local_partition, const int *send_counts /*[P]*/) -> void { + uint64_t *row = mask_row_(local_partition); + std::fill(row, row + mask_words_, uint64_t{0}); for (int b = 0; b < r_; ++b) { - long long send_sum = 0; - long long recv_sum = 0; + const int *counts = send_counts + b * s_; for (int t = 0; t < s_; ++t) { - for (int su = 0; su < s_; ++su) { - send_sum += slots_[static_cast(su)].send_counts[b * s_ + t]; - recv_sum += recv_count(t, b, su); + if (counts[t] != 0) { + row[static_cast(b) >> 6] |= 1ULL << (static_cast(b) & 63U); + break; } } - mpi_send_counts_[static_cast(b)] = checked_int_(send_sum); - mpi_recv_counts_[static_cast(b)] = checked_int_(recv_sum); } - long long send_running = 0; - long long recv_running = 0; - for (int b = 0; b < r_; ++b) { - mpi_send_displs_[static_cast(b)] = checked_int_(send_running); - mpi_recv_displs_[static_cast(b)] = checked_int_(recv_running); - send_running += mpi_send_counts_[static_cast(b)]; - recv_running += mpi_recv_counts_[static_cast(b)]; - } - const size_t total_send = static_cast(checked_int_(send_running)); - const size_t total_recv = static_cast(checked_int_(recv_running)); - // Grow-only, no zero-fill: pack_send_'s blocks tile [0, total_send) exactly and MPI_Alltoallv - // fills every live byte of stage_recv_, so stale bytes past a prior high-water mark are never read. - grow_(stage_send_, total_send * elem); - grow_(stage_recv_, total_recv * elem); - // Precompute each (rank b, dest partition t, source partition u) block start in elements, dest-major/ - // source-minor to match staging: pack/scatter become O(1) lookups that read no peer slot past B4 - // (re-summing peer count matrices instead would be O(R*S^3)). + } + + // Partition u writes its own source row of the exchanged count matrix. Disjoint per u and + // contiguous in t, so it needs no barrier of its own and no coherence ping-pong. + auto fill_count_row_(int local_partition, const int *send_counts /*[P]*/) -> void { + const int u = local_partition; for (int b = 0; b < r_; ++b) { - size_t cur = static_cast(mpi_send_displs_[static_cast(b)]); + const int *row = send_counts + b * s_; + int *dst = counts_send_.data() + cnt_idx_(b, u, 0); for (int t = 0; t < s_; ++t) { - for (int u = 0; u < s_; ++u) { - pack_off_[block_idx_(b, t, u)] = cur; - cur += static_cast(slots_[static_cast(u)].send_counts[b * s_ + t]); - } + dst[t] = row[t]; } } - for (int a = 0; a < r_; ++a) { - size_t cur = static_cast(mpi_recv_displs_[static_cast(a)]); - for (int t = 0; t < s_; ++t) { + fill_send_mask_(local_partition, send_counts); + } + + // Non-resolve alltoallv path: recv counts come from partition t's own published recv_counts, so + // in phase A below every partition reads only its own slot. + auto size_staging_parallel_(int local_partition, size_t elem) -> void { + size_staging_parallel_(local_partition, elem, [this](int t, int rank, int su) { + return slots_[static_cast(t)].recv_counts[rank * s_ + su]; + }); + } + + // Replaces the O(R*S^2) serial sizing pass with a parallel prefix. recv_count(t, rank, su) is the + // count partition t on this rank receives from (rank, source partition su). + // + // The offset of block (b, t, u) in the staged message decomposes as + // mpi_send_displs_[b] + sum_{t' < t} col_send_[b][t'] + sum_{u' < u} c[u'][b][t] + // \------------- tpre_send_[b][t], one O(R*S) scan -----/ \--- pack_off_[b][t][u] ---/ + // The right-hand term is an independent scan per (b,t) pair, so partition t owns every pair with its + // own t: R*S work each, S-way parallel, covering the same R*S^2 entries. Only the middle term needs + // global knowledge, and at O(R*S) it is small enough to leave on partition 0. + // + // The two barriers this costs (phase A complete before partition 0 reduces it, bases visible before + // anyone packs) buy S-way parallelism over what measured 84% of wall time at S=112. + template + auto size_staging_parallel_(int local_partition, size_t elem, RecvCountFn recv_count) -> void { + const int t = local_partition; + { + ScopedNs timer{par_table_ns(local_partition)}; + // Send half, SOURCE-partition-outer so each peer's veto row is loaded once for all R ranks + // instead of probing its count array R times on R separate cache lines -- the dominant cost + // of this phase once the serial fill was gone. Skipped blocks leave a stale pack_off_ entry, + // which is safe: an offset is read only where the matching count is nonzero. + size_t *run = run_row_(t); + std::fill(run, run + r_, size_t{0}); + for (int u = 0; u < s_; ++u) { + const uint64_t *mask = mask_row_(u); + if (mask_empty_(mask)) { + continue; + } + const int *counts = slots_[static_cast(u)].send_counts; + for (int b = 0; b < r_; ++b) { + if (!mask_test_(mask, b)) { + continue; // u sends nothing to rank b, so it contributes 0 to every (b,t) scan + } + pack_off_[block_idx_(b, t, u)] = run[b]; + run[b] += static_cast(counts[b * s_ + t]); + } + } + for (int b = 0; b < r_; ++b) { + col_send_[static_cast(b) * static_cast(s_) + static_cast(t)] = run[b]; + size_t got = 0; for (int su = 0; su < s_; ++su) { - scatter_off_[block_idx_(a, t, su)] = cur; - cur += static_cast(recv_count(t, a, su)); + scatter_off_[block_idx_(b, t, su)] = got; + got += static_cast(recv_count(t, b, su)); } + col_recv_[static_cast(b) * static_cast(s_) + static_cast(t)] = got; } } + sync(local_partition); // phase A complete on every partition + + if (local_partition == 0) { + ScopedNs timer{p0_table_ns(local_partition)}; + long long send_running = 0; + long long recv_running = 0; + for (int b = 0; b < r_; ++b) { + long long send_sum = 0; + long long recv_sum = 0; + const size_t base = static_cast(b) * static_cast(s_); + for (int k = 0; k < s_; ++k) { + send_sum += static_cast(col_send_[base + static_cast(k)]); + recv_sum += static_cast(col_recv_[base + static_cast(k)]); + } + mpi_send_counts_[static_cast(b)] = checked_int_(send_sum); + mpi_recv_counts_[static_cast(b)] = checked_int_(recv_sum); + mpi_send_displs_[static_cast(b)] = checked_int_(send_running); + mpi_recv_displs_[static_cast(b)] = checked_int_(recv_running); + // tpre_*_ folds the per-rank base in, so pack/scatter add exactly two numbers. + size_t cur_s = static_cast(send_running); + size_t cur_r = static_cast(recv_running); + for (int k = 0; k < s_; ++k) { + tpre_send_[base + static_cast(k)] = cur_s; + cur_s += col_send_[base + static_cast(k)]; + tpre_recv_[base + static_cast(k)] = cur_r; + cur_r += col_recv_[base + static_cast(k)]; + } + send_running += mpi_send_counts_[static_cast(b)]; + recv_running += mpi_recv_counts_[static_cast(b)]; + } + // Grow-only, no zero-fill: pack_send_'s blocks tile [0, total_send) exactly and MPI_Alltoallv + // fills every live byte of stage_recv_, so stale bytes past a prior high-water mark are never read. + grow_(stage_send_, static_cast(checked_int_(send_running)) * elem); + grow_(stage_recv_, static_cast(checked_int_(recv_running)) * elem); + } + sync(local_partition); // bases and staging visible to every packer } auto pack_send_(int local_partition, size_t elem) -> void { @@ -491,10 +785,13 @@ class HybridComm { const int *my_send_counts = slots_[static_cast(u)].send_counts; const int *my_send_displs = slots_[static_cast(u)].send_displs; for (int b = 0; b < r_; ++b) { + const size_t base = static_cast(b) * static_cast(s_); for (int t = 0; t < s_; ++t) { const int cnt = my_send_counts[b * s_ + t]; if (cnt != 0) { - std::memcpy(stage_send_.data() + pack_off_[block_idx_(b, t, u)] * elem, + // Absolute start = per-(b,t) base (partition 0) + within-(b,t) scan (partition t). + const size_t off = tpre_send_[base + static_cast(t)] + pack_off_[block_idx_(b, t, u)]; + std::memcpy(stage_send_.data() + off * elem, src + static_cast(my_send_displs[b * s_ + t]) * elem, static_cast(cnt) * elem); } @@ -502,6 +799,12 @@ class HybridComm { } } + // Absolute start of the block partition t receives from (rank a, source partition su). + auto recv_block_off_(int a, int t, int su) const -> size_t { + return tpre_recv_[static_cast(a) * static_cast(s_) + static_cast(t)] + + scatter_off_[block_idx_(a, t, su)]; + } + static auto checked_int_(long long v) -> int { if (v < 0 || v > static_cast(2147483647)) { throw std::runtime_error("HybridComm: aggregated per-rank count overflows int (message too large)"); @@ -521,7 +824,31 @@ class HybridComm { std::abort(); // MPI_Abort is not marked [[noreturn]]; unreachable in practice } - auto sync() -> void { barrier_.sync(); } + // Barrier wait is attributed to the partition that waited, which is the whole point: when the + // master monopolises a serial phase, its own wait stays near zero while every peer's grows. + auto sync(int local_partition) -> void { + if (prof_ == nullptr) { + barrier_.sync(local_partition); + return; + } + CommProfile::Slot &sl = prof_->slot(local_partition); + ++sl.n_barriers; + ScopedNs t{&sl.barrier_ns}; + barrier_.sync(local_partition); + } + + // nullptr target ⇒ ScopedNs is inert, so instrumented regions need no #ifdef or duplicate code. + auto p0_table_ns(int u) -> uint64_t * { return prof_ == nullptr ? nullptr : &prof_->slot(u).table_p0_ns; } + auto par_table_ns(int u) -> uint64_t * { return prof_ == nullptr ? nullptr : &prof_->slot(u).table_par_ns; } + auto move_ns(int u) -> uint64_t * { return prof_ == nullptr ? nullptr : &prof_->slot(u).table_move_ns; } + auto mpi_ns(int u) -> uint64_t * { return prof_ == nullptr ? nullptr : &prof_->slot(u).mpi_ns; } + auto count_verb(int u) -> void { + if (prof_ != nullptr) { + ++prof_->slot(u).n_verbs; + } + } + + std::unique_ptr prof_; MPI_Comm parent_; int s_; @@ -538,12 +865,32 @@ class HybridComm { std::vector mpi_send_displs_; std::vector mpi_recv_counts_; std::vector mpi_recv_displs_; - // [R*S*S] block starts (elements) in the staging buffers. + // [R*S*S] within-(rank, dest partition) exclusive scans over the SOURCE partition, filled by the + // owning dest partition; absolute starts are these plus tpre_*_ (see size_staging_parallel_). std::vector pack_off_; std::vector scatter_off_; + // [R*S] per-(rank, dest partition) column totals and their prefix, the only globally-reduced part. + std::vector col_send_; + std::vector col_recv_; + std::vector tpre_send_; + std::vector tpre_recv_; + // [S * mask_words_] one cache-line-padded veto row per source partition, and [S * run_stride_] + // per-partition scratch for phase A's per-rank running offsets. Both written only by their owner. + std::vector send_mask_; + std::vector run_scratch_; + size_t mask_words_ = 8; + size_t run_stride_ = 8; // Aggregated MPI payload staging, HWM-sized. std::vector stage_send_; std::vector stage_recv_; + // [R] reverse-round per-rank counts/displs (the forward round's, swapped and divided by the stride). + // Only partition 0 touches these and tables_gen_/reverse_ready_gen_, inside a barriered window. + std::vector rev_send_counts_; + std::vector rev_send_displs_; + std::vector rev_recv_counts_; + std::vector rev_recv_displs_; + uint64_t tables_gen_ = 0; + uint64_t reverse_ready_gen_ = ~0ULL; double red_f64_ = 0.0; uint64_t red_u64_ = 0; std::vector red_vec_; diff --git a/src/monoprop/detail/mpi/MPICompat.h b/src/monoprop/detail/mpi/MPICompat.h index 786d7342..20344bc5 100644 --- a/src/monoprop/detail/mpi/MPICompat.h +++ b/src/monoprop/detail/mpi/MPICompat.h @@ -236,11 +236,18 @@ struct PendingAlltoallv { // skip_self: do not send the self slot (the caller handles self inline) — self send/recv = 0. // known_recv_counts: recv counts already known (e.g. the transpose of the query counts), so skip the // count exchange. The self slot is also zeroed when skip_self is set. +// reverse_of_previous: this exchange is the answer leg of the immediately preceding begin_alltoallv on +// the same comm, and forward_stride is that leg's elements per record (a query is several words, its +// answer one value). Only the hybrid transport acts on it, reusing that round's offset tables instead of +// rebuilding them (see HybridComm::alltoallv_reverse); it requires known_recv_counts, and no other +// collective may intervene on that comm. template inline auto begin_alltoallv(const std::vector> &send_data, Comm comm, bool skip_self = false, - const std::vector *known_recv_counts = nullptr) -> PendingAlltoallv { + const std::vector *known_recv_counts = nullptr, + bool reverse_of_previous = false, + int forward_stride = 1) -> PendingAlltoallv { const int num_ranks = size(comm); if (static_cast(send_data.size()) != num_ranks) { throw CollectiveArgumentError( @@ -342,15 +349,29 @@ inline auto begin_alltoallv(const std::vector> &send_data, } #ifdef monoprop_ENABLE_MPI else if (comm.kind == Comm::Kind::Hybrid) { - comm.hyb->alltoallv(comm.shm_rank, - h.send_buffer.data(), - h.send_counts.data(), - h.send_displs.data(), - h.recv_buffer.data(), - h.recv_counts.data(), - h.recv_displs.data(), - sizeof(T), - datatype::get()); + if (reverse_of_previous) { + comm.hyb->alltoallv_reverse(comm.shm_rank, + h.send_buffer.data(), + h.send_counts.data(), + h.send_displs.data(), + h.recv_buffer.data(), + h.recv_counts.data(), + h.recv_displs.data(), + sizeof(T), + datatype::get(), + forward_stride); + } + else { + comm.hyb->alltoallv(comm.shm_rank, + h.send_buffer.data(), + h.send_counts.data(), + h.send_displs.data(), + h.recv_buffer.data(), + h.recv_counts.data(), + h.recv_displs.data(), + sizeof(T), + datatype::get()); + } } #endif else { diff --git a/src/monoprop/detail/mpi/PartitionBarrier.h b/src/monoprop/detail/mpi/PartitionBarrier.h index e8cac2ec..02032adc 100644 --- a/src/monoprop/detail/mpi/PartitionBarrier.h +++ b/src/monoprop/detail/mpi/PartitionBarrier.h @@ -14,9 +14,11 @@ #pragma once +#include #include #include #include +#include #include "monoprop/detail/mpi/CpuRelax.h" @@ -31,34 +33,77 @@ class ShmCommPoisoned : public std::runtime_error { // Sense-reversing generation barrier for a fixed number of in-process partition threads. Each barrier // word gets a private cache line: if `gen_` shared `arrived_`'s line, spinners' reloads would miss to // L3 on every peer arrival — O(S) coherence bounces (measured top hotspot at S=112). +// +// Given a group id per participant (its L3 domain, from CpuTopology) the barrier becomes two-level: fan +// in within a domain, then across domains, then release within the domain. Both the arrival fetch_add +// and the release store then cost O(S/G) coherence transactions instead of O(S), and stay inside one L3 +// slice. No group ids (pinning off, or /sys unreadable) or a single domain degrades to the flat barrier. class PartitionBarrier { public: - explicit PartitionBarrier(int participants) : participants_(participants) {} + explicit PartitionBarrier(int participants, const std::vector &group_of = {}) : participants_(participants) { + if (static_cast(group_of.size()) != participants || participants <= 0) { + return; // flat + } + // Compact the domain ids to 0..G-1 in first-seen order, and count each group. + std::vector domains; + group_of_.resize(static_cast(participants)); + for (int p = 0; p < participants; ++p) { + const auto it = std::find(domains.begin(), domains.end(), group_of[static_cast(p)]); + if (it == domains.end()) { + group_of_[static_cast(p)] = static_cast(domains.size()); + domains.push_back(group_of[static_cast(p)]); + } + else { + group_of_[static_cast(p)] = static_cast(it - domains.begin()); + } + } + if (domains.size() < 2) { + group_of_.clear(); + return; // flat + } + groups_ = static_cast(domains.size()); + group_arrived_ = std::vector(domains.size()); + group_gen_ = std::vector(domains.size()); + group_size_.assign(domains.size(), 0); + for (int p = 0; p < participants; ++p) { + ++group_size_[static_cast(group_of_[static_cast(p)])]; + } + } PartitionBarrier(const PartitionBarrier &) = delete; auto operator=(const PartitionBarrier &) -> PartitionBarrier & = delete; - auto sync() -> void { - const unsigned g = gen_.load(std::memory_order_acquire); - if (arrived_.fetch_add(1, std::memory_order_acq_rel) + 1 == participants_) { - arrived_.store(0, std::memory_order_relaxed); - gen_.store(g + 1, std::memory_order_release); + // `participant` selects the caller's group; ignored on the flat path. + auto sync(int participant) -> void { + if (groups_ < 2) { + const unsigned g = gen_.load(std::memory_order_acquire); + if (arrived_.fetch_add(1, std::memory_order_acq_rel) + 1 == participants_) { + arrived_.store(0, std::memory_order_relaxed); + gen_.store(g + 1, std::memory_order_release); + } + else { + spin_until_(gen_, g); + } } else { - // Bounded on-core spin first (pinned partitions ⇒ the release store lands in the pause window, - // no syscall); only long waits (imbalance, oversubscription) fall to yield. - int spins = 0; - while (gen_.load(std::memory_order_acquire) == g) { - if (poisoned_.load(std::memory_order_acquire)) { - throw ShmCommPoisoned(); - } - if (spins < detail::kSpinPauseIters) { - ++spins; - detail::cpu_relax(); + const size_t gi = static_cast(group_of_[static_cast(participant)]); + const unsigned seen = group_gen_[gi].v.load(std::memory_order_acquire); + if (group_arrived_[gi].v.fetch_add(1, std::memory_order_acq_rel) + 1 == group_size_[gi]) { + group_arrived_[gi].v.store(0, std::memory_order_relaxed); + // The domain's last arriver represents it upstream, and releases the domain only after + // the root barrier, so passing the domain word still implies everyone arrived. + const unsigned root_seen = gen_.load(std::memory_order_acquire); + if (arrived_.fetch_add(1, std::memory_order_acq_rel) + 1 == groups_) { + arrived_.store(0, std::memory_order_relaxed); + gen_.store(root_seen + 1, std::memory_order_release); } else { - std::this_thread::yield(); + spin_until_(gen_, root_seen); } + group_gen_[gi].v.store(seen + 1, std::memory_order_release); + } + else { + spin_until_(group_gen_[gi].v, seen); } } if (poisoned_.load(std::memory_order_acquire)) { @@ -66,20 +111,60 @@ class PartitionBarrier { } } + // L3 domains the participants were grouped into; < 2 means the flat barrier is in use. Reported by + // CommProfile so a run states which barrier it ran instead of leaving it inferred from the timing. + auto group_count() const -> int { return groups_; } + // Signal that this participant is unwinding (e.g. an engine exception), releasing peers spinning in a // barrier. Idempotent. auto poison() -> void { poisoned_.store(true, std::memory_order_release); } // Must be called only when every participant is quiescent (between rounds), so a poison-aborted round - // leaves no dirty state. `gen_` deliberately stays monotonic: each participant re-reads it at its next - // barrier. + // leaves no dirty state. The generations deliberately stay monotonic: each participant re-reads its + // own at the next barrier. auto reset() -> void { poisoned_.store(false, std::memory_order_relaxed); arrived_.store(0, std::memory_order_relaxed); + for (auto &c : group_arrived_) { + c.v.store(0, std::memory_order_relaxed); + } } private: + // Spin until `word` leaves `seen`. Bounded on-core spin first (pinned partitions ⇒ the release store + // lands in the pause window, no syscall); only long waits (imbalance, oversubscription) fall to + // yield. A poisoned peer releases us with an exception rather than a hang. + auto spin_until_(const std::atomic &word, unsigned seen) const -> void { + int spins = 0; + while (word.load(std::memory_order_acquire) == seen) { + if (poisoned_.load(std::memory_order_acquire)) { + throw ShmCommPoisoned(); + } + if (spins < detail::kSpinPauseIters) { + ++spins; + detail::cpu_relax(); + } + else { + std::this_thread::yield(); + } + } + } + + // Per-domain words, each on its own cache line for the same reason the flat pair is split. + struct alignas(64) Count { + std::atomic v{0}; + }; + struct alignas(64) Gen { + std::atomic v{0}; + }; + int participants_; + int groups_ = 0; // < 2 ⇒ flat: arrived_/gen_ count participants, not domains + std::vector group_of_; + std::vector group_size_; + std::vector group_arrived_; + std::vector group_gen_; + // Flat barrier, or (two-level) the root barrier the domains' last arrivers meet at. alignas(64) std::atomic arrived_{0}; alignas(64) std::atomic gen_{0}; alignas(64) std::atomic poisoned_{false}; diff --git a/src/monoprop/detail/mpi/ShmComm.h b/src/monoprop/detail/mpi/ShmComm.h index a9e021b3..657a95ea 100644 --- a/src/monoprop/detail/mpi/ShmComm.h +++ b/src/monoprop/detail/mpi/ShmComm.h @@ -24,7 +24,9 @@ #include #include +#include "monoprop/detail/EnvConfig.h" #include "monoprop/detail/mpi/CheckedCount.h" +#include "monoprop/detail/mpi/CommProfile.h" #include "monoprop/detail/mpi/PartitionBarrier.h" // In-process shared-memory SPMD transport: S partition-master threads each call the same collective @@ -37,21 +39,41 @@ namespace monoprop::mpi { class ShmComm { public: - explicit ShmComm(int n) : n_(n), slots_(static_cast(n)), barrier_(n) {} + // partition_l3_domains (optional, one entry per partition) makes the barrier two-level; see + // PartitionBarrier. + explicit ShmComm(int n, const std::vector &partition_l3_domains = {}) + : n_(n), + slots_(static_cast(n)), + barrier_(n, partition_l3_domains) { + if (config::get().comm_profile) { + prof_ = std::make_unique(n, /*mpi_rank=*/-1); // -1 marks the single-rank transport + prof_->barrier_groups = barrier_.group_count(); + } + } ShmComm(const ShmComm &) = delete; auto operator=(const ShmComm &) -> ShmComm & = delete; + ~ShmComm() { + if (prof_ != nullptr) { + prof_->dump(); + } + } + auto size() const -> int { return n_; } // recv_counts[s] = what rank s sends to me (the transpose of the send-count matrix). auto alltoall_counts(int rank, const int *send_counts, int *recv_counts) -> void { + count_verb(rank); slots_[static_cast(rank)].counts = send_counts; - sync(); - for (int s = 0; s < n_; ++s) { - recv_counts[s] = slots_[static_cast(s)].counts[rank]; + sync(rank); + { + ScopedNs timer{par_table_ns(rank)}; + for (int s = 0; s < n_; ++s) { + recv_counts[s] = slots_[static_cast(s)].counts[rank]; + } } - sync(); + sync(rank); } // Variable all-to-all over caller-owned flat buffers (counts/displs in elements, `elem` = element @@ -64,23 +86,27 @@ class ShmComm { const int *recv_counts, const int *recv_displs, size_t elem) -> void { + count_verb(rank); Slot &me = slots_[static_cast(rank)]; me.ptr = send; me.displs = send_displs; - sync(); - auto *dst = static_cast(recv); - for (int s = 0; s < n_; ++s) { - const Slot &src = slots_[static_cast(s)]; - const auto *sp = static_cast(src.ptr); - const auto count = static_cast(recv_counts[s]); - if (count == 0) { - continue; + sync(rank); + { + ScopedNs timer{move_ns(rank)}; + auto *dst = static_cast(recv); + for (int s = 0; s < n_; ++s) { + const Slot &src = slots_[static_cast(s)]; + const auto *sp = static_cast(src.ptr); + const auto count = static_cast(recv_counts[s]); + if (count == 0) { + continue; + } + std::memcpy(dst + static_cast(recv_displs[s]) * elem, + sp + static_cast(src.displs[rank]) * elem, + count * elem); } - std::memcpy(dst + static_cast(recv_displs[s]) * elem, - sp + static_cast(src.displs[rank]) * elem, - count * elem); } - sync(); + sync(rank); } // Fused count-resolve + payload all-to-all in one round (2 syncs vs 4). Same contiguous @@ -97,27 +123,34 @@ class ShmComm { me.ptr = send; me.displs = send_displs; me.counts = send_counts; - sync(); // B1: send buffers published - long long total = 0; - for (int s = 0; s < n_; ++s) { - const int c = slots_[static_cast(s)].counts[rank]; // what s sends to me - recv_counts[s] = c; - recv_displs[s] = checked_mpi_count(total, "Recv displacement"); - total += c; + count_verb(rank); + sync(rank); // B1: send buffers published + { + ScopedNs timer{par_table_ns(rank)}; + long long total = 0; + for (int s = 0; s < n_; ++s) { + const int c = slots_[static_cast(s)].counts[rank]; // what s sends to me + recv_counts[s] = c; + recv_displs[s] = checked_mpi_count(total, "Recv displacement"); + total += c; + } + recv.resize(static_cast(checked_mpi_count(total, "Total recv count"))); } - recv.resize(static_cast(checked_mpi_count(total, "Total recv count"))); - auto *dst = reinterpret_cast(recv.data()); - for (int s = 0; s < n_; ++s) { - const Slot &src = slots_[static_cast(s)]; - const auto count = static_cast(recv_counts[s]); - if (count == 0) { - continue; + { + ScopedNs timer{move_ns(rank)}; + auto *dst = reinterpret_cast(recv.data()); + for (int s = 0; s < n_; ++s) { + const Slot &src = slots_[static_cast(s)]; + const auto count = static_cast(recv_counts[s]); + if (count == 0) { + continue; + } + std::memcpy(dst + static_cast(recv_displs[s]) * sizeof(T), + reinterpret_cast(src.ptr) + static_cast(src.displs[rank]) * sizeof(T), + count * sizeof(T)); } - std::memcpy(dst + static_cast(recv_displs[s]) * sizeof(T), - reinterpret_cast(src.ptr) + static_cast(src.displs[rank]) * sizeof(T), - count * sizeof(T)); } - sync(); // B2: peers finished reading our send buffer before the caller may reuse it + sync(rank); // B2: peers finished reading our send buffer before the caller may reuse it } template @@ -129,7 +162,8 @@ class ShmComm { else { me.u64 = static_cast(local_val); } - sync(); + count_verb(rank); + sync(rank); T acc{}; for (int s = 0; s < n_; ++s) { const Slot &src = slots_[static_cast(s)]; @@ -140,30 +174,34 @@ class ShmComm { acc += static_cast(src.u64); } } - sync(); + sync(rank); return acc; } // Safe in place: each element is read then overwritten by its single slice owner, and slices are // cache-line-rounded. auto allreduce_sum_inplace(int rank, double *values, size_t len) -> void { + count_verb(rank); slots_[static_cast(rank)].vec = values; - sync(); - constexpr size_t kLine = 64 / sizeof(double); - const size_t lines = (len + kLine - 1) / kLine; - const size_t per = (lines + static_cast(n_) - 1) / static_cast(n_); - const size_t lo = std::min(len, static_cast(rank) * per * kLine); - const size_t hi = std::min(len, lo + per * kLine); - for (size_t k = lo; k < hi; ++k) { - double acc = 0.0; - for (int s = 0; s < n_; ++s) { // ascending rank order - acc += slots_[static_cast(s)].vec[k]; - } - for (int s = 0; s < n_; ++s) { // publish the same bits into every rank's buffer - slots_[static_cast(s)].vec[k] = acc; + sync(rank); + { + ScopedNs timer{move_ns(rank)}; + constexpr size_t kLine = 64 / sizeof(double); + const size_t lines = (len + kLine - 1) / kLine; + const size_t per = (lines + static_cast(n_) - 1) / static_cast(n_); + const size_t lo = std::min(len, static_cast(rank) * per * kLine); + const size_t hi = std::min(len, lo + per * kLine); + for (size_t k = lo; k < hi; ++k) { + double acc = 0.0; + for (int s = 0; s < n_; ++s) { // ascending rank order + acc += slots_[static_cast(s)].vec[k]; + } + for (int s = 0; s < n_; ++s) { // publish the same bits into every rank's buffer + slots_[static_cast(s)].vec[k] = acc; + } } } - sync(); // peers write into our buffer (and read from it) until here + sync(rank); // peers write into our buffer (and read from it) until here } // See PartitionBarrier::poison / ::reset for when each is legal to call. @@ -183,8 +221,28 @@ class ShmComm { uint64_t u64 = 0; }; - auto sync() -> void { barrier_.sync(); } + // Same per-partition attribution as HybridComm::sync. Every non-barrier phase here is already + // parallel, so a run's whole floor lands in barrier_ns and reads off as cost-per-sync. + auto sync(int rank) -> void { + if (prof_ == nullptr) { + barrier_.sync(rank); + return; + } + CommProfile::Slot &sl = prof_->slot(rank); + ++sl.n_barriers; + ScopedNs t{&sl.barrier_ns}; + barrier_.sync(rank); + } + + auto par_table_ns(int rank) -> uint64_t * { return prof_ == nullptr ? nullptr : &prof_->slot(rank).table_par_ns; } + auto move_ns(int rank) -> uint64_t * { return prof_ == nullptr ? nullptr : &prof_->slot(rank).table_move_ns; } + auto count_verb(int rank) -> void { + if (prof_ != nullptr) { + ++prof_->slot(rank).n_verbs; + } + } + std::unique_ptr prof_; int n_; std::vector slots_; PartitionBarrier barrier_; diff --git a/src/monoprop/detail/partition/CpuTopology.h b/src/monoprop/detail/partition/CpuTopology.h index 8eb146dd..3f973b21 100644 --- a/src/monoprop/detail/partition/CpuTopology.h +++ b/src/monoprop/detail/partition/CpuTopology.h @@ -229,6 +229,32 @@ inline auto partition_cpusets(size_t n, size_t group_index = 0, size_t group_cou return sets; } +// The L3 domain each partition cpuset lands in, in partition_cpusets order -- what a two-level +// PartitionBarrier groups by. Derived from the sets, not from the placement logic, so the two cannot +// drift apart. Empty (⇒ flat barrier) if the sets are empty or one names no core the scan knows. +inline auto cpuset_l3_domains(const std::vector &sets) -> std::vector { + if (sets.empty()) { + return {}; + } + const auto cores = enumerate_physical_cores(); + std::vector domains; + domains.reserve(sets.size()); + for (const CpuSet &set : sets) { + int found = -1; + for (const auto &core : cores) { + if (CPU_ISSET(core.cpu, &set)) { + found = core.l3_domain; + break; + } + } + if (found < 0) { + return {}; + } + domains.push_back(found); + } + return domains; +} + // A failing pthread call is ignored: only performance depends on it. inline auto pin_this_thread(const CpuSet &set) -> void { pthread_setaffinity_np(pthread_self(), sizeof(CpuSet), &set); @@ -256,6 +282,9 @@ inline auto partition_cpusets(size_t /*n*/, size_t /*group_index*/ = 0, size_t / -> std::vector { return {}; } +inline auto cpuset_l3_domains(const std::vector & /*sets*/) -> std::vector { + return {}; +} inline auto pin_this_thread(const CpuSet & /*set*/) -> void {} #endif // __linux__ diff --git a/src/monoprop/detail/partition/PartitionGroup.h b/src/monoprop/detail/partition/PartitionGroup.h index fc3f64fd..d2661402 100644 --- a/src/monoprop/detail/partition/PartitionGroup.h +++ b/src/monoprop/detail/partition/PartitionGroup.h @@ -57,9 +57,11 @@ class PartitionGroup { parent_(parent), partitions_(static_cast(n_partitions)), errs_(static_cast(n_partitions)) { - make_transport_(); + // Placement is decided before the transport, because the transport's barrier is grouped by the + // L3 domain each partition will be pinned to. discover_node_peers_(); cpusets_ = topo_partition_cpusets(n_, node_rank_, node_size_); + make_transport_(); 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. @@ -81,8 +83,8 @@ class PartitionGroup { node_size_(src.node_size_), partitions_(static_cast(src.n_)), errs_(static_cast(src.n_)) { - make_transport_(); cpusets_ = topo_partition_cpusets(n_, node_rank_, node_size_); + make_transport_(); // after cpusets_, as in the primary ctor start_masters_(); try { // see the primary ctor: a throw past live masters would std::terminate run_on_all([&](int r) { @@ -168,13 +170,16 @@ class PartitionGroup { } auto make_transport_() -> void { + // Empty unless the partitions are pinned and /sys was readable ⇒ flat barrier (see + // PartitionBarrier); cpusets_ must already be set. + const std::vector domains = monoprop::detail::partition::cpuset_l3_domains(cpusets_); #ifdef monoprop_ENABLE_MPI if (parent_.kind == mpi::Comm::Kind::Mpi && mpi::size(parent_) > 1) { - hyb_ = std::make_unique(parent_.mpi, n_); + hyb_ = std::make_unique(parent_.mpi, n_, domains); return; } #endif - shm_ = std::make_unique(n_); + shm_ = std::make_unique(n_, domains); } auto comm_for_(int r) -> mpi::Comm { #ifdef monoprop_ENABLE_MPI diff --git a/tests/cpp/hybrid_comm_tests.cpp b/tests/cpp/hybrid_comm_tests.cpp index 7c0e80f4..24fa3a39 100644 --- a/tests/cpp/hybrid_comm_tests.cpp +++ b/tests/cpp/hybrid_comm_tests.cpp @@ -328,4 +328,132 @@ BOOST_AUTO_TEST_CASE(hybrid_comm_poison_releases_waiters) { } } +// The cases above give every destination the SAME count per source, which cannot tell the +// (rank, dest partition, source partition) index order the offset tables are built from from its +// transpose. So drive counts that depend on source AND destination, and tag every element with both ends +// of its leg: a swapped index or a wrong base then misroutes payload instead of yielding a +// coincidentally-correct total. S=5 (> the 3 used above) leaves R*S^2 room to go wrong. +BOOST_AUTO_TEST_CASE(hybrid_comm_alltoallv_resolve_asymmetric_counts) { + if (world_size() < 2) { + return; + } + const int R = world_size(); + const int S = 5; + const int P = R * S; + const int rounds = 12; + // Deliberately not symmetric under swapping src/dst, and hits 0 for some legs. + const auto leg_len = [](int src, int dst, int round) { return (src * 7 + dst * 3 + round) % 5; }; + const auto tag = [](int src, int dst, int j) { return ((src * 1000 + dst) * 100) + j; }; + std::atomic failures{0}; + auto errs = run_hybrid(S, [&](HybridComm &hyb, int u) { + Comm c = Comm::make_hybrid(&hyb, u); + const int g = monoprop::mpi::rank(c); + std::vector recv; + std::vector rc(static_cast(P)), rd(static_cast(P)); + for (int round = 0; round < rounds; ++round) { + std::vector send; + std::vector sc(static_cast(P)), sd(static_cast(P)); + int off = 0; + for (int d = 0; d < P; ++d) { + const int len = leg_len(g, d, round); + sc[static_cast(d)] = len; + sd[static_cast(d)] = off; + for (int j = 0; j < len; ++j) { + send.push_back(tag(g, d, j)); + } + off += len; + } + hyb.alltoallv_resolve(u, + send.data(), + sc.data(), + sd.data(), + recv, + rc.data(), + rd.data(), + sizeof(int), + monoprop::mpi::datatype::get()); + int expected_total = 0; + for (int src = 0; src < P; ++src) { + const int len = leg_len(src, g, round); + if (rc[static_cast(src)] != len || rd[static_cast(src)] != expected_total) { + failures.fetch_add(1); + } + for (int j = 0; j < len; ++j) { + if (recv[static_cast(expected_total + j)] != tag(src, g, j)) { + failures.fetch_add(1); + } + } + expected_total += len; + } + if (static_cast(recv.size()) != expected_total) { + failures.fetch_add(1); + } + } + }); + for (const auto &e : errs) { + BOOST_CHECK(e == nullptr); + } + BOOST_CHECK_EQUAL(failures.load(), 0); +} + +// Same asymmetric legs through the count-exchange + flat-alltoallv pair, which uses the other +// sizing entry point (recv counts published per partition rather than resolved from the wire). +BOOST_AUTO_TEST_CASE(hybrid_comm_alltoall_counts_then_alltoallv_asymmetric) { + if (world_size() < 2) { + return; + } + const int R = world_size(); + const int S = 4; + const int P = R * S; + const auto leg_len = [](int src, int dst) { return (src * 5 + dst * 11) % 4; }; + const auto tag = [](int src, int dst, int j) { return ((src * 1000 + dst) * 100) + j; }; + std::atomic failures{0}; + auto errs = run_hybrid(S, [&](HybridComm &hyb, int u) { + const int g = hyb.global_rank(u); + std::vector sc(static_cast(P)), sd(static_cast(P)); + std::vector rc(static_cast(P)), rd(static_cast(P)); + std::vector send; + int off = 0; + for (int d = 0; d < P; ++d) { + const int len = leg_len(g, d); + sc[static_cast(d)] = len; + sd[static_cast(d)] = off; + for (int j = 0; j < len; ++j) { + send.push_back(tag(g, d, j)); + } + off += len; + } + hyb.alltoall_counts(u, sc.data(), rc.data()); + int total = 0; + for (int src = 0; src < P; ++src) { + if (rc[static_cast(src)] != leg_len(src, g)) { + failures.fetch_add(1); + } + rd[static_cast(src)] = total; + total += rc[static_cast(src)]; + } + std::vector recv(static_cast(total)); + hyb.alltoallv(u, + send.data(), + sc.data(), + sd.data(), + recv.data(), + rc.data(), + rd.data(), + sizeof(int), + monoprop::mpi::datatype::get()); + for (int src = 0; src < P; ++src) { + for (int j = 0; j < rc[static_cast(src)]; ++j) { + if (recv[static_cast(rd[static_cast(src)] + j)] != tag(src, g, j)) { + failures.fetch_add(1); + } + } + } + }); + for (const auto &e : errs) { + BOOST_CHECK(e == nullptr); + } + BOOST_CHECK_EQUAL(failures.load(), 0); +} + #endif // monoprop_ENABLE_MPI diff --git a/tests/cpp/partition_barrier_tests.cpp b/tests/cpp/partition_barrier_tests.cpp new file mode 100644 index 00000000..dcc31591 --- /dev/null +++ b/tests/cpp/partition_barrier_tests.cpp @@ -0,0 +1,161 @@ +// 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. + +#include + +#include +#include +#include +#include +#include + +#include "monoprop/detail/mpi/PartitionBarrier.h" + +using monoprop::mpi::PartitionBarrier; +using monoprop::mpi::ShmCommPoisoned; + +namespace { + +// Every participant stamps its own slot with the round number; after the barrier every slot must read as +// stamped, so anyone released early sees a stale one. Tested directly because the two-level path needs +// pinning and >=2 L3 domains to engage through a real ShmComm, which no test host is guaranteed to have. +auto stamp_rounds(int n, const std::vector &groups, int rounds) -> std::vector { + PartitionBarrier barrier(n, groups); + std::vector slots(static_cast(n), -1); + std::vector errs(static_cast(n)); + std::atomic mismatches{0}; + std::vector threads; + threads.reserve(static_cast(n)); + for (int p = 0; p < n; ++p) { + threads.emplace_back([&, p] { + try { + for (int round = 0; round < rounds; ++round) { + slots[static_cast(p)] = round; + barrier.sync(p); + for (int q = 0; q < n; ++q) { + if (slots[static_cast(q)] != round) { + ++mismatches; + } + } + barrier.sync(p); // peers must finish reading before the next stamp overwrites + } + } + catch (...) { + errs[static_cast(p)] = std::current_exception(); + } + }); + } + for (auto &t : threads) { + t.join(); + } + BOOST_CHECK_EQUAL(mismatches.load(), 0); + return errs; +} + +auto no_errors(const std::vector &errs) -> void { + for (const auto &e : errs) { + BOOST_CHECK(e == nullptr); + } +} + +} // namespace + +// No group ids ⇒ the flat barrier, the path every unpinned run takes. +BOOST_AUTO_TEST_CASE(partition_barrier_flat_releases_together) { + for (const int n : {1, 2, 3, 8}) { + no_errors(stamp_rounds(n, {}, 20)); + } +} + +// Two-level: same semantics, and the group sizes are deliberately uneven and not powers of two, so a +// barrier that mixed up "last in group" with "last overall" would deadlock or release early. +BOOST_AUTO_TEST_CASE(partition_barrier_grouped_releases_together) { + no_errors(stamp_rounds(8, {0, 0, 0, 0, 1, 1, 1, 1}, 20)); + no_errors(stamp_rounds(7, {0, 0, 0, 1, 1, 2, 2}, 20)); + no_errors(stamp_rounds(6, {3, 9, 3, 9, 3, 9}, 20)); // interleaved, non-contiguous domain ids + no_errors(stamp_rounds(5, {0, 1, 1, 1, 1}, 20)); // a domain of one +} + +// One domain is no reason to pay for a root barrier: it must behave exactly like the flat path. +BOOST_AUTO_TEST_CASE(partition_barrier_single_group_is_flat) { + no_errors(stamp_rounds(4, {2, 2, 2, 2}, 20)); +} + +// A group id list that does not cover every participant is a programming error upstream, not a reason +// to hang: it falls back to flat rather than indexing out of range. +BOOST_AUTO_TEST_CASE(partition_barrier_short_group_list_falls_back) { + no_errors(stamp_rounds(4, {0, 1}, 20)); +} + +// poison() must release waiters in BOTH levels: a peer that throws mid-round leaves the others parked +// on their domain's word, not the root's. +BOOST_AUTO_TEST_CASE(partition_barrier_poison_releases_grouped_waiters) { + constexpr int kN = 6; + const std::vector groups{0, 0, 0, 1, 1, 1}; + PartitionBarrier barrier(kN, groups); + std::atomic poisoned_count{0}; + std::vector threads; + threads.reserve(kN - 1); + for (int p = 0; p < kN - 1; ++p) { // participant kN-1 never arrives + threads.emplace_back([&, p] { + try { + barrier.sync(p); + } + catch (const ShmCommPoisoned &) { + ++poisoned_count; + } + }); + } + // Let the arrivers reach the barrier before poisoning, so this exercises the release of parked + // waiters rather than the post-barrier poison check. + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + barrier.poison(); + for (auto &t : threads) { + t.join(); + } + // The last arriver of the complete domain parks at the root; the rest park on their domain word. + BOOST_CHECK_EQUAL(poisoned_count.load(), kN - 1); +} + +// reset() after an aborted round must leave no partial arrival count behind, in any group. +BOOST_AUTO_TEST_CASE(partition_barrier_reset_clears_partial_arrivals) { + constexpr int kN = 4; + const std::vector groups{0, 0, 1, 1}; + PartitionBarrier barrier(kN, groups); + std::thread lone([&] { + try { + barrier.sync(0); // arrives alone, then is released by the poison below + } + catch (const ShmCommPoisoned &) { + } + }); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + barrier.poison(); + lone.join(); + barrier.reset(); + + std::atomic through{0}; + std::vector threads; + threads.reserve(kN); + for (int p = 0; p < kN; ++p) { + threads.emplace_back([&, p] { + barrier.sync(p); + ++through; + }); + } + for (auto &t : threads) { + t.join(); + } + BOOST_CHECK_EQUAL(through.load(), kN); +} From 69a793272a0556a5e145944a8522db0e0ead31aa Mon Sep 17 00:00:00 2001 From: Aaron Miller Date: Wed, 29 Jul 2026 09:36:47 +0200 Subject: [PATCH 3/3] =?UTF-8?q?docs(parallelism):=20=F0=9F=93=9D=20documen?= =?UTF-8?q?t=20monoprop=5FCOMM=5FPROFILE=20and=20the=20two-level=20barrier?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both are new behaviour from the HybridComm floor work: a runtime knob belongs in the environment-variable table, and the barrier's L3-domain grouping is a performance property a reader tuning partition placement needs to know about. Assisted-by: ClaudeCode:claude-opus-5 --- docs/content/docs/features/parallelism.mdx | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/content/docs/features/parallelism.mdx b/docs/content/docs/features/parallelism.mdx index 8ebacb7b..943626b8 100644 --- a/docs/content/docs/features/parallelism.mdx +++ b/docs/content/docs/features/parallelism.mdx @@ -22,6 +22,9 @@ gives each partition a pinned worker thread that runs it serially. Each partitio keeps its own small, cache-resident term index; a gate is applied to all partitions at once, synchronised by a lightweight barrier, and anticommuting terms whose partner lives in another partition are resolved through a per-gate exchange. +When the partitions are pinned and span more than one L3 domain, that barrier fans in +within each domain and then across domains, so both the arrival counter and the +release store stay inside one L3 slice instead of crossing the socket interconnect. ## Runtime environment variables @@ -30,6 +33,7 @@ whose partner lives in another partition are resolved through a per-gate exchang | `monoprop_NUM_THREADS` | one partition per physical core | Caps the number of partitions. Set it to run fewer partitions than cores. | | `monoprop_PARTITIONS` | `auto` | `auto` = one partition per core (capped by `monoprop_NUM_THREADS`); an integer `N` = exactly `N` partitions; `off` = one partition holding the whole operator. | | `monoprop_PARTITION_PINNING` | `on` | `0`/`false`/`no` disables pinning each partition to a core. Has an effect only on Linux. | +| `monoprop_COMM_PROFILE` | `off` | `1`/`true`/`yes` makes each transport print one `COMMPROF` line per rank to stderr when it is destroyed, accounting for where its collectives' wall time went. Diagnostic only; off costs one branch per collective phase. | ```bash # Run 8 partitions instead of one-per-core: