Two-Face: Combining Collective and One-Sided Communication for Efficient Distributed SpMM

Multiplying a huge sparse matrix by a dense one across a supercomputer is almost entirely a communication problem, and there are two ways to move the data — broadcast everything to everyone and waste bandwidth, or fetch exactly what each node needs and pay software overhead on every request; measured across eight large matrices each wins on half of them by up to eleven times, so Two-Face (Illinois) splits a single matrix into regions and runs both at once, though it ends up nearly twice as slow as the baseline on two of the eight.

Why this problem is worth solving

The operation is unglamorous and sits underneath a great deal of current machine learning.

The tension has a name in distributed programming.

What people did before, and where it broke

• Replicate the whole dense matrix everywhere with a collective gather. The simplest sparsity-unaware approach, and it makes every subsequent access local. Where it broke. Memory. On one of the evaluated matrices at a moderate dense width, full replication exceeds the memory of a single node, so the experiment cannot even run.
• Shift blocks of the dense matrix around the ring in cyclic steps. The strongest published baseline, replicating a tunable number of blocks and rotating them between computation phases. Where it broke. It transfers rows regardless of whether the destination needs them, and the waste grows in direct proportion to how wide the dense matrix is. Higher replication factors also run out of memory — one setting fails on half the matrices at the widest configuration tested.
• Issue fine-grained one-sided reads for exactly the rows needed. The sparsity-aware extreme, moving nothing unnecessary and requiring no synchronisation. Where it broke. Per-request software overhead and extra round trips. Measured against the collective approach it wins on half the matrices and loses on the other half, with the margin reaching an order of magnitude in both directions.
• Pick whichever of the two suits the input. The obvious response to that measurement, and nobody appears to have gone further. Where it broke. It treats the matrix as having one character. Real matrices do not — the same matrix contains column ranges wanted by nearly every node and column ranges wanted by one.
• Coarse-grained one-sided fetches of whole blocks. A middle option using remote reads but at block granularity. Where it broke. It has the memory problem of replication without the bandwidth efficiency of fine-grained requests, and performs worse than either extreme on average.
• The gap. Everyone treated the choice of communication style as a property of the algorithm or at best of the input matrix. It is neither — it is a property of a region within the input matrix, and once you see that, the question becomes where to draw the line rather than which side to stand on.

How Two-Face schedules communication

The idea

The right way to move the data is not a property of the algorithm, or even of the matrix, but of a region inside it. The measurement that sets this up is unusually clean. Run the fine-grained one-sided scheme against the collective scheme on eight large matrices and each wins on four, with margins reaching 11.5 times one way and 10.5 times the other. The natural reading is that you should pick per matrix, and that reading is what the paper argues past. Nonzeros are not evenly distributed inside a matrix — a column range that is dense means the corresponding dense rows are wanted by almost every node, which is exactly the situation a broadcast is built for, while a column range that is nearly empty means one node wants one row, which is exactly what a one-sided fetch is built for. Both situations exist inside the same matrix, often side by side. So the design cuts the matrix into narrow stripes and classifies each one independently, and a stripe wanted by nobody is simply never transferred at all. Discovering that the right mechanism varies not just between matrices but between regions of one matrix is the same observation as the adaptive dataflow work covered elsewhere on this site, arrived at for communication rather than for loop order.

Because the two halves run concurrently, the objective is to balance them, not to minimise either. This is the part that makes the classification tractable, and it is a genuinely different optimisation problem from the one you would write down by instinct. The broadcasting stripes and the fetching stripes are processed by separate thread groups at the same time, so total time is whichever half takes longer, and the best split is the one where they finish together. That turns into a clean procedure: score each stripe by what it would cost as a fetch — a weighted combination of how many distinct dense rows it needs and how many nonzeros it holds — sort ascending, and convert stripes from broadcast to fetch in that order until the two predicted times meet. Sorting by cheapest-to-fetch does double duty, because a broadcast stripe costs a fixed amount regardless of its contents, so maximising how many stripes become fetches also minimises the broadcast side. The coefficients come from a one-time regression on a single matrix, and they are informative in their own right — the fitted numbers say a byte moved by one-sided fetch costs about 18.5 times a byte moved collectively, which is the whole trade-off in a single ratio. Turning the choice between two execution styles into a calibrated model with a tunable balance point is the same move as the adaptive execution work covered elsewhere on this site.

The communication decision reaches all the way down into how the data is stored, and each choice is bad at the other job. This is the detail that makes the design feel earned rather than assembled. A broadcast stripe is stored row by row, because that lets a thread walk one row of nonzeros, accumulate the result in a thread-local buffer, and write it into the output with a single atomic operation. A fetch stripe is stored column by column, because before a thread can request anything it must know which distinct dense rows the stripe needs, and column order makes that immediate. The cost is that column order makes buffering impractical, so the fetching side pays one atomic per nonzero — which is precisely why the model bothers to account for computation time on the fetching half and neglects it on the broadcasting half, and why only eight of a node’s 128 threads are given to it. The authors tried storing fetch stripes row-major instead and report that it was not faster, because determining which columns held nonzeros became drastically more expensive. That is a negative result worth having. All of it is decided once, ahead of time, which is what confines the approach to full-graph training where a single matrix is reused for hundreds of epochs — the paper states plainly that it is incompatible with the sampled mini-batch training that the batch-level graph partitioning work covered elsewhere on this site is built to serve.

Classifying one stripe

The partitioning is what makes this clean. Each node owns a band of sparse rows plus the matching bands of the dense input and output, which means output accesses are always local and only dense-input accesses can be remote. Everything the paper says about communication is about that one matrix.

Why it works

• Density within a column range predicts how many nodes want the same rows. A dense column range means many nodes hold nonzeros there → so the corresponding dense rows are wanted broadly and a single collective serves all of them → while a nearly empty range means one node wants one row, which a fetch delivers with nothing wasted.
• Concurrency turns minimisation into balancing. The two halves are processed by separate thread groups at the same time → so total time is the larger of the two rather than their sum → and the optimum is where they finish together, which is a condition you can solve for directly.
• A broadcast stripe costs the same regardless of what is in it. The whole group of dense rows moves whether the stripe is nearly full or nearly empty → so converting the cheapest stripes to fetches maximises how many broadcasts are eliminated per unit of fetching time → and one sort order satisfies both the balance condition and the total cost.
• Each storage order buys exactly what its half needs. Row order lets a thread accumulate a whole row before one atomic write, which suits the broadcast half where computation must be cheap → column order lets a thread immediately list which dense rows to request, which suits the fetch half where the request list is the critical path → and the authors measured that swapping them is worse.
• Coalescing turns scattered row requests into fewer transfers. Needed rows are grouped into contiguous runs before being requested → so software overhead is paid per run rather than per row → and when the dense matrix is narrow enough that overhead dominates, the grouping deliberately spans small gaps and fetches a few rows nobody wanted.

What it costs (trade-offs and scope)

• It is substantially slower than the baseline on two of eight matrices, at every configuration. On the two largest social graphs the absolute times are close to twice the baseline — 20.24 seconds against 11.57 at one width, and 86.62 against 52.77 at another. At the narrowest dense width two further matrices also lose.
• The failure mode is its own collectives, and the paper diagnoses it precisely. On those two matrices the average multicast has 35.7 and 43.5 recipients, against 5.7 for the next highest matrix. When nearly every node wants the data, the cyclic shifting of the baseline beats a broadcast, and the effect only appears at high node counts.
• The classification model never looks at recipient count. It scores stripes by rows needed and nonzeros held. The paper explicitly names the alternative — classify by how many nodes need a dense stripe — and leaves it for future work, which is candid given that this is exactly what goes wrong.
• Scaling from one node to 64 makes one matrix slower. The average improvement is 7.47 times with a best case of 12.12, but the worst case is 0.76 times.
• Preprocessing is expensive and must be amortised. Around 24 times the cost of a single multiplication on average excluding file I/O, ranging up to 102 times, and 134 times on average when I/O is counted. Break-even is roughly 15 multiplications at one dense width and 3 at a wider one. The step is not parallelised across nodes, which the authors note makes these a pessimistic bound.
• Incompatible with sampled or mini-batch training, and the paper says so directly. Each iteration would use a different reduced matrix, forcing preprocessing to re-run. Only full-graph training is supported, which rules out much of how graph neural networks are trained at scale.
• The model neglects two of the four cost components. Computation on the broadcast stripes and everything about local stripes are assumed negligible, justified by giving those 120 threads against eight. Reasonable, and it is why one matrix with dense regions classified as fetches fails to improve — its per-nonzero atomics were never in the model.
• Coefficients are fitted per system and the surface is not flat. Calibration is a regression on one matrix at one configuration, deliberately kept small so it can be done at installation. Perturbing the fitted values by 25 percent mostly makes things worse, in one case by 52 percent.
• Processors only, and one baseline excluded by design. Everything is measured on a processor-based supercomputer; the argument that accelerators would behave similarly is a reasoned expectation, not a measurement. A published sparse-shifting alternative is left out because its partitioning would force extra all-to-all traffic for the graph workloads targeted here.

The evaluation

Check yourself

Sources

The paper is “Two-Face: Combining Collective and One-Sided Communication for Efficient Distributed SpMM” by Charles Block, Gerasimos Gerogiannis, Charith Mendis, Ariful Azad and Josep Torrellas, published at ASPLOS 2024 Volume 2, pages 1200-1217, DOI 10.1145/3620665.3640427. The first two authors are noted as contributing equally.

This explainer is written from the full text of the published paper. Everything above comes from its own sections 2 through 7: the description of the partitioning under which only dense-input accesses can be remote; the motivating comparison across eight matrices showing each communication style winning on half of them with margins reaching 11.5 and 10.5 times, and the note that full replication exceeds single-node memory for one matrix; the worked example distinguishing local, broadcast and fetch nonzeros by how many nodes need the corresponding dense rows; the section 4 structure of megatiles, sparse stripes and dense stripes, including the observation that a multicast may have a single destination and that an unwanted dense stripe is never transferred; the thread split and the rule that broadcast communication completes before broadcast computation begins while the fetching half overlaps them; the row-major and column-major storage decision with its stated consequences for buffering and atomics; the section 4.2 cost model with its three components, the neglect of broadcast computation and local stripes, the balance condition, the per-stripe scoring and ascending-order conversion, and the acknowledgement of an unexplored alternative classification based on how many nodes need a dense stripe; the section 5 algorithms including request coalescing across gaps at narrow dense widths, the tuning parameters, and the statement that the method is in its current form incompatible with sampling or mini-batching; and the section 7 evaluation covering the speedups at three dense widths, the absolute execution times, the execution breakdown identifying inefficient collectives on two matrices and expensive fetch-side computation on a third, the strong scaling results with their best and worst cases, the multicast recipient profiling, the preprocessing costs with and without file I/O and the resulting break-even counts, and the sensitivity study perturbing the fitted coefficients.