perf(mpi): remove the serial O(R·S²) protocol floor from HybridComm

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:

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<int> {
 struct Settings {
     std::optional<int> 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;

src/monoprop/detail/evolution/layer_build/Engine.h

@@ -381,7 +381,16 @@ struct LayerBuildEngine {
         auto resp = resolve_incoming<NumModes>(inc_q, local_op, R, is_leader_pass, matched, combined_size, sink);
         std::vector<int> resp_recv = response_recv_counts();
         std::vector<std::vector<typename Sink::Response>> 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<int>(Sink::kStride))
+            .wait_into(inc_r);
         process_responses<NumModes>(inc_r, src_idx_r, queries_r, R, my_rank, sink);
     }
 

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 <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <cstdio>
+#include <print>
+#include <vector>
+
+// 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<size_t>(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<size_t>(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<double>(slots_.size()),
+                   to_s(total.table_move_ns) / static_cast<double>(slots_.size()),
+                   to_s(p0.mpi_ns),
+                   to_s(p0.barrier_ns),
+                   to_s(peer_barrier_ns) / static_cast<double>(slots_.size() > 1 ? slots_.size() - 1 : 1),
+                   total.n_barriers == 0
+                       ? 0.0
+                       : (static_cast<double>(total.barrier_ns) / static_cast<double>(total.n_barriers)) / 1000.0);
+        std::fflush(stderr);
+    }
+
+    static auto to_s(uint64_t ns) -> double { return static_cast<double>(ns) / 1e9; }
+
+    std::vector<Slot> 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<uint64_t>(
+                std::chrono::duration_cast<std::chrono::nanoseconds>(CommProfile::Clock::now() - start_).count());
+        }
+    }
+
+private:
+    uint64_t *target_;
+    CommProfile::Clock::time_point start_;
+};
+
+} // namespace monoprop::mpi