Rings, Trees, and Butterflies: HPX Collectives (Part 1)

Eight machines. Each one holds a number. Every one of them needs the sum of all eight. In HPX that is one line:

double total = all_reduce(comm, my_value, std::plus{}).get();

Every machine calls it, every machine gets back the same total, and nobody wrote a single send or recv. This is a collective: an operation that a whole group of machines performs together, the primitive underneath MPI_Allreduce and a gradient reduction in a distributed training run. The GPU investigation followed arithmetic onto one device. The networking investigation got bytes across one wire. This call has to move values among many machines, combine them, and leave every participant with the same answer.

I ended up caring about the cost because I added hierarchical all-reduce, all-gather, and all-to-all paths to HPX. The commits are in the source history, so that claim is inspectable rather than biographical decoration. To add them I first had to understand what the line above actually does. The communicator-backed collectives reduce to one server type, one synchronization gadget, and small functions that say how values enter and leave. That statement needs the qualifier. HPX also has latches, SPMD blocks, direct action broadcasts, and channel communicators that do not pass through this exact server.

The distinction became important as soon as I tried to predict performance from the server. A flat communicator does make every participant meet at one component. A tree changes that path. A ring and a butterfly change it again. All four can return the same eight sums while moving different values, sending different byte counts, and waiting for different ranks. The output alone cannot tell them apart.

the handle named one box

The line above hides its most important argument. That comm is a communicator, and you make one like this:

auto comm = create_communicator(
    "/my_app/gradient_sync",   // basename
    num_sites_arg(8),          // how many machines
    this_site_arg(rank));      // which one am I

A communicator is not a data structure sitting in your process. It is an HPX component, a live object registered in HPX’s global address space (AGAS) under that string, the basename. The string is the rendezvous point: every machine that wants to join this particular collective looks the component up by the same basename, and HPX hands them all a handle to the one object, which physically lives at a single locality (the root_site, defaulting to zero). So eight machines scattered across a cluster all end up talking to one named box that lives on machine zero.

What does the box do when they talk to it? It counts. Inside the communicator_server is a gadget HPX calls an and_gate: a latch that stays shut until all num_sites participants have checked in, then makes the shared readiness state complete. Each machine checks in by invoking an action on the component, either a set for a participant that only contributes or a get for a participant that needs a result. Their continuations cannot obtain the collective result before the last required arrival.

And the collective itself is almost nothing. It is a pair of functions handed to a generic engine called handle_data: a step, run once per machine as it checks in, and a finalizer, invoked for each waiting result after everyone has arrived. Expensive final work is guarded so it happens once. Here is the core of all_reduce from the pinned source:

// step: each machine writes its value into its own slot
[&t](auto& data, std::size_t which) {
    data[which] = HPX_FORWARD(T, t);
},
// finalizer: after all have checked in, combine once
[op = HPX_FORWARD(F, op)](
    auto& data, bool& data_available, std::size_t) mutable {
    if (!data_available && data.size() > 1) {
        auto it = data.begin();
        data[0] = hpx::reduce(
            ++it, data.end(), data[0], HPX_MOVE(op));
        data_available = true;
    }
    return data[0];
}

Machine which writes its value into data[which]. Once the gate becomes ready, the first finalizer that sees data_available == false folds the vector with your op, here std::plus, and stores the result in data[0]. Later callbacks return that stored value. Change the step and finalizer and the communicator server can implement another value collective. broadcast copies the root’s value; gather returns the vector instead of reducing it; communicator-backed barrier uses no payload. Naming, address-space plumbing, generation checks, and operation-mismatch failures remain shared machinery. Once I saw that, the communicator collectives stopped looking like unrelated functions and started looking like one server with several value paths.

the direction was hidden in the function name

There is a second thing that confused me before it clicked, and the HPX docs are quiet about it. A collective is symmetric in its effect, everyone ends up with the answer, but it is not symmetric in the code. Look at reduce. It comes in two halves:

reduce_here (comm, value, std::plus{}, this_site);  // I am the root; I get the result
reduce_there(comm, value,             this_site);   // I am a participant; I send and get void

The root calls reduce_here, supplies the combining op, and receives the reduced value. Everyone else calls reduce_there, supplies only their value, and gets back void. Same collective, two sides. And the pattern repeats across the whole rooted family, with a naming convention worth memorizing because it is the key to the entire API:

• reduce_here / reduce_there
• gather_here / gather_there
• broadcast_to / broadcast_from
• scatter_to / scatter_from

The _here and _to side is always the root, the one machine that owns the result or the source. The _there and _from side is everyone else. The convenience overloads, plain reduce(comm, ...), dispatch to one side or the other by checking whether this_site == root_site, so you usually don’t type the halves yourself. But they are there, and they are the honest shape of the operation. Three argument tags name the players: this_site is me, root_site is who owns the box, and that_site is my specific peer, which only the point-to-point operations use. Once you read every collective as “am I the root or a participant,” fifteen functions collapse into one question.

four values made the names concrete

So here is everything that is implemented, grouped not by name but by what the step and finalizer do. This is a tour; I’ll linger where it’s interesting and move fast where it isn’t.

Rooted (data flows to or from one machine). broadcast sends the root’s one value to everyone. scatter takes the root’s vector of N elements and gives machine i element i (the root’s data, sliced). gather is its mirror: each machine’s value flows in, the root gets the vector back. reduce is gather with a fold on top, every value combined by your op into one result at the root. These four are the atoms. The finalizer is the whole difference: copy, slice, collect, or reduce.

Symmetric (everyone gets the answer). all_gather has the effect of gather followed by broadcast, so every machine ends with the full vector of everyone’s values. all_reduce has the effect of reduce followed by broadcast, so everyone ends with the single combined result. That composition is literally how the hierarchical all-reduce is built, which is the next section. The third, all_to_all, has the widest peer fan-out when every source-to-destination piece has a fixed size: every machine brings a vector of N pieces, and source i’s piece j is delivered to destination j. It is a distributed matrix transpose, each machine handing every other machine one specific piece and walking away with a whole column.

Scans (prefix reductions). inclusive_scan gives machine i the fold of values from machines 0 through i, itself included; exclusive_scan gives it 0 through i-1, itself excluded, matching MPI_Exscan down to the detail that machine zero’s result is undefined and there’s an optional init value to seed it. Same box, a finalizer that computes a running combine instead of a single one. I’ll be honest that these two are the corners of the library I’ve spent the least time in.

Synchronization (no payload at all). The communicator-backed barrier is the degenerate value collective: nobody brings data, and readiness waits for all N. HPX also has a reusable hpx::distributed::barrier class you construct once and wait() on repeatedly. latch is a distributed countdown: initialize it to a count, let machines count_down, and block until it hits zero. Their public synchronization effect is related, but their implementation paths are not all the communicator server with a payload removed.

The three that share the words but not the lineage. This one is a trap I want to defuse, because it caught me. There are functions called broadcast_direct, reduce_direct, and fold, and they are not the communicator collectives above. They belong to an older HPX layer (hpx::lcos) and they operate on a plain std::vector<hpx::id_type> of targets: broadcast_direct invokes an action on every target and collects the results, reduce_direct folds those action-results pairwise, fold chains them with an initial value. No communicator, no basename, no gate, a compile-time fan-out constant instead. They broadcast a function call across a list of objects, where the real collectives shuffle values across a synchronized group. Same English words, different machine. If you go looking in the source, don’t conflate them; I did, briefly, and it cost me an afternoon.

Point-to-point and SPMD, the persistent siblings. channel_communicator is not a collective at all; it is MPI-style directed messaging. You set a value to a named peer and get a value from a named peer, addressed by that_site and an optional tag, over an endpoint you can reuse. It is the one place that_site earns its keep. spmd_block hands each image a persistent handle inside an SPMD region and exposes explicit sync_all and sync_images calls. The one-shot collectives are fused data movements; these facilities support a longer conversation.

That is not one implementation family. The communicator-backed value collectives share the server below. The direct action operations, channel communicator, SPMD block, latch, and reusable barrier have different paths. The correction matters because a performance claim about the communicator server cannot be applied to all fifteen names merely because the documentation lists them together.

every arrival reached the same lock

Go back to what a flat communicator actually is: a single component, living on a single locality, that every participant checks into. Every one of the N sites invokes an action on that component. Its and_gate accounts for N arrivals. Its mutex protects the shared data and operation state. The finalizer runs after the last arrival and folds the stored values there. The component is correct at eight sites and still correct at eight hundred. Its incoming messages, stored values, callbacks, reduction work, and root-link traffic all grow with N.

Calling this a flaw was too broad. It is a scaling hypothesis derived from the execution path. The source proves centralization. It does not prove the site count at which that centralization loses, because message startup, payload size, transport aggregation, memory copies, scheduling, and topology all affect the crossover. That threshold needs a model and measurements.

the first fix grew a tree of boxes

The fix is the one every distributed system eventually reaches for: stop funneling everything into one box, and build a tree of boxes. That is what create_hierarchical_communicator does, and it is the part of this story with my copyright on it:

auto hcomm = create_hierarchical_communicator(
    "/my_app/gradient_sync",
    num_sites_arg(800),
    this_site_arg(rank),
    arity_arg(2));          // fan-out of the tree

A hierarchical_communicator is not one communicator. It is a vector of flat communicators, one per level of a tree with fan-out arity (default 2). The 800 machines are partitioned into arity balanced groups, each group into arity subgroups, and so on down; the leftmost machine of each group is its representative and carries the group’s value one level up. The partition itself is deliberately even, the first num_sites % arity groups get the ceiling and the rest get the floor, so no branch of the tree is lopsided. Each level is just a small flat communicator, the exact machine from the first half of this post, only now no single box ever sees more than arity children.

With the tree in hand, a hierarchical all_reduce is almost embarrassingly literal. It is reduce up the tree, then broadcast back down:

if (this_site == root_site) {
    auto reduced = reduce_here(hcomm, value, op, this_site, reduce_gen).get();
    return broadcast_to(hcomm, reduced, this_site, broadcast_gen);
} else {
    reduce_there(hcomm, value, op, this_site, reduce_gen).get();
    return broadcast_from<T>(hcomm, this_site, broadcast_gen);
}

Read that against the earlier claim that “all_reduce is reduce followed by broadcast.” Here it is, not as a mnemonic but as the actual implementation, threaded through the tree so that reduction climbs level by level to the root and the answer flows back down the same branches. The fan-in at any single box drops from N to arity; the depth grows to about log-base-arity of N. Eight hundred machines with a binary tree is a depth of ten, and no box ever handles more than two children instead of eight hundred.

One detail in that code is a small trick I’m fond of. The reduce phase and the broadcast phase both run over the same tree of communicators, one right after the other, and a communicator distinguishes repeated operations by a generation number. A single user-level generation k maps to two internal generations: 2k-1 for reduce and 2k for broadcast. The mapping needs no extra communication phase, and it keeps the two phases from stepping on each other. The price is that the hierarchical path needs an explicit generation from you because it has to reserve the pair. That’s a rough edge, and I’d rather show it than hide it.

all_gather follows a gather-up-broadcast-down skeleton. The thing traveling back down the tree is the full vector of everyone’s values, so the broadcast phase grows with the participant count when each participant contributes a fixed-size value. all_to_all, the transpose, is gnarlier still, with subtree collection, exchange among representatives, and distribution back down. These are the collectives I wired through the tree, and the all-to-all hierarchy took the longest to get right.

Then there is the knob that I think is the most honest line in the whole contribution:

using flat_fallback_threshold_arg =
    detail::argument_type<detail::flat_fallback_threshold_tag, 16>;

For fewer than sixteen sites, not sixteen or fewer, the builder replaces the requested arity with the site count. Its leaf condition then produces one flat communicator. Hierarchical all_reduce notices arity >= num_sites and calls the flat operation directly while advancing the gate by two generations. That second detail is necessary. The next collective on the same hierarchical communicator must see the same generation position whether the previous call used a tree or collapsed to one box.

The implementation described here is pinned to HPX source checkout acf2394d5f3f9ea0c8622cec0c0099f61d525a94. The built tests report HPX 2.0.0 at revision e6324bd801. The checkout is newer than the binary, but both include the hierarchical all-reduce, all-gather, generation mapping, and flat fallback discussed here. The source probe checks that the relevant commits are ancestors of the checkout and that the collectives directory has no uncommitted changes.

Sixteen is still a guess. The source shows exactly what the threshold does. Passing unit tests shows that the flat and tree paths return the same values for the tested cases. Neither fact says where one path becomes faster.

the sum needed a byte ledger

The smallest example hid the most important variable. A scalar double is eight bytes. A gradient can be hundreds of megabytes. Both calls have the same C++ type shape:

auto result = all_reduce(comm, local_value, std::plus{}).get();

The collective cannot be costed from that line until the payload has a size. Let:

• (P) be the number of ranks;
• (M) be the bytes contributed by each rank;
• (\alpha) be the time to start one message and make it useful at its peer;
• (\beta) be seconds per byte on the relevant path.

The familiar first model for one message is:

$$T_{\text{message}} = \alpha + \beta M$$

The units catch mistakes. (\alpha) is seconds. (\beta) is seconds per byte. Multiplying (\beta) by (M) bytes produces seconds. This is not a law of a network. It is a local approximation that ignores contention, software progress, packetization, cache copies, protocol changes, topology, and computation. It is still useful because each algorithm gives (\alpha) and (\beta) different coefficients.

Big-O notation is too coarse here. A ring and a recursive-halving algorithm both move (O(M)) bytes per rank. One takes (2(P-1)) rounds. The other takes (2\log_2P) rounds when (P) is a power of two. Those coefficients decide which path wins at the sizes in front of us.

Four ledgers are needed:

1. messages or rounds on the critical path;
2. payload bytes sent by each rank;
3. payload bytes through the busiest rank or link;
4. temporary bytes stored while partial results exist.

Counting only total cluster traffic can hide a hot root. Counting only root traffic can hide that every rank sends a full vector at every tree level. Counting only rounds can make a ring look terrible even when it keeps every link carrying small, useful chunks.

rank zero did the obvious thing

The first runnable implementation does exactly what the flat server suggests. Every nonzero rank sends its complete array to rank zero. Rank zero adds each arrival into its output. It then sends the complete result back to every rank.

if (rank == 0) {
    memcpy(out, input, count * sizeof(double));
    for (int source = 1; source < size; ++source) {
        MPI_Recv(tmp, count, MPI_DOUBLE, source, 100,
                 MPI_COMM_WORLD, MPI_STATUS_IGNORE);
        for (int i = 0; i < count; ++i)
            out[i] += tmp[i];
    }
    for (int destination = 1; destination < size; ++destination)
        MPI_Send(out, count, MPI_DOUBLE, destination, 101, MPI_COMM_WORLD);
} else {
    MPI_Send(input, count, MPI_DOUBLE, 0, 100, MPI_COMM_WORLD);
    MPI_Recv(out, count, MPI_DOUBLE, 0, 101,
             MPI_COMM_WORLD, MPI_STATUS_IGNORE);
}

This is real C from the probe, shortened only by removing error checks. It compiles with Open MPI 5.0.9 and Apple clang 15. Every output element is checked after every timed call.

In the serial root model, rank zero receives (P-1) arrays and sends (P-1) arrays:

$$T_{\text{naive}} \approx 2(P-1)(\alpha + \beta M)$$

The busiest rank moves (2(P-1)M) payload bytes. Every other rank moves (2M). The imbalance grows with (P). If incoming transfers overlap perfectly, elapsed time need not equal the serial expression. The root link and memory path still have to absorb all bytes, so perfect overlap cannot create unbounded root bandwidth.

For eight ranks and a one-megabyte array, the root payload ledger is:

$$2(8-1)(1\ \text{MiB}) = 14\ \text{MiB}$$

Each other rank sends one mebibyte and receives one. The algorithm asks the root to move seven times as many payload bytes as a participant. That is the centralization the HPX source made visible.

The naive path has one property worth keeping in mind. It works for any positive rank count and any array length. It does not need equal chunks or a power-of-two group. It also gives rank zero one fixed reduction order. Better communication schedules often give up one or more of those conveniences.

the tree removed the crowd from the root

With four ranks holding [1,1,1,1], [2,2,2,2], [3,3,3,3], and [4,4,4,4], the binomial tree probe produces this state:

initial
r0 [1 1 1 1]   r1 [2 2 2 2]   r2 [3 3 3 3]   r3 [4 4 4 4]

reduce distance 1
r0 [3 3 3 3]   r1 [2 2 2 2]   r2 [7 7 7 7]   r3 [4 4 4 4]

reduce distance 2
r0 [10 10 10 10]

broadcast distance 2, then distance 1
r0 [10 10 10 10]   r1 [10 10 10 10]
r2 [10 10 10 10]   r3 [10 10 10 10]

At the first reduction step, rank 1 sends to rank 0 and rank 3 sends to rank 2. Both pairs work concurrently. At the second step, rank 2 sends its partial sum to rank 0. Broadcast reverses the information flow. No step asks three ranks to send to rank zero.

For a power-of-two rank count, reduction takes (\log_2 P) rounds and broadcast takes another (\log_2 P):

$$T_{\text{tree}} \approx 2\log_2(P)(\alpha + \beta M)$$

The busiest rank sends or receives a full (M)-byte array at each relevant level. The root no longer receives (P-1) complete arrays directly, but the critical path still carries a complete array in every round. With eight ranks and one mebibyte, the model charges six full-array message times, three upward and three downward.

This is the shape of the HPX hierarchical all-reduce. It is not the exact transport schedule because an HPX communicator server, action invocation, future, and gate add their own work. The value dependency is the same: full values reduce upward and the full answer broadcasts downward.

The phrase “all-reduce is reduce followed by broadcast” is semantically correct. If op is valid and every rank participates, the composition returns the right result. It is not a statement of bandwidth optimality. The full (M)-byte value climbs and descends the tree. A large array contains independent chunks that do not all need to visit the same representative.

the first butterfly kept the whole array

A root is not necessary. In recursive doubling, every rank exchanges its current partial result with a partner whose rank differs in one bit.

For four ranks, the first partner is rank xor 1:

r0 exchanges with r1: [1 1 1 1] + [2 2 2 2] = [3 3 3 3]
r1 exchanges with r0: [2 2 2 2] + [1 1 1 1] = [3 3 3 3]
r2 exchanges with r3: [3 3 3 3] + [4 4 4 4] = [7 7 7 7]
r3 exchanges with r2: [4 4 4 4] + [3 3 3 3] = [7 7 7 7]

The next partner is rank xor 2. Each lower pair exchanges with the corresponding upper pair, and every rank obtains [10,10,10,10]. The pair edges form the crossing pattern that gives the butterfly its name.

The implementation is only one loop:

memcpy(out, input, count * sizeof(double));
for (int mask = 1; mask < size; mask <<= 1) {
    int partner = rank ^ mask;
    MPI_Sendrecv(out, count, MPI_DOUBLE, partner, 300 + mask,
                 tmp, count, MPI_DOUBLE, partner, 300 + mask,
                 MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    for (int i = 0; i < count; ++i)
        out[i] += tmp[i];
}

After step (s), each partial result contains contributions from (2^{s+1}) ranks. After (\log_2 P) steps, that set has (P) members. Every rank has the answer without a separate broadcast:

$$T_{\text{doubling}} \approx \log_2(P)(\alpha + \beta M)$$

This halves the tree’s modeled rounds. It also sends the entire array in every round. Per rank payload becomes (M\log_2P). That is excellent for a scalar and expensive for a large gradient.

The probe rejects non-power-of-two rank counts. Production implementations can peel extra ranks, use recursive exchange variants, or select another algorithm. Silently applying rank xor mask to six ranks would eventually name partners that do not exist.

the same sum was not always the same bits

The four integer arrays make reduction order look irrelevant. Integer addition within range is associative:

$$(a+b)+c = a+(b+c)$$

Floating-point addition is not. Consider three values:

1e20, -1e20, 3.14

Reducing left to right can produce:

$$(10^{20} + (-10^{20})) + 3.14 = 3.14$$

Pairing the last two first can lose 3.14 beside the much larger magnitude:

$$10^{20} + ((-10^{20}) + 3.14) = 0$$

The exact result depends on format and rounding, but the mechanism does not. A tree, ring, and butterfly can legally produce different last bits for floating-point sum because they associate operands differently. MPI reduction operations declare whether they are commutative, and implementations may use that information when choosing an order. An operation also has to be associative for the collective’s regrouping to preserve its mathematical meaning.

Subtraction is an immediate counterexample. “Reduce all values with subtraction” has no schedule-independent answer. String concatenation is associative but not commutative, so rank order matters even though parentheses do not. An algorithm that permutes rank order is not valid for every associative operation.

The custom benchmark uses addition and checks with a tolerance because different schedules are expected to round slightly differently. Correctness means agreement with the mathematical sum within that stated floating-point tolerance. It does not mean byte identity across every possible algorithm and compiler.

the ring stopped sending the whole array

The next implementation changes the unit of movement. Split an (M)-byte array into (P) equal chunks. With four ranks and four elements, each chunk is one element. Arrange ranks in a circle. Each step sends one chunk clockwise and receives one chunk from the counterclockwise neighbor.

The first phase is reduce-scatter. The name describes the state at its end: the array has been reduced, but the result is scattered so that each rank owns only one complete chunk.

The exact trace begins:

initial
r0 [1 1 1 1]   r1 [2 2 2 2]   r2 [3 3 3 3]   r3 [4 4 4 4]

reduce-scatter step 0
r0 [1 1 1 5]   r1 [3 2 2 2]   r2 [3 5 3 3]   r3 [4 4 7 4]

reduce-scatter step 1
r0 [1 1 8 5]   r1 [3 2 2 7]   r2 [6 5 3 3]   r3 [4 9 7 4]

reduce-scatter step 2
r0 [1 10 8 5]  r1 [3 2 10 7]  r2 [6 5 3 10]  r3 [10 9 7 4]

The unreduced entries in each row are no longer meaningful. After three steps, rank 0 owns reduced chunk 1, rank 1 owns chunk 2, rank 2 owns chunk 3, and rank 3 owns chunk 0. Every owned chunk is 10.

The second phase is all-gather. No arithmetic remains. Each rank circulates the complete chunks until everyone has all four:

all-gather step 0
r0 [10 10 8 5]  r1 [3 10 10 7]  r2 [6 5 10 10]  r3 [10 9 7 10]

all-gather step 1
r0 [10 10 8 10] r1 [10 10 10 7] r2 [6 10 10 10] r3 [10 9 10 10]

all-gather step 2
r0 [10 10 10 10] r1 [10 10 10 10]
r2 [10 10 10 10] r3 [10 10 10 10]

There are (P-1) reduce-scatter steps and (P-1) all-gather steps. Every message contains (M/P) bytes. Per rank:

$$\text{ring bytes} = 2(P-1)\frac{M}{P} = 2\frac{P-1}{P}M$$

As (P) grows, the coefficient approaches two. Eight ranks reducing a one-mebibyte array each send (1.75) MiB of payload across both phases, not the naive root’s 14 MiB and not recursive doubling’s 3 MiB per rank.

The price appears in the startup term:

$$T_{\text{ring}} \approx 2(P-1)\alpha + 2\frac{P-1}{P}\beta M$$

For a scalar, dividing into (P) chunks is impossible without padding and pointless even if padding is allowed. For a large array, the smaller byte coefficient can outweigh the additional rounds.

The implementation requires count % size == 0. A production path can distribute a remainder, pad, or select a fallback. The probe rejects the shape instead of hiding that condition.

the second butterfly kept only half

Recursive halving and doubling combines the ring’s chunked byte count with logarithmic rounds, under stricter shape assumptions.

During reduce-scatter, every rank starts with the full array. At mask 1, ranks exchange half. A rank whose low bit is zero keeps the lower half and sends the upper half. A rank whose low bit is one keeps the upper half and sends the lower. The received half is reduced into the half being kept.

For four ranks:

after xor 1
r0 owns lower half [3 3]
r1 owns upper half [3 3]
r2 owns lower half [7 7]
r3 owns upper half [7 7]

At mask 2, each rank exchanges half of what it still owns with the rank that differs in the next bit. Each participant finishes with one reduced quarter:

r0 owns chunk 0 [10]
r2 owns chunk 1 [10]
r1 owns chunk 2 [10]
r3 owns chunk 3 [10]

The unusual rank-to-chunk order is a bit-reversal consequence of consuming rank bits from low to high. It does not matter as long as the all-gather reverses the same schedule.

The exchanged sizes are:

$$\frac{M}{2},\frac{M}{4},\ldots,\frac{M}{P}$$

Their sum is a finite geometric series:

$$\frac{M}{2}+\frac{M}{4}+\cdots+\frac{M}{P} = \frac{P-1}{P}M$$

The doubling all-gather exchanges the same sizes in reverse. Total payload per rank therefore matches the ring:

$$2\frac{P-1}{P}M$$

The round count is (2\log_2P):

$$T_{\text{halving/doubling}} \approx 2\log_2(P)\alpha + 2\frac{P-1}{P}\beta M$$

In this ideal model, recursive halving and doubling dominates the ring for a power-of-two group because both have the same payload term and the butterfly has fewer rounds. That does not make rings obsolete. A ring maps naturally to neighbor links, pipelines uniform chunks, tolerates arbitrary rank counts more directly, and can avoid the simultaneous long-distance exchanges a butterfly requests. On a topology where several butterfly edges share one physical link, the ideal model’s independent edges do not exist.

The implementation requires a power-of-two rank count and an array divisible by that count. It allocates a full output plus a temporary receive buffer. It is a useful executable mechanism, not a general replacement for a tuned collective library.

zero bytes still took time

The cost model needs local values for (\alpha) and (\beta). The probe launches two Open MPI ranks on the same Apple M4 MacBook Air. It forces the ob1 point-to-point layer and the self,sm byte transport components. sm is Open MPI’s same-host shared-memory transport. No packet crossed an Ethernet or InfiniBand cable.

Rank zero sends a buffer to rank one. Rank one sends it back. Dividing the round-trip time by two estimates one direction. The probe batches one hundred exchanges per sample for messages up to 4,096 bytes, twenty for 32 KiB and 256 KiB, and five for 1 MiB. It records 401, 201, or 81 samples depending on size. Twenty untimed batches warm each size.

The final verifier run produced:

bytes      median one-way   MAD       p10       p90
0             0.065 us     0.000     0.065     0.070
8             0.070 us     0.000     0.070     0.080
64            0.080 us     0.005     0.080     0.090
512           0.155 us     0.005     0.150     0.170
4,096         0.505 us     0.005     0.500     0.530
32,768        1.150 us     0.100     1.050     1.300
262,144       5.100 us     0.150     4.825     5.325
1,048,576    18.600 us     0.600    17.600    19.300

The zero-byte result is an operational estimate of startup in this exact process and transport configuration. It includes calls, matching, shared-memory coordination, and timer behavior. It is not the latency of a NIC.

A linear fit over the three largest sizes gives:

$$T_{\text{one way}} = 0.591740\ \mu s + (0.000017175\ \mu s/\text{byte})M$$

The slope corresponds to:

$$\frac{0.001}{0.000017175} = 58.223\ \text{GB/s}$$

The factor 0.001 converts bytes per microsecond into decimal gigabytes per second. This number describes one same-host MPI payload path over the fitted sizes. It is not DRAM bandwidth, aggregate bidirectional bandwidth, or network line rate.

The fit intercept differs from the zero-byte median. That is already a warning. A single straight line does not describe all sizes. The explicit model calculations use the zero-byte median for (\alpha) and the large-message slope for (\beta), then test whether that mixed approximation predicts anything useful.

the first prediction put the crossover near eight kilobytes

For eight ranks, (\log_2P=3). Equating the tree and ring expressions gives:

$$6(\alpha+\beta M) =14\alpha+\frac{14}{8}\beta M$$

Collecting terms:

$$\left(6-\frac{14}{8}\right)\beta M =(14-6)\alpha$$

With the measured (\alpha) and fitted (\beta):

$$M \approx 7{,}124\ \text{bytes}$$

The same calculation for recursive doubling and recursive halving/doubling predicts a crossover near 9,083 bytes. Below that point, sending the whole payload for only three rounds should be cheaper. Above it, the chunked byte term should win despite the second butterfly’s six rounds.

The predicted eight-rank times make the assumptions visible:

payload    naive    tree    doubling    ring    halving/doubling
64 B       0.925    0.397     0.198      0.912       0.392 us
8 KiB      2.880    1.234     0.617      1.156       0.636 us
128 KiB   32.427   13.897     6.949      4.850       4.330 us
1 MiB    253.044  108.448    54.224     32.427      31.907 us

These are predictions written before interpreting the all-reduce measurements. The ideal model expects recursive doubling to win at 64 bytes and 8 KiB. It expects recursive halving/doubling to edge out the ring at 1 MiB. It expects the naive root to be worst at large size. A measurement that disagrees tells us which omitted cost matters.

six algorithms ran in randomized order

The benchmark compares:

1. the naive root gather and broadcast;
2. the binomial reduction and broadcast tree;
3. recursive doubling over complete arrays;
4. ring reduce-scatter and all-gather;
5. recursive halving and doubling;
6. Open MPI’s MPI_Allreduce.

Each rank’s element (i) starts as:

$$x_{r,i}=(r+1)+10^{-6}i$$

The expected sum for (P) ranks is:

$$\sum_{r=0}^{P-1}x_{r,i} =\frac{P(P+1)}{2}+P(10^{-6}i)$$

Every element is checked after every timed call. The check occurs outside the timed region. A failed check aborts all ranks rather than letting a fast wrong algorithm enter the table.

Before each call, all ranks enter MPI_Barrier. The timer starts after the barrier returns. Every rank measures its local elapsed time. A reduction takes the maximum elapsed time across ranks, because the collective is not operationally finished for the application while one participant remains inside it. Using rank zero’s time alone could hide a straggler elsewhere.

There are five warmups and 31 recorded trials. Rank zero shuffles the six algorithms with a fixed seed before every repetition and broadcasts that order. The benchmark reports median, median absolute deviation, p10, p90, minimum, and maximum. The displayed table keeps median and MAD compact, but the probe retains the remaining statistics.

Eight ranks run on one ten-core machine. They are processes, not eight physical nodes. The forced shared-memory transport is useful for exercising actual MPI matching and movement while keeping the environment available. It cannot validate a fabric topology claim.

sixty-four bytes paid almost entirely for rounds

For eight doubles, (M=64) bytes. The final verifier run gave:

algorithm               median       MAD       p10       p90
naive                      2 us        0 us       1 us       2 us
tree                       1 us        0 us       1 us       2 us
recursive doubling         1 us        0 us       1 us       2 us
ring                       4 us        0 us       4 us       5 us
halving/doubling           2 us        0 us       2 us       2 us
MPI_Allreduce              1 us        0 us       1 us       2 us

The timer and implementation quantize these small results to microseconds, so a one-microsecond difference should not be read as a stable fractional speedup. The ordering is still coherent with the startup term. Recursive doubling uses three exchanges. The ring uses fourteen. Halving/doubling uses six. Open MPI matches the fastest custom median.

The naive median is only two microseconds, despite fourteen root messages. Its maximum was 195 microseconds. The median says the common same-host case is small. The maximum says the schedule occasionally found a much slower path. Reporting only the best sample would have erased that event. Reporting only the maximum would have described an unusual event as ordinary.

At 8 KiB, the table became:

algorithm               median       MAD       p10       p90
naive                     19 us        1 us      18 us      20 us
tree                      12 us        1 us      11 us      12 us
recursive doubling         9 us        0 us       9 us      10 us
ring                      13 us        0 us      13 us      14 us
halving/doubling           9 us        0 us       8 us       9 us
MPI_Allreduce              9 us        0 us       9 us      11 us

The model put the tree and ring crossover at about 7,124 bytes, near this payload. Their measured medians are 12 and 13 microseconds. That closeness is useful, but it should not be mistaken for model validation. The model predicted 1.234 and 1.156 microseconds, almost ten times smaller than the measurements. It found a plausible crossing while missing most of the elapsed time.

Recursive halving/doubling reached the same nine-microsecond median as recursive doubling even though the model put their crossover slightly above this size. Open MPI also reached nine. At this resolution, the claim supported by the run is a tie in medians, not an ordered three-way win.

one mebibyte made the ring win

At 131,072 doubles, each rank contributes 1,048,576 bytes:

algorithm               median       MAD       p10       p90
naive                    832 us      127 us     721 us   1,054 us
tree                     603 us       47 us     544 us     692 us
recursive doubling     1,003 us       10 us     978 us   1,195 us
ring                      433 us       12 us     421 us     476 us
halving/doubling          479 us       16 us     456 us     517 us
MPI_Allreduce              510 us       29 us     467 us     712 us

The byte-saving algorithms moved to the front, as predicted. The ring beat recursive doubling by 570 microseconds at the median. Sending the full mebibyte three times per rank was more expensive than taking fourteen chunk rounds.

The ring also beat recursive halving/doubling by 46 microseconds, contradicting the ideal model. Both algorithms send the same modeled payload per rank. The butterfly uses fewer rounds. Under the model, it should not lose.

The missing variable is the path taken by simultaneous exchanges. Every ring rank talks to the same two neighbors at every step, with uniform 128 KiB chunks. Recursive halving starts with 512 KiB partner exchanges, then 256 KiB, then 128 KiB, and reverses those sizes. On this Open MPI shared-memory path, those concurrent copy and reduction patterns do not cost the same merely because their payload sums match. Cache placement, shared-memory queues, copy concurrency, reduction locality, and process scheduling all sit outside (\alpha+\beta M).

The ideal ring prediction was 32.427 microseconds. The measured median was 433. The two-rank ping-pong slope did not include eight simultaneous ranks, arithmetic over received chunks, barriers, per-step matching, or the benchmark’s maximum-rank completion. A thirteen-fold absolute miss is not a rounding error. The model remains useful as a ledger and a hypothesis generator. It is not a substitute for the collective run.

The naive result moved more than the ring. Its MAD was 127 microseconds, versus 12 for the ring. Maxima did not preserve median ordering either: recursive doubling reached 2,779 microseconds, while the ring reached 1,504 and the tuned library call reached 1,898. The run does not identify the cause of those tails. It does show that choosing from one median without preserving the distribution would hide a deployment risk.

the library did not have one all-reduce

MPI_Allreduce names a semantic operation, not an algorithm. Open MPI 5’s tuned collective component exposes basic_linear, nonoverlapping, recursive_doubling, ring, segmented_ring, rabenseifner, and allgather_reduce choices. Algorithm zero tells it to use fixed rules. The official Open MPI collective tuning documentation records those names and explains fixed, forced, and dynamic decision modes.

The default fixed mode uses built-in decisions based on communicator and message size. A forced mode can short-circuit that choice for an experiment. A dynamic rule file can map communicator and message-size ranges to algorithms. The documentation explicitly warns that rules derived on one cluster can be poor on different hardware.

The custom benchmark leaves Open MPI in its default mode. Therefore the MPI_Allreduce row is “whatever the selected components and fixed rules chose for this call,” not a seventh known source schedule. Proving its selected algorithm would require component selection logs or a forced-algorithm sweep. The table does not infer the algorithm from similar timing.

The Rostam archive discussed later did force native MPI algorithms 0, 3, and 6. In Open MPI’s numbering, 3 is recursive doubling and 6 is Rabenseifner. Algorithm 0 retains fixed selection. Keeping IDs with the archive matters because “MPI” by itself hides which mechanism was measured.

the communicator needed a generation because calls have no tag

Point-to-point MPI messages in the probe use tags such as 100, 200, and 400 to keep phases separate. Collective APIs do not expose a matching tag. The MPI 4.1 nonblocking collective rules require all processes to call collectives in the same order on a communicator. A blocking collective does not match a nonblocking one. Starting a different sequence on two ranks is erroneous, not a request for the library to guess.

HPX’s reusable communicator has the same coordination problem expressed through generations. Generation (k) identifies one use of the gate. Every site in that operation has to name a compatible generation and operation type. The current server records current_operation_ and rejects a different collective arriving while the generation is active. It also caches the first step or finalizer exception so that all participants observe the same failure instead of some ranks receiving values after another rank saw an exception.

Hierarchical all-reduce consumes two internal phases:

$$g_{\text{reduce}}=2k-1$$ $$g_{\text{broadcast}}=2k$$

All hierarchical collectives advance their member communicators by two positions per user call, including paths that logically touch one communicator once. That invariant permits one hierarchical communicator instance to be shared across collective types without its internal gates drifting apart.

The source check is exact, not a claim about an older release. It finds hierarchical_phase_generations(generation), the flat arity >= num_sites dispatch, operation_error_, and the and_gate in checkout acf2394d.... Existing binaries then run cross-collective sharing, flat fallback, all-reduce, all-to-all, inclusive-scan, and exclusive-scan tests at built revision e6324bd801.

arity traded depth for another small crowd

The hierarchy has a second threshold hidden in its arity. A binary tree over (P) leaves has depth close to:

$$\left\lceil \log_2 P \right\rceil$$

An arity-(a) tree has depth close to:

$$\left\lceil \log_a P \right\rceil$$

For 800 sites, those ideal depths are 10 for arity 2, 5 for arity 4, 4 for arity 8, and 3 for arity 16. Fewer levels reduce sequential dependencies and generation work along one path.

The other side is inside each level’s flat communicator. A representative of an arity-16 group can receive contributions from up to fifteen peers at that level. A binary representative receives from at most one peer in its two-site communicator. Raising arity removes levels by recreating a smaller version of the root crowd.

A simplistic extension of the tree model might charge:

$$2\left\lceil\log_a P\right\rceil(\alpha+\beta M)$$

That assumes the (a-1) child arrivals inside one level cost the same as one message time. The HPX server still stores every group member’s value and runs one finalizer. If arrivals serialize on its lock, action processing, memory path, or link, the per-level term grows with group size. If they overlap efficiently, the depth reduction can dominate.

The local HPX correctness binary prints timings for several arities, but it runs logical sites as tasks inside one process. In the final run at 64 logical sites, medians were not collected and the one reported mean per arity ranged from roughly 0.11 to 0.27 milliseconds. Those values moved substantially between verifier runs. They are enough to reject a claim that one arity always wins even in the local test, but not enough to select a cluster arity.

An arity experiment needs randomized repeated runs for each (P), (M), transport, and placement. It also needs to record the actual group tree so a surprising result can be tied to which representatives and physical links carried the work. The current default arity of two is an implementation default, not a derived optimum.

uneven groups were a correctness problem before a performance problem

A binary tree over eight ranks is easy to draw. Production groups are not obliged to be powers of two. With ten sites and arity four, the top level cannot make four equal groups of size 2.5.

The HPX builder computes a bounded top-level group count and distributes sites so group sizes differ by at most one. Representatives are the leftmost sites in their groups. It recurses only into the group containing the current site, so each site stores the communicators along its own path rather than a materialized global tree.

Correctness tests exercise site counts:

arity 2: 3, 5, 6, 7, 9, 10, 11, 15
arity 4: 5, 6, 7, 9, 10, 11, 13, 15

The local HPX all-reduce test launches one asynchronous task per logical site in a single runtime process. Each task creates its hierarchical communicator, contributes site + iteration, and checks the sum for 500 iterations in a release build. These are real HPX gates, futures, components, and reduction code. They are not remote localities and must not be called network timings.

The same test sweeps 2, 4, 8, 16, 32, and 64 logical sites with arities 2, 4, 8, and 16 where applicable. The output varies from run to run and is dominated by local scheduling. Its value here is that every shape completes and produces the expected sum. The benchmark table above, not this unit test, is the fresh performance evidence.

The flat fallback tests compare a threshold that forces one communicator with a threshold of zero that forces the tree. Cross-collective tests reuse the same hierarchy across operation types. All passed in the final verifier. That result closes a correctness path. It does not establish that the default threshold of sixteen minimizes time on any cluster.

all-reduce was only one arrangement of the values

The earlier list of collective names becomes easier to reason about if the values remain visible. A separate four-rank MPI probe executes the core operations rather than defining them only in prose. Each rank begins with one integer:

rank:   0  1  2  3
value:  1  2  3  4

A broadcast has one source and no reduction. With rank 2 as root, its value 3 appears at all four ranks:

broadcast(root=2): 3 3 3 3

A reduce combines every value but returns the answer only to the root:

reduce(sum, root=0): 10 ? ? ?

The question marks do not mean zero. The receive buffers on nonroot ranks are not results of the operation. Reading them as if they were defined is an application bug.

All-reduce changes the destination set:

all_reduce(sum): 10 10 10 10

That can be built as reduce plus broadcast, a recursive-doubling exchange, a ring reduce-scatter plus all-gather, recursive halving plus doubling, or another schedule. The semantic line does not choose one.

Gather does no arithmetic. Rank (r) contributes one value and the root receives the rank-ordered array:

gather(root=0): [1 2 3 4] ? ? ?

All-gather gives that complete array to every rank:

all_gather: [1 2 3 4] at every rank

The byte difference from reduce matters. Reducing four (M)-byte arrays can return one (M)-byte array. Gathering them returns (4M) bytes at the root. All-gather returns (4M) bytes at every rank. There is no combining operator that compresses contributions into one array.

Scatter reverses the ownership direction. A root array [10,20,30,40] is divided so rank (r) receives element (r):

scatter(root=0): rank0=10 rank1=20 rank2=30 rank3=40

Reduce-scatter combines both ideas. The probe gives rank (r) this vector:

$$[10(r+1)+0,\ 10(r+1)+1,\ 10(r+1)+2,\ 10(r+1)+3]$$

Summing each column produces:

[100 104 108 112]

MPI_Reduce_scatter_block leaves one element on each rank:

rank0=100 rank1=104 rank2=108 rank3=112

This is the ring’s first phase exposed as a public operation. A training system can reduce gradient shards and keep them sharded instead of immediately paying for all-gather. Whether it later needs the complete tensor depends on how parameters and computation are partitioned.

all-to-all transposed ownership

All-to-all is easier to misread because every rank is both a scatter root and a gather destination. Give rank (r) a row whose destination (d) contains (10r+d):

rank 0 sends [ 0  1  2  3]
rank 1 sends [10 11 12 13]
rank 2 sends [20 21 22 23]
rank 3 sends [30 31 32 33]

Destination rank 0 receives the first element of every row:

rank 0 receives [0 10 20 30]

The full result is:

rank 0 [0 10 20 30]
rank 1 [1 11 21 31]
rank 2 [2 12 22 32]
rank 3 [3 13 23 33]

It is a distributed matrix transpose. If each source-to-destination piece is (m) bytes, each rank sends approximately ((P-1)m) bytes to peers and receives the same. Total peer payload across the group grows as (P(P-1)m). When an expert-parallel model routes tokens to experts on other ranks, this ownership transpose is the expensive core hidden behind the framework call.

A flat HPX all-to-all stores every source row at one communicator server before constructing destination columns. The hierarchical version gathers within subtrees, exchanges between representatives, and scatters downward. Flattening its carrier representation was a separate source change because nested vectors add allocation, serialization, and indexing work even when the mathematical transpose is unchanged.

The local MPI catalog verifies the four-by-four transpose exactly. It does not benchmark all-to-all. Contention, pairwise ordering, Bruck-style schedules, topology, and unequal message sizes need their own investigation. Treating the all-reduce table as evidence for all-to-all would be the same mistake as treating a function name as an algorithm.

a scan kept every prefix

An inclusive scan gives rank (r) the reduction of ranks zero through (r):

input:           1  2  3  4
inclusive sum:   1  3  6 10

An exclusive scan excludes the calling rank:

exclusive sum:  ?  1  3  6

MPI leaves rank zero’s exclusive result undefined. A separate initial value can make a library interface return a useful identity there, but that is an API decision on top of the dependency.

The data dependency differs from all-reduce. Rank 1 needs contributions from ranks 0 and 1, not ranks 2 and 3. Rank 3 needs all four. A finalizer that first computes all prefixes centrally is correct, but a parallel scan algorithm can use an upsweep and downsweep or recursive doubling to distribute work.

This is why grouping collectives by English similarity is weak. reduce, all_reduce, reduce_scatter, and scan all apply the same binary operator. Their output ownership creates different communication graphs.

a barrier moved no user bytes and still had a cost

The catalog finishes with MPI_Barrier. No user payload moves, yet rank zero cannot return while another rank has not participated enough for the implementation to complete the operation. Control messages, counters, or tree notifications still move.

A barrier is not a universal flush. It does not make an fsync happen, drain an unrelated GPU stream, publish an object-store write, or repair a data race in memory outside the programming model. It orders participation in that communicator operation.

HPX’s communicator-backed barrier calls the same gate machinery with void data. Its distributed barrier class is reusable and has its own wrapper and lifetime. The distinction mirrors the article’s opening correction: similar semantics do not guarantee the same object path.

Adding barriers around a benchmark can both help and hurt. A barrier before timing aligns starts and prevents one rank from entering the measured operation much earlier during a controlled microbenchmark. It also creates a synchronized burst that may not resemble production arrivals. A barrier after every operation prevents natural overlap and can turn a throughput workload into a sequence of latency tests.

The all-reduce probe uses a barrier before each algorithm deliberately. It wants one controlled call and then takes the maximum rank time. The arrival-skew probe later removes that alignment on one rank to measure the consequence.

the old cluster run answered a different question

Fresh same-host MPI runs are not the only evidence in the workspace. A timestamped archive records a separate run on the Rostam cluster. That archive existed before this expansion; it was not rerun for the article.

Its provenance says:

UTC start:             2026-07-15 21:30:28
host:                  medusa11.rostam.cct.lsu.edu
Slurm job:             171502
HPX source:            51fda78e543f2c5ff9738dd1f76685229dba6f2d
nodes:                 1, 2, 4
HPX localities/node:   20
HPX threads/locality:  2
payload sizes:         1, 1,024, 16,384 int values
warmups:               5
measured iterations:   30
compiler:              GCC 14.2.0
Open MPI:              5.0.5

The archive contains raw HPX rows, raw native-MPI rows, per-run outputs, configuration and build logs, a summary, and SHA-256 records. The committed probe checks the full tar archive against hash 3f81eba0... when the neighboring HPX repository is available. It also checks a compact 45-row extract stored with this article’s private evidence.

The provenance is incomplete for a portable hardware claim. It does not name the CPU model, memory capacity, NIC, link rate, kernel, process affinity, NUMA placement, or switch topology. Those omissions prevent using the archive to derive a fabric (\alpha), (\beta), or cross-cluster expectation. They do not erase the recorded comparison on that job.

The archive’s HPX rows in the compact extract use the tuned LCI parcelport. They compare a reused flat communicator, labeled multi_use, with direct channel algorithms. They are not measurements of the hierarchical communicator above. That difference is central. The archive can test whether bypassing the flat communicator shape helped those operations. It cannot reveal the flat-versus-hierarchical threshold.

the direct path was slower until one large case

The selected all-reduce medians are:

nodes  ints/rank   HPX reused flat   HPX direct channel   best native MPI
1           1          0.219 ms            0.337 ms            0.012 ms
1       1,024          0.247 ms            0.380 ms            0.076 ms
1      16,384          0.772 ms            0.894 ms            0.192 ms

2           1          0.357 ms            0.564 ms            0.017 ms
2       1,024          0.454 ms            0.678 ms            0.105 ms
2      16,384          1.491 ms            1.624 ms            0.375 ms

4           1          0.601 ms            0.928 ms            0.019 ms
4       1,024          0.820 ms            1.390 ms            0.124 ms
4      16,384          4.393 ms            3.896 ms            0.381 ms

The configure log records x86-64 and GCC, but the provenance does not print sizeof(int). Interpreting its int as the four-byte type used by that ABI makes 16,384 integers 64 KiB per rank. Four nodes mean 80 HPX localities under this profile.

The direct channel path removes the central reusable communicator from the value path, but it loses in eight of the nine displayed cases. At four nodes and 64 KiB, it finally wins against reused flat HPX by:

$$\frac{4.393-3.896}{4.393}=0.113$$

about 11.3 percent.

This is the result that the source alone could not predict. Centralization is a plausible scaling cost. Removing it also adds a schedule, more direct peer actions, and different synchronization. At small payloads, those costs dominate in this archive. At the largest four-node case, the balance reverses.

The direct path remains 10.2 times slower than the best measured native MPI median in that row:

$$\frac{3.896}{0.3811}=10.22$$

That ratio is not a statement that HPX must always be ten times slower. It belongs to one operation, payload, job, transport profile, code revision, and selected native MPI sweep. It does show that improving HPX from 4.393 to 3.896 milliseconds did not make the implementation competitive with the strongest baseline in the archive.

the fastest MPI id changed with size and node count

The cluster run measured native Open MPI algorithms 0, 3, and 6:

• 0 asks Open MPI’s fixed rules to select;
• 3 is recursive doubling;
• 6 is Rabenseifner, a reduce-scatter plus all-gather family related to the halving/doubling probe.

The best median among those three changed:

nodes  ints/rank   best id
1           1          3
1       1,024          3
1      16,384          0
2           1          0
2       1,024          0
2      16,384          0
4           1          3
4       1,024          6
4      16,384          6

At one node, recursive doubling won the two smaller payloads. At four nodes, Rabenseifner won the two larger ones. The observed selection aligns with the message-and-byte argument without proving a universal threshold.

The differences among native algorithms were sometimes small. At four nodes and 1,024 integers, medians were 0.124526 ms for algorithm 0, 0.125119 ms for algorithm 3, and 0.124306 ms for algorithm 6. Calling 6 a meaningful winner from those medians alone would require repeated randomized comparisons or an interval on paired differences. The archive contains 30 iterations per configuration but only one transport repetition, and the compact report does not preserve a cross-configuration paired order.

At four nodes and 16,384 integers, the medians separated more clearly: 0.388748, 0.391066, and 0.381100 ms. Even there, the right claim is that algorithm 6 had the lowest recorded median in this run. A deployment rule should be validated against fresh traffic and more repetitions.

The archive also records that long Open MPI parcelport runs using MPI_THREAD_MULTIPLE were unstable on this cluster. That profile was bounded to one node with two warmups and five measured iterations. Hiding the excluded multinode parcelport rows would make the transport comparison look cleaner and less true. The LCI and native MPI rows above are the evidence that survived the stated profile.

one late rank charged everybody else

All algorithm tables began with a barrier, so their ranks entered each call together. Production ranks often arrive at different times. One finishes a longer batch, takes a page fault, waits on a GPU event, loses a CPU time slice, or reaches a different branch.

The arrival probe runs four ranks and delays rank 3 by 20 milliseconds after a common barrier but before its MPI_Allreduce. It records two times per rank:

• total time from the common release through collective completion;
• time spent after that rank enters the collective.

The control had no deliberate delay:

control total milliseconds
r0 0.012   r1 0.012   r2 0.010   r3 0.011

control inside collective
r0 0.011   r1 0.012   r2 0.010   r3 0.011

With rank 3 delayed:

skew total milliseconds
r0 24.805  r1 24.806  r2 24.803  r3 24.804

skew inside collective
r0 24.805  r1 24.805  r2 24.803  r3 0.003

All four completed at nearly the same wall-clock point. Rank 3’s collective call looked extraordinarily fast because it arrived after the other ranks had waited and the data path was ready to finish. Ranks 0 through 2 charged the missing arrival to their collective duration.

If a profiler records only rank 3, it can report a three-microsecond all-reduce during a nearly 25-millisecond global stall. If it records the maximum rank duration, it exposes the stall but cannot say whether the communication algorithm or arrival skew caused it. A useful trace needs both arrival and completion.

The extra 4.8 milliseconds beyond the requested sleep belongs to scheduler timing, timer resolution, and process wakeup in this run. The probe does not reinterpret a nominal 20-millisecond nanosleep as exact.

This changes algorithm comparison. A tree representative that arrives late can block a subtree. A ring step cannot advance past a missing neighbor contribution. A butterfly partner missing at one mask delays the corresponding half before later masks can form the complete result. Different graphs spread the wait differently, but none can correctly sum a value that has not arrived.

the generation prevented confusion but not a straggler

HPX generations stop one invocation from being mistaken for another. They cannot make a late site early. The and_gate waits for the required number of arrivals in its current generation. A site that supplies generation (k+1) while peers are still completing (k) needs the library’s sequencing rules to avoid corrupting shared data. It does not let generation (k) finish without that site’s value.

This distinction separates correctness metadata from progress. A generation answers “which collective is this?” A tag answers a similar question for point-to-point matching. Neither guarantees that the matching peer will execute.

Long-running systems also have to decide how generations interact with cancellation and shutdown. The current hierarchical all-reduce returns an HPX future, but its root path performs reduce_here(...).get() before starting broadcast. Once participants have entered the reduction, abandoning one caller does not automatically cancel the shared operation at every site. The article has not executed a distributed cancellation protocol for HPX collectives.

The server’s cached operation_error_ improves a different failure. If a step or finalizer throws after state has been moved or partially combined, rerunning it independently for later callbacks could give ranks different outcomes. Caching the first exception makes the operation fail consistently for the participants whose callbacks run. It does not recover a departed machine or synthesize its missing contribution.

nonblocking was an opportunity, not overlap

The blocking benchmark leaves no application computation between collective start and completion. MPI also provides MPI_Iallreduce. According to the MPI 4.1 nonblocking collective specification, initiation is local and returns a request. The send buffer must not be modified and the receive buffer must not be accessed until completion. The operation may progress independently. “May” is the important word.

The local overlap probe uses four ranks, a 2 MiB all-reduce, a 0.5 millisecond CPU work window, five warmups, and 31 randomized trials. It compares:

1. blocking all-reduce, then 0.5 milliseconds of work;
2. MPI_Iallreduce, work without MPI calls, then MPI_Wait;
3. MPI_Iallreduce, work split by repeated MPI_Test, then MPI_Wait if needed.

The final medians were:

blocking then compute             0.924 ms
Iallreduce, compute, no polling    0.810 ms
Iallreduce, compute, with polling  0.500 ms

Initiation without polling saved 0.114 milliseconds relative to the blocking sequence, so this implementation did some useful work before the final wait or changed the schedule enough to reduce total time. It did not hide the whole collective.

Polling reached the 0.5-millisecond work window. That result is not free overlap. Each MPI_Test call consumes CPU time and gives the MPI implementation a progress opportunity. The probe performs time-bounded synthetic arithmetic, not a fixed quantity of model computation, so polling overhead displaces some arithmetic inside the same deadline. A stronger production experiment would fix the computation, verify its output, measure CPU consumption, inspect progress threads, and repeat with the application’s actual collective sizes.

The negative lesson is still direct: replacing MPI_Allreduce with MPI_Iallreduce and immediately doing computation does not prove communication overlapped. A trace needs initiation, progress calls or progress threads, completion, buffer dependencies, and useful compute on the same timeline.

HPX futures offer composition points, but the current hierarchical source blocks between its upward reduction and downward broadcast at the root. Removing that .get() requires preserving the value lifetime, generation order, error propagation, and the fact that broadcast cannot begin until the reduced value exists. An asynchronous return type does not remove the dependency graph.

chunk size added another axis

The ring probe sends exactly one (M/P) chunk per step. A larger implementation can divide those chunks again and pipeline segments. The first segment can move to the next neighbor while a later segment is still being reduced behind it.

Segmentation trades byte concurrency against startup:

$$T_{\text{segment}} \approx \alpha+\beta S$$

where (S) is segment bytes. Halving (S) doubles the number of segment starts for the same payload. Very large segments can leave links idle between dependency steps or reduce opportunities to overlap arithmetic. Very small segments pay (\alpha), queue operations, headers, and progress work too often.

Open MPI exposes both ring and segmented_ring as distinct all-reduce algorithms. The difference exists because “use a ring” does not finish the design. Segment size, number of in-flight requests, copy placement, and reduction granularity can move the result.

The local custom benchmark intentionally avoids autotuning. It tests one transparent schedule. A production selection system would need a search space containing at least:

• algorithm family;
• rank count and non-power-of-two handling;
• payload and datatype;
• segment or chunk size;
• reduction operator cost;
• in-place or out-of-place buffers;
• transport and protocol;
• topology embedding;
• progress strategy;
• arrival-skew distribution;
• concurrent compute and other collectives.

Selecting the fastest candidate on the same 31 samples used to compare it would overfit measurement noise. Candidate selection needs one sample set and validation needs fresh randomized runs, especially when several medians differ by less than one timer tick.

one logical edge could cross several physical links

The cost equations treated every rank pair as one equal link. Hardware does not.

Eight ranks might mean:

• eight processes sharing one CPU and memory controller;
• eight GPUs behind one PCIe switch;
• four GPUs on each of two sockets;
• eight nodes in one rack;
• one rank per node across two network rails;
• a mixture of NVLink, PCIe, and InfiniBand paths.

A ring can be embedded so neighbor edges follow fast links, but a poor rank order can force the ring across a slow intersocket or internode boundary many times. A butterfly’s rank xor mask partners change each round. The high masks can cross groups of switches simultaneously and contend on an uplink. A hierarchical tree can reduce within a node before sending one value upward, but only if its groups match physical locality.

The missing quantity is load on a physical cut. Draw a line between two groups of four GPUs. If a ring crosses that line on two directed edges, each reduce-scatter step asks those edges to carry two chunks. A recursive-doubling round can place all four pairs across the same line at once. It still has one logical message per rank, but four messages now divide the cut’s available bytes per second. Replacing the per-link (\beta) in the ideal equation with the fastest link’s value would underpredict that round. The relevant byte term is bounded by aggregate bytes crossing the cut divided by the cut bandwidth. This does not make a ring universally better. A ring uses the cut in more rounds, while a tree or butterfly can use it in fewer, heavier rounds. It does explain why topology cannot be reduced to a longer point-to-point latency.

The HPX hierarchy partitions contiguous site numbers. That is topology aware only when site numbering itself reflects topology. If ranks 0 through 19 share node A, 20 through 39 share node B, and the grouping preserves those ranges, early reduction can remain local. If placement interleaves site numbers across nodes, a mathematically balanced tree can be physically poor.

Multi-rail communication adds a second mapping. Two NICs do not double throughput if both logical streams reach the same rail, one GPU has a slow path to the chosen NIC, or the switch routes both flows through one bottleneck. Rail alignment is the assignment problem between ranks, GPUs, NICs, and remote peers. The message ledger needs a physical path column before it can predict multi-rail gain.

No fresh probe here uses a NIC, RDMA, CUDA, NVLink, or multiple rails. The Rostam archive confirms multiple nodes but lacks the committed hardware topology needed for that analysis. Those mechanisms remain outside the measured boundary.

the GPU library exposed the same choice without the same implementation

NCCL offers all-reduce, reduce, broadcast, all-gather, reduce-scatter, all-to-all, gather, and scatter operations over GPU buffers. Its current algorithm-selection documentation lists Ring, Tree, CollNet variants, NVLS variants, and PAT among the selectable families. The default uses available algorithms according to topology and architecture.

That list does not make the CPU MPI timings transferable to NCCL. A GPU collective includes device kernels, protocol choices such as LL, LL128, and Simple, GPU memory transactions, copy engines or SM participation, PCIe or NVLink paths, GPUDirect RDMA, and stream ordering. The startup term can include kernel launch and device synchronization. The byte term can be limited by a different memory or interconnect path.

The old environment variable NCCL_TREE_THRESHOLD cannot be cited as a current selection rule. Current documentation marks it removed in NCCL 2.5. NCCL_ALGO can constrain candidates, but NVIDIA discourages casual protocol forcing and allows defaults to change with versions and hardware.

The useful transfer from this investigation is the method:

1. make the exact values and ownership visible;
2. count critical rounds and bytes;
3. map logical edges to physical paths;
4. predict the size-dependent winner;
5. force candidates only for diagnosis;
6. measure distributions with real arrival patterns;
7. return to automatic selection only after validating it.

The numeric thresholds do not transfer.

a missing rank did not produce a partial sum

Every successful probe had all ranks alive and calling operations in the same order. Failure changes the contract.

If one MPI process exits during the default Open MPI all-reduce, the normal result is job-level failure handling, not a seven-rank sum returned under an eight-rank communicator. Open MPI’s MPI_Allreduce documentation says the default communicator error handlers abort and that MPI does not guarantee an application can continue past an error merely because an error-return handler is installed.

A retry is not trivial. Suppose rank 0 does not know whether rank 7’s value was included before a failure. Reissuing the operation on surviving ranks can omit it. Reissuing after rank 7 restarts can include some ranks’ values twice unless operation identity and input state are durable. Exactly-once arithmetic requires more than calling the collective again.

The HPX communicator’s generation and cached exception give an operation identity and consistent local failure propagation. They do not reconfigure membership. A future that waits on an and_gate sized for eight cannot become a correct seven-site collective merely because one locality disappeared.

Fault-tolerant collectives need an explicit policy:

• abort the job and restore from a checkpoint;
• shrink membership and restart the step;
• reconstruct the missing contribution from replicated state;
• declare the operation failed and let a higher layer decide;
• use an algorithm and runtime designed for recoverable process failure.

Each policy changes ordering, state ownership, and the meaning of a generation. The current article executed none of them. The supported claim is narrower: the success paths are checked, same-operation exceptions are cached in current HPX source, and participant loss remains an unmeasured recovery gap.

A timeout alone does not solve that gap. It can tell a waiting rank that an expected completion took too long. It cannot tell whether a peer is dead, paused, partitioned, or about to deliver a value that another participant already included. Returning a partial sum after the timeout would change the collective’s membership without agreement. Retrying on the same buffers could duplicate an uncertain contribution. Recovery therefore needs a membership decision and an operation state transition, not only a clock. The consensus investigation reached the same distinction: suspicion can trigger a protocol, but suspicion is not proof of failure.

the cost of the wrong shape reached beyond the call

A collective’s operational cost is not only its median duration.

Memory pressure matters. Recursive doubling needs a temporary full payload in the simple implementation. Ring and recursive halving can operate chunk by chunk, but library staging, registration, and protocol buffers add storage. On a GPU, an extra 500 MiB temporary can reduce batch size or KV-cache capacity even when the collective itself is fast.

Tail behavior matters. The one-mebibyte recursive-doubling run had a 1,003-microsecond median and a 2,779-microsecond maximum. A synchronous training step waits for the slowest required collective. A rare long call can extend every worker’s step and reduce useful accelerator occupancy.

CPU progress matters. The overlap probe reached 0.500 milliseconds with polling, but those progress calls used CPU cycles. If the same core feeds kernels, handles requests, or runs a task scheduler, moving wait time out of the collective span can move work into another bottleneck.

Initialization and reuse matter. The Rostam archive’s multi_use path reuses a communicator. A single-use path also pays naming, component creation, or setup. Comparing a warmed reusable MPI communicator with one-shot HPX creation would mix lifecycle cost with transport cost.

Correctness constraints matter. A faster schedule that reorders a noncommutative operator is not an optimization. A floating-point change may be acceptable for training and unacceptable for a bitwise reproducibility requirement. That decision belongs in the result contract before timing begins.

The financial translation needs a production workload and fleet model. This article has neither. Multiplying the local 570-microsecond ring advantage by an invented number of training steps and GPUs would create a dollar figure without utilization, overlap, step structure, power, or checkpoint behavior. The measured result stops at mechanism and time.

the original future finally had a path

The opening line returns one HPX future:

double total = all_reduce(comm, my_value, std::plus{}).get();

On the pinned flat communicator path, each site action stores its value in the server’s slot. One and_gate waits for every site. One finalizer reduces the slots. Every callback reads the result.

On the pinned hierarchical path, values reduce through a path of smaller flat communicators and broadcast back down. User generation (k) becomes internal generations (2k-1) and (2k). Fewer than sixteen sites collapse to one flat call by default. The source proves those statements. Local HPX tests prove the checked shapes return the expected values in one runtime process.

The five MPI implementations expose alternatives the one-line API hides. The naive root centralizes bytes. The tree reduces root fan-in but moves a full array at every level. Recursive doubling removes the broadcast and wins the small same-host cases. The ring sends uniform chunks for many rounds and wins the one-mebibyte run. Recursive halving and doubling matches the ring’s modeled bytes in fewer rounds but loses to it on this shared-memory path, which is precisely where the ideal model runs out.

The Rostam archive adds actual nodes and a harder result. Removing the flat server through a direct channel was slower in eight selected HPX cases and faster in one. Native MPI remained substantially ahead. The default threshold of sixteen cannot be defended from the source, the local logical-site tests, or that archive.

Eight machines can all return the same sum and still take paths different enough to change latency, bandwidth, memory, tails, overlap, and failure behavior. The call stopped looking like one primitive once the values were made to move.