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:
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<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;
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<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);
     }
 
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 <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
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 <cstdlib>
 #include <cstring>
 #include <exception>
+#include <memory>
 #include <print>
 #include <stdexcept>
 #include <thread>
@@ -29,6 +30,8 @@
 
 #include <mpi.h>
 
+#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<int> &partition_l3_domains = {})
         : parent_(parent),
           s_(n_local_partitions),
           slots_(static_cast<size_t>(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<size_t>(r_));
         pack_off_.resize(rss);
         scatter_off_.resize(rss);
+        const size_t rs = static_cast<size_t>(r_) * static_cast<size_t>(s_);
+        col_send_.resize(rs);
+        col_recv_.resize(rs);
+        tpre_send_.resize(rs);
+        tpre_recv_.resize(rs);
+        rev_send_counts_.resize(static_cast<size_t>(r_));
+        rev_send_displs_.resize(static_cast<size_t>(r_));
+        rev_recv_counts_.resize(static_cast<size_t>(r_));
+        rev_recv_displs_.resize(static_cast<size_t>(r_));
+        mask_words_ = pad_to_line_(static_cast<size_t>((r_ + 63) / 64));
+        send_mask_.assign(static_cast<size_t>(s_) * mask_words_, 0);
+        run_stride_ = pad_to_line_(static_cast<size_t>(r_));
+        run_scratch_.assign(static_cast<size_t>(s_) * run_stride_, 0);
+        if (config::get().comm_profile) {
+            prof_ = std::make_unique<CommProfile>(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 <class T>
     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<size_t>(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<size_t>(b) * static_cast<size_t>(s_)) + static_cast<size_t>(t))
-                                               * static_cast<size_t>(s_)
-                                           + static_cast<size_t>(su);
-                        counts_send_[idx] = slots_[static_cast<size_t>(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<size_t>(a) * static_cast<size_t>(s_) * static_cast<size_t>(s_))
-                                   + (static_cast<size_t>(t) * static_cast<size_t>(s_)) + static_cast<size_t>(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<size_t>(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<char *>(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<size_t>(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<size_t>(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<size_t>(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<size_t>(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<size_t>(rank) * static_cast<size_t>(s_) * static_cast<size_t>(s_))
-                                    + (static_cast<size_t>(t) * static_cast<size_t>(s_)) + static_cast<size_t>(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<size_t>(a) * static_cast<size_t>(s_) * static_cast<size_t>(s_))
-                                 + (static_cast<size_t>(t) * static_cast<size_t>(s_)) + static_cast<size_t>(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<size_t>(checked_int_(total)));
         }
-        recv.resize(static_cast<size_t>(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<char *>(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<size_t>(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<size_t>(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<size_t>(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<size_t>(send_total) * elem);
+            grow_(stage_recv_, static_cast<size_t>(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<const char *>(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<size_t>(forward_stride);
+                        std::memcpy(stage_send_.data() + off * elem,
+                                    src + static_cast<size_t>(send_displs[g]) * elem,
+                                    static_cast<size_t>(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<char *>(recv);
+        for (int b = 0; b < r_; ++b) {
+            const size_t base = static_cast<size_t>(b) * static_cast<size_t>(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<size_t>(t)] + pack_off_[block_idx_(b, t, u)])
+                                       / static_cast<size_t>(forward_stride);
+                    std::memcpy(dst + static_cast<size_t>(recv_displs[g]) * elem,
+                                stage_recv_.data() + off * elem,
+                                static_cast<size_t>(cnt) * elem);
+                }
+            }
+        }
+    }
+
     template <class T>
     auto allreduce_sum_impl_(int local_partition, T local_val) -> T {
+        count_verb(local_partition);
         Slot &me = slots_[static_cast<size_t>(local_partition)];
         if constexpr (std::is_floating_point_v<T>) {
             me.f64 = static_cast<double>(local_val);
@@ -339,8 +527,9 @@ class HybridComm {
         else {
             me.u64 = static_cast<uint64_t>(local_val);
         }
-        sync();
+        sync(local_partition);
         if (local_partition == 0) {
+            ScopedNs timer{mpi_ns(local_partition)};
             if constexpr (std::is_floating_point_v<T>) {
                 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<T>) {
             out = static_cast<T>(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<size_t>(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<size_t>(s_) - 1) / static_cast<size_t>(s_);
         const size_t lo = std::min(len, static_cast<size_t>(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<size_t>(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<size_t>(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<int>(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<size_t>(b) * static_cast<size_t>(s_) + static_cast<size_t>(t)) * static_cast<size_t>(s_)
                + static_cast<size_t>(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<size_t>(b) * static_cast<size_t>(s_) + static_cast<size_t>(u)) * static_cast<size_t>(s_)
+               + static_cast<size_t>(t);
+    }
+
     template <class V>
     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<size_t>(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 <class RecvCountFn>
-    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<size_t>(u) * mask_words_; }
+    static auto mask_test_(const uint64_t *row, int b) -> bool {
+        return (row[static_cast<size_t>(b) >> 6] & (1ULL << (static_cast<unsigned>(b) & 63U))) != 0;
+    }
+    auto mask_empty_(const uint64_t *row) const -> bool {
+        const size_t live = static_cast<size_t>((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<size_t>(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<size_t>(su)].send_counts[b * s_ + t];
-                    recv_sum += recv_count(t, b, su);
+                if (counts[t] != 0) {
+                    row[static_cast<size_t>(b) >> 6] |= 1ULL << (static_cast<unsigned>(b) & 63U);
+                    break;
                 }
             }
-            mpi_send_counts_[static_cast<size_t>(b)] = checked_int_(send_sum);
-            mpi_recv_counts_[static_cast<size_t>(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<size_t>(b)] = checked_int_(send_running);
-            mpi_recv_displs_[static_cast<size_t>(b)] = checked_int_(recv_running);
-            send_running += mpi_send_counts_[static_cast<size_t>(b)];
-            recv_running += mpi_recv_counts_[static_cast<size_t>(b)];
-        }
-        const size_t total_send = static_cast<size_t>(checked_int_(send_running));
-        const size_t total_recv = static_cast<size_t>(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<size_t>(mpi_send_displs_[static_cast<size_t>(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<size_t>(slots_[static_cast<size_t>(u)].send_counts[b * s_ + t]);
-                }
+                dst[t] = row[t];
             }
         }
-        for (int a = 0; a < r_; ++a) {
-            size_t cur = static_cast<size_t>(mpi_recv_displs_[static_cast<size_t>(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<size_t>(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 <class RecvCountFn>
+    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<size_t>(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<size_t>(counts[b * s_ + t]);
+                }
+            }
+            for (int b = 0; b < r_; ++b) {
+                col_send_[static_cast<size_t>(b) * static_cast<size_t>(s_) + static_cast<size_t>(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<size_t>(recv_count(t, a, su));
+                    scatter_off_[block_idx_(b, t, su)] = got;
+                    got += static_cast<size_t>(recv_count(t, b, su));
                 }
+                col_recv_[static_cast<size_t>(b) * static_cast<size_t>(s_) + static_cast<size_t>(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<size_t>(b) * static_cast<size_t>(s_);
+                for (int k = 0; k < s_; ++k) {
+                    send_sum += static_cast<long long>(col_send_[base + static_cast<size_t>(k)]);
+                    recv_sum += static_cast<long long>(col_recv_[base + static_cast<size_t>(k)]);
+                }
+                mpi_send_counts_[static_cast<size_t>(b)] = checked_int_(send_sum);
+                mpi_recv_counts_[static_cast<size_t>(b)] = checked_int_(recv_sum);
+                mpi_send_displs_[static_cast<size_t>(b)] = checked_int_(send_running);
+                mpi_recv_displs_[static_cast<size_t>(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<size_t>(send_running);
+                size_t cur_r = static_cast<size_t>(recv_running);
+                for (int k = 0; k < s_; ++k) {
+                    tpre_send_[base + static_cast<size_t>(k)] = cur_s;
+                    cur_s += col_send_[base + static_cast<size_t>(k)];
+                    tpre_recv_[base + static_cast<size_t>(k)] = cur_r;
+                    cur_r += col_recv_[base + static_cast<size_t>(k)];
+                }
+                send_running += mpi_send_counts_[static_cast<size_t>(b)];
+                recv_running += mpi_recv_counts_[static_cast<size_t>(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<size_t>(checked_int_(send_running)) * elem);
+            grow_(stage_recv_, static_cast<size_t>(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<size_t>(u)].send_counts;
         const int *my_send_displs = slots_[static_cast<size_t>(u)].send_displs;
         for (int b = 0; b < r_; ++b) {
+            const size_t base = static_cast<size_t>(b) * static_cast<size_t>(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<size_t>(t)] + pack_off_[block_idx_(b, t, u)];
+                    std::memcpy(stage_send_.data() + off * elem,
                                 src + static_cast<size_t>(my_send_displs[b * s_ + t]) * elem,
                                 static_cast<size_t>(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<size_t>(a) * static_cast<size_t>(s_) + static_cast<size_t>(t)]
+               + scatter_off_[block_idx_(a, t, su)];
+    }
+
     static auto checked_int_(long long v) -> int {
         if (v < 0 || v > static_cast<long long>(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<CommProfile> prof_;
 
     MPI_Comm parent_;
     int s_;
@@ -538,12 +865,32 @@ class HybridComm {
     std::vector<int> mpi_send_displs_;
     std::vector<int> mpi_recv_counts_;
     std::vector<int> 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<size_t> pack_off_;
     std::vector<size_t> scatter_off_;
+    // [R*S] per-(rank, dest partition) column totals and their prefix, the only globally-reduced part.
+    std::vector<size_t> col_send_;
+    std::vector<size_t> col_recv_;
+    std::vector<size_t> tpre_send_;
+    std::vector<size_t> 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<uint64_t> send_mask_;
+    std::vector<size_t> run_scratch_;
+    size_t mask_words_ = 8;
+    size_t run_stride_ = 8;
     // Aggregated MPI payload staging, HWM-sized.
     std::vector<std::byte> stage_send_;
     std::vector<std::byte> 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<int> rev_send_counts_;
+    std::vector<int> rev_send_displs_;
+    std::vector<int> rev_recv_counts_;
+    std::vector<int> 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<double> 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 <class T>
 inline auto begin_alltoallv(const std::vector<std::vector<T>> &send_data,
                             Comm comm,
                             bool skip_self = false,
-                            const std::vector<int> *known_recv_counts = nullptr) -> PendingAlltoallv<T> {
+                            const std::vector<int> *known_recv_counts = nullptr,
+                            bool reverse_of_previous = false,
+                            int forward_stride = 1) -> PendingAlltoallv<T> {
     const int num_ranks = size(comm);
     if (static_cast<int>(send_data.size()) != num_ranks) {
         throw CollectiveArgumentError(
@@ -342,15 +349,29 @@ inline auto begin_alltoallv(const std::vector<std::vector<T>> &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<T>::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<T>::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<T>::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 <algorithm>
 #include <atomic>
 #include <stdexcept>
 #include <thread>
+#include <vector>
 
 #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<int> &group_of = {}) : participants_(participants) {
+        if (static_cast<int>(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<int> domains;
+        group_of_.resize(static_cast<size_t>(participants));
+        for (int p = 0; p < participants; ++p) {
+            const auto it = std::find(domains.begin(), domains.end(), group_of[static_cast<size_t>(p)]);
+            if (it == domains.end()) {
+                group_of_[static_cast<size_t>(p)] = static_cast<int>(domains.size());
+                domains.push_back(group_of[static_cast<size_t>(p)]);
+            }
+            else {
+                group_of_[static_cast<size_t>(p)] = static_cast<int>(it - domains.begin());
+            }
+        }
+        if (domains.size() < 2) {
+            group_of_.clear();
+            return; // flat
+        }
+        groups_ = static_cast<int>(domains.size());
+        group_arrived_ = std::vector<Count>(domains.size());
+        group_gen_ = std::vector<Gen>(domains.size());
+        group_size_.assign(domains.size(), 0);
+        for (int p = 0; p < participants; ++p) {
+            ++group_size_[static_cast<size_t>(group_of_[static_cast<size_t>(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<size_t>(group_of_[static_cast<size_t>(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<unsigned> &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<int> v{0};
+    };
+    struct alignas(64) Gen {
+        std::atomic<unsigned> v{0};
+    };
+
     int participants_;
+    int groups_ = 0; // < 2 ⇒ flat: arrived_/gen_ count participants, not domains
+    std::vector<int> group_of_;
+    std::vector<int> group_size_;
+    std::vector<Count> group_arrived_;
+    std::vector<Gen> group_gen_;
+    // Flat barrier, or (two-level) the root barrier the domains' last arrivers meet at.
     alignas(64) std::atomic<int> arrived_{0};
     alignas(64) std::atomic<unsigned> gen_{0};
     alignas(64) std::atomic<bool> 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 <type_traits>
 #include <vector>
 
+#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<size_t>(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<int> &partition_l3_domains = {})
+        : n_(n),
+          slots_(static_cast<size_t>(n)),
+          barrier_(n, partition_l3_domains) {
+        if (config::get().comm_profile) {
+            prof_ = std::make_unique<CommProfile>(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<size_t>(rank)].counts = send_counts;
-        sync();
-        for (int s = 0; s < n_; ++s) {
-            recv_counts[s] = slots_[static_cast<size_t>(s)].counts[rank];
+        sync(rank);
+        {
+            ScopedNs timer{par_table_ns(rank)};
+            for (int s = 0; s < n_; ++s) {
+                recv_counts[s] = slots_[static_cast<size_t>(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<size_t>(rank)];
         me.ptr = send;
         me.displs = send_displs;
-        sync();
-        auto *dst = static_cast<char *>(recv);
-        for (int s = 0; s < n_; ++s) {
-            const Slot &src = slots_[static_cast<size_t>(s)];
-            const auto *sp = static_cast<const char *>(src.ptr);
-            const auto count = static_cast<size_t>(recv_counts[s]);
-            if (count == 0) {
-                continue;
+        sync(rank);
+        {
+            ScopedNs timer{move_ns(rank)};
+            auto *dst = static_cast<char *>(recv);
+            for (int s = 0; s < n_; ++s) {
+                const Slot &src = slots_[static_cast<size_t>(s)];
+                const auto *sp = static_cast<const char *>(src.ptr);
+                const auto count = static_cast<size_t>(recv_counts[s]);
+                if (count == 0) {
+                    continue;
+                }
+                std::memcpy(dst + static_cast<size_t>(recv_displs[s]) * elem,
+                            sp + static_cast<size_t>(src.displs[rank]) * elem,
+                            count * elem);
             }
-            std::memcpy(dst + static_cast<size_t>(recv_displs[s]) * elem,
-                        sp + static_cast<size_t>(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<size_t>(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<size_t>(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<size_t>(checked_mpi_count(total, "Total recv count")));
         }
-        recv.resize(static_cast<size_t>(checked_mpi_count(total, "Total recv count")));
-        auto *dst = reinterpret_cast<char *>(recv.data());
-        for (int s = 0; s < n_; ++s) {
-            const Slot &src = slots_[static_cast<size_t>(s)];
-            const auto count = static_cast<size_t>(recv_counts[s]);
-            if (count == 0) {
-                continue;
+        {
+            ScopedNs timer{move_ns(rank)};
+            auto *dst = reinterpret_cast<char *>(recv.data());
+            for (int s = 0; s < n_; ++s) {
+                const Slot &src = slots_[static_cast<size_t>(s)];
+                const auto count = static_cast<size_t>(recv_counts[s]);
+                if (count == 0) {
+                    continue;
+                }
+                std::memcpy(dst + static_cast<size_t>(recv_displs[s]) * sizeof(T),
+                            reinterpret_cast<const char *>(src.ptr) + static_cast<size_t>(src.displs[rank]) * sizeof(T),
+                            count * sizeof(T));
             }
-            std::memcpy(dst + static_cast<size_t>(recv_displs[s]) * sizeof(T),
-                        reinterpret_cast<const char *>(src.ptr) + static_cast<size_t>(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 <class T>
@@ -129,7 +162,8 @@ class ShmComm {
         else {
             me.u64 = static_cast<uint64_t>(local_val);
         }
-        sync();
+        count_verb(rank);
+        sync(rank);
         T acc{};
         for (int s = 0; s < n_; ++s) {
             const Slot &src = slots_[static_cast<size_t>(s)];
@@ -140,30 +174,34 @@ class ShmComm {
                 acc += static_cast<T>(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<size_t>(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<size_t>(n_) - 1) / static_cast<size_t>(n_);
-        const size_t lo = std::min(len, static_cast<size_t>(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<size_t>(s)].vec[k];
-            }
-            for (int s = 0; s < n_; ++s) { // publish the same bits into every rank's buffer
-                slots_[static_cast<size_t>(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<size_t>(n_) - 1) / static_cast<size_t>(n_);
+            const size_t lo = std::min(len, static_cast<size_t>(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<size_t>(s)].vec[k];
+                }
+                for (int s = 0; s < n_; ++s) { // publish the same bits into every rank's buffer
+                    slots_[static_cast<size_t>(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<CommProfile> prof_;
     int n_;
     std::vector<Slot> 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<CpuSet> &sets) -> std::vector<int> {
+    if (sets.empty()) {
+        return {};
+    }
+    const auto cores = enumerate_physical_cores();
+    std::vector<int> 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<CpuSet> {
     return {};
 }
+inline auto cpuset_l3_domains(const std::vector<CpuSet> & /*sets*/) -> std::vector<int> {
+    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<size_t>(n_partitions)),
           errs_(static_cast<size_t>(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<size_t>(src.n_)),
           errs_(static_cast<size_t>(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<int> 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<mpi::HybridComm>(parent_.mpi, n_);
+            hyb_ = std::make_unique<mpi::HybridComm>(parent_.mpi, n_, domains);
             return;
         }
 #endif
-        shm_ = std::make_unique<mpi::ShmComm>(n_);
+        shm_ = std::make_unique<mpi::ShmComm>(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<int> 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<int> recv;
+        std::vector<int> rc(static_cast<size_t>(P)), rd(static_cast<size_t>(P));
+        for (int round = 0; round < rounds; ++round) {
+            std::vector<int> send;
+            std::vector<int> sc(static_cast<size_t>(P)), sd(static_cast<size_t>(P));
+            int off = 0;
+            for (int d = 0; d < P; ++d) {
+                const int len = leg_len(g, d, round);
+                sc[static_cast<size_t>(d)] = len;
+                sd[static_cast<size_t>(d)] = off;
+                for (int j = 0; j < len; ++j) {
+                    send.push_back(tag(g, d, j));
+                }
+                off += len;
+            }
+            hyb.alltoallv_resolve<int>(u,
+                                       send.data(),
+                                       sc.data(),
+                                       sd.data(),
+                                       recv,
+                                       rc.data(),
+                                       rd.data(),
+                                       sizeof(int),
+                                       monoprop::mpi::datatype<int>::get());
+            int expected_total = 0;
+            for (int src = 0; src < P; ++src) {
+                const int len = leg_len(src, g, round);
+                if (rc[static_cast<size_t>(src)] != len || rd[static_cast<size_t>(src)] != expected_total) {
+                    failures.fetch_add(1);
+                }
+                for (int j = 0; j < len; ++j) {
+                    if (recv[static_cast<size_t>(expected_total + j)] != tag(src, g, j)) {
+                        failures.fetch_add(1);
+                    }
+                }
+                expected_total += len;
+            }
+            if (static_cast<int>(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<int> failures{0};
+    auto errs = run_hybrid(S, [&](HybridComm &hyb, int u) {
+        const int g = hyb.global_rank(u);
+        std::vector<int> sc(static_cast<size_t>(P)), sd(static_cast<size_t>(P));
+        std::vector<int> rc(static_cast<size_t>(P)), rd(static_cast<size_t>(P));
+        std::vector<int> send;
+        int off = 0;
+        for (int d = 0; d < P; ++d) {
+            const int len = leg_len(g, d);
+            sc[static_cast<size_t>(d)] = len;
+            sd[static_cast<size_t>(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<size_t>(src)] != leg_len(src, g)) {
+                failures.fetch_add(1);
+            }
+            rd[static_cast<size_t>(src)] = total;
+            total += rc[static_cast<size_t>(src)];
+        }
+        std::vector<int> recv(static_cast<size_t>(total));
+        hyb.alltoallv(u,
+                      send.data(),
+                      sc.data(),
+                      sd.data(),
+                      recv.data(),
+                      rc.data(),
+                      rd.data(),
+                      sizeof(int),
+                      monoprop::mpi::datatype<int>::get());
+        for (int src = 0; src < P; ++src) {
+            for (int j = 0; j < rc[static_cast<size_t>(src)]; ++j) {
+                if (recv[static_cast<size_t>(rd[static_cast<size_t>(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 <boost/test/unit_test.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <exception>
+#include <thread>
+#include <vector>
+
+#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<int> &groups, int rounds) -> std::vector<std::exception_ptr> {
+    PartitionBarrier barrier(n, groups);
+    std::vector<int> slots(static_cast<size_t>(n), -1);
+    std::vector<std::exception_ptr> errs(static_cast<size_t>(n));
+    std::atomic<int> mismatches{0};
+    std::vector<std::thread> threads;
+    threads.reserve(static_cast<size_t>(n));
+    for (int p = 0; p < n; ++p) {
+        threads.emplace_back([&, p] {
+            try {
+                for (int round = 0; round < rounds; ++round) {
+                    slots[static_cast<size_t>(p)] = round;
+                    barrier.sync(p);
+                    for (int q = 0; q < n; ++q) {
+                        if (slots[static_cast<size_t>(q)] != round) {
+                            ++mismatches;
+                        }
+                    }
+                    barrier.sync(p); // peers must finish reading before the next stamp overwrites
+                }
+            }
+            catch (...) {
+                errs[static_cast<size_t>(p)] = std::current_exception();
+            }
+        });
+    }
+    for (auto &t : threads) {
+        t.join();
+    }
+    BOOST_CHECK_EQUAL(mismatches.load(), 0);
+    return errs;
+}
+
+auto no_errors(const std::vector<std::exception_ptr> &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<int> groups{0, 0, 0, 1, 1, 1};
+    PartitionBarrier barrier(kN, groups);
+    std::atomic<int> poisoned_count{0};
+    std::vector<std::thread> 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<int> 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<int> through{0};
+    std::vector<std::thread> 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);
+}