PyTorch DDP over FrogNet Network Shared Memory

Experimental results, profiling, and analysis of an unmodified PyTorch DDP workload using Gloo and Psychedelic FrogTorch across heterogeneous machines in Seattle and New York.

1. Description

I built an experimental PyTorch distributed backend called Psychedelic FrogTorch. The purpose was to determine whether PyTorch's existing distributed contract could be implemented against network shared memory rather than conventional rank-to-rank messaging.

PyTorch itself is unmodified and installed from pip. Gloo is the reference implementation.

The programming difference is straightforward. In the direct backend, shard, ranks do not send gradients to other ranks. Each rank publishes its contribution to FrogNet Memory, and the other ranks independently read the contributions they require.

For an all-reduce, state is published under a generational address of the form:

(psyc10d, ars.<round>.<slice>, <rank>)

The producer does not identify the consumers, and consumers do not require the producer's physical location. The synchronization contract presented to PyTorch remains an ordinary all-reduce.

I also implemented three experimental variants:

shard is the direct comparison with Gloo. faithful is deliberately less efficient and provides a control for implementing more conventional whole-bucket behavior over FrogNet Memory. fresh_only and every_k investigate what else becomes possible once state is independently observable; they are not used in the direct Gloo comparison.

2. Experiments

The DDP campaign uses four heterogeneous machines: rank 0 is the Seattle AI host; rank 1 is the Seattle gateway; rank 2 is a second Seattle node reached over a separate Internet path; and rank 3 is in New York, approximately 2,400 miles away. The Internet-connected FrogNet networks are joined by tunnels.

Measured from rank 0, throughput to the nearby rank has been approximately 25–37 MB/s, while the slower paths have been approximately 0.6–0.8 MB/s. The topology therefore contains roughly 40× throughput asymmetry.

The training workload is a six-layer MLP with hidden width 1024, batch size 64, 4,216,833 parameters, and approximately 16.9 MB of gradients per step. Every rank starts with seed 17 and an identical model but different training data. Each run has eight warmup steps followed by 40 measured steps.

Correctness is checked using both a common 4,096-sample held-out set and SHA-256 of the flattened final parameter vector.

The archive contains eight DDP campaigns and ten experimental arms. Comparisons between implementations are made only within the same campaign because the physical network changed substantially between campaigns. gradient_as_bucket_view is tested both disabled and enabled.

In addition to end-to-end training time, later FrogTorch runs instrument publication time, phase 1 and phase 2 time, per-peer waits, per-peer transfer times, observed transfer rates, and bytes offered and transmitted by the FrogNet data plane.

3. Results

The three implementations preserving ordinary synchronization semantics produced:

Every rank within each implementation produces the same final parameter hash. The hashes differ between implementations because the floating-point summation orders differ. The held-out result is the same to four decimal places.

For the direct shard/Gloo comparison, there are four completed same-campaign pairs:

The direct backend is therefore 4–0 against the completed same-campaign Gloo controls.

Across the five completed direct-backend campaigns, median step times were 12.3, 12.9, 13.0, 13.4 and 15.3 seconds. Across six completed Gloo campaigns they were 13.1, 13.4, 22.0, 22.7, 29.8 and 62.0 seconds.

The skipping algorithms also produced useful experimental results, but because they intentionally alter synchronization semantics they are kept separate from the direct comparison.

4. Examination and analysis

Where shard spends its time

The current archive contains five successful direct-shard campaigns with complete phase instrumentation across all four ranks. Each rank contains 95 observations for each phase, giving 1,900 observations per phase across the direct-backend corpus.

Publishing the local contribution into the collective costs well under a millisecond at the median. The seconds are elsewhere. Across all 20 instrumented rank/campaign combinations, median publication is approximately 0.3–1.3 ms, while phase 1 and phase 2 are measured in seconds. The implementation is not spending seconds writing the local tensor into FrogNet Memory. It is spending them obtaining remote state.

Waiting and transferring are different costs

The link instrumentation separates waiting for state from transferring that state. This distinction matters. A representative direct run, ddp10006 rank 1, illustrates it:

Phase 1 on the slow paths is predominantly a transfer problem. Phase 2 is different: even a fast nearby transfer can be preceded by several seconds waiting for the required state to become available. A multi-second collective operation is therefore not equivalent to a multi-second network transfer. The instrumentation separates data movement from dependency.

W=3 versus W=4

Earlier phase decomposition on the three-rank LAN configuration showed very different magnitudes. At W=3, all on one local network, representative phase medians were on the order of:

• publish: 0.6 ms
• phase 1: 13.7 ms
• phase 2: 16.0 ms

Those two figures are not a clean world-size comparison: the W=3 measurement was taken on a LAN and the W=4 measurement spans 2,400 miles, so adding a rank and adding the wide-area path are confounded. What the pair does establish is that local publication stays near a millisecond in both. Separating world size from topology is exactly what the sweep described below is for.

Byte accounting

The amount of tensor data moved by Gloo and shard is useful because the idealized totals are essentially the same.

For tensor size S and world size W, ring all-reduce has per-rank communication volume:

2(W−1)/W × S

At W=4 this is 1.5S. For the 16.9 MB gradient in this experiment, that is approximately 25.35 MB per rank per all-reduce.

shard arrives at the same idealized total by a different route. In phase 1 a rank reads (W−1) slices of size S/W, and it does the same in phase 2:

2 × (W−1) × S/W = 1.5S

Again, approximately 25.35 MB per rank at W=4.

Measured shard byte telemetry was approximately 1.648 tensor equivalents, about 10% above the idealized 1.5 figure. The whole-bucket implementation was approximately 3.390 tensor equivalents. Putting the shared-memory programming surface underneath a more expensive data-movement pattern does not make the bytes free.

gradient_as_bucket_view

During the campaign I discovered gradient_as_bucket_view. In the clean same-machine, same-night A/B available in the archive, enabling it changed Gloo from 29.8 to 13.1 seconds median per step, approximately a 56% improvement.

This was large enough to become a candidate explanation for the entire result, so the implementations were subsequently tested with bucket views both disabled and enabled. With bucket views enabled, completed direct comparisons include both 1.03× and 1.75× in favor of shard.

Bucket views therefore explain a very large PyTorch/Gloo-side cost in at least one campaign, but they do not explain the full cross-campaign behavior.

Campaign variance

The five completed direct-backend campaign medians occupy 12.3–15.3 seconds, roughly a 24% band. Gloo produced 13.1–62.0 seconds. Even discarding the 62-second campaign gives 13.1–29.8 seconds, approximately a 127% band.

On the best campaign, Gloo almost catches shard. What changes dramatically is Gloo's sensitivity to the environment.

5. Interpretation

The current evidence points toward coupling, rather than byte count, as the important difference.

Gloo and shard have essentially the same idealized per-rank tensor traffic at W=4, but their dependencies are different. A ring collective couples rank progress through the collective communication sequence. A slow portion of that sequence can therefore affect collective completion beyond the rank directly attached to the slow path.

shard publishes state and allows ranks to read the required slices independently. The phase instrumentation makes that distinction visible: a rank can spend milliseconds transferring from one peer, seconds transferring from another, or seconds waiting for state that subsequently transfers in milliseconds.

That behavior is consistent with the campaign result. As the network environment changes, Gloo's step time changes dramatically while the direct FrogTorch backend remains in a relatively narrow band.

6. Things the PyTorch team needs to be aware of

First, PyTorch's backend abstraction worked. I approached c10d as a device-driver contract, put a very different communication model behind it, and was able to run an ordinary DDP workload without modifying PyTorch.

Second, PyTorch's own distributed acceptance tests have been valuable development tools. An earlier checkpoint was 31 pass / 60 skip / 5 fail out of 96 tests. That snapshot predates several subsequent fixes and should not be read as the backend's current score. The tests exposed real implementation problems, including future handling, tensor lifetime, store behavior and collective lifetime assumptions. I intend to rerun and publish the current result rather than present the historical number as current.

Third, the Gloo and shard byte models are unusually useful here. At W=4 they independently arrive at the same idealized 1.5 tensor equivalents per rank. That gives us a way to investigate communication structure without starting from implementations with radically different theoretical traffic requirements.

Fourth, gradient_as_bucket_view matters enormously in this workload. The observed 29.8 → 13.1 second Gloo change is too large to treat as incidental, and further comparisons need to continue controlling it.

Finally, the next experiment is already fairly well specified. The local ar_rank.py work fitted T(n) = a + b·n and found substantially different per-element slopes for the two implementations. What I have not yet run is the corresponding real-network world-size sweep.

I want to run the same size sweep at W = 2, 3, 4 and 6 on the actual network and fit a and b independently for Gloo and FrogTorch at each world size.

If the changing relationship is primarily in b and scales with world size in the manner predicted by the communication structure, then the data-movement/coupling explanation becomes considerably stronger. Optimizing the Python implementation or replacing it with C++ will move the implementation cost but will not remove that scaling term.

If the difference remains predominantly in a, or the slope relationship stays essentially independent of W, then the implementation itself deserves much more attention.

Either result is useful.

The existing campaign has given me enough information to know what experiment I want to run next.