Skip to content
Markdown

Geo-distributed training placement

Scope: assigning one pipeline-plus-data-parallel tasklet to each GPU on a slow, heterogeneous fabric. The page covers the DT-FM cost model, matrix measurement, workload sizing, placement search, rank-map launch, and production limits. It assumes static membership and homogeneous compute. Overlay and mesh networking covers routability; DiLoCo covers training that avoids per-step WAN synchronization.

What it is

A global rank encodes communication relationships. Under the rank convention used by DT-FM and the solver in scripts/geo_placement.py:

rank = replica_index * D_PP + stage_index

Contiguous ranks form one D_PP-stage pipeline. Ranks separated by D_PP form one data-parallel group. For D_PP = D_DP = 8:

             stage 0  stage 1  ...  stage 7
replica 0       0        1     ...     7
replica 1       8        9     ...    15
...
replica 7      56       57     ...    63

row    = one pipeline
column = one data-parallel group

The mapping is framework-specific. A launcher or training framework that derives groups with a different rank formula must translate the emitted map before use.

DT-FM formulates the outer problem as balanced graph partitioning with a non-additive communication objective. Each partition is one stage's data-parallel group. Given those partitions, two exact inner problems determine the pipelines:

  • Bottleneck perfect matching pairs devices between adjacent stage groups.
  • Open-loop travelling salesperson search orders the stage groups as a Hamiltonian path.

The balanced partition is NP-hard. The script uses topology-derived seeds and swap hill-climbing; it does not certify a global optimum.

Why use it

A random or host-enumeration rank order can place large transfers on the slowest links. DT-FM reports up to a 2.7x end-to-end improvement from its scheduler over random assignment in its ablation. Cases 3 to 5 in that evaluation use 64 V100 GPUs with network conditions emulated from geographic measurements, so the result is evidence for the method, not a general speedup guarantee.

The repository demo evaluates a scheduler-derived workload proxy against 200 random placements. On the DT-FM Table 2 world-wide matrix, the best topology-derived seed scores 51.50 model seconds against a 97.61 median, or 1.90x. On the script's explicitly synthetic two-datacenter example, the corresponding scores are 10.87 and 33.57, or 3.09x. These are communication-model scores. They are not wall-clock reproductions of the paper.

Rank remapping leaves the intended training algorithm unchanged. It can still change collective trees, reduction order, and floating-point roundoff, so bitwise-equivalent numerics are not guaranteed.

When to use it (and when not)

Situation Decision
Static GPUs span regions or providers and pairwise NCCL costs differ materially Measure the fabric and solve placement
Pipeline parallelism and a sharded parameter-server exchange match the model Use the objective directly after validating payload and message counts
The implementation uses ring all-reduce, FSDP collectives, tensor parallelism, or different rank groups Re-derive the traffic objective before using the map
Links are approximately uniform after accounting for host, NVLink, PCIe, and switch hierarchy Skip the WAN-specific solver; use the framework's topology-aware defaults
Per-step synchronization is slower than the useful compute window for every plausible map Reduce synchronization frequency; consider DiLoCo
GPU generations or stage compute times differ Extend the objective with compute and memory constraints
Workers join, leave, or fail during a run This static model is unsuitable; checkpoint and reschedule

Architecture

flowchart TD
    A["Measure directional NCCL transfer time"] --> B["Fit alpha and beta per device pair"]
    B --> C["Symmetric equal-payload cost matrix"]
    C --> D["Balanced stage partitions: heuristic"]
    D --> E["Bottleneck matching: exact"]
    E --> F["Coarsened stage graph"]
    F --> G["Hamiltonian path: exact Held-Karp"]
    G --> H["Device index to global rank map"]
    H --> I["Launcher and communicator construction"]

For directional measurements a_ij, b_ij and a_ji, b_ji, the script builds an equal-payload symmetric model:

alpha_ij = (a_ij + a_ji) / 2
beta_ij  = 2 / (1 / b_ij + 1 / b_ji)
T_ij(c, m) = m * alpha_ij / 1000 + 8 * c / beta_ij

c is decimal GB, beta is Gbps, and m is the number of messages represented by the aggregate payload. Harmonic-mean bandwidth preserves the sum of forward and reverse serialization times. DT-FM's paper averages directional bandwidth arithmetically; the script deliberately differs because arithmetic averaging does not preserve transfer time on an asymmetric link.

For stage group C_j, the sharded parameter-server term is:

DataP(C_j) = max over d in C_j of
             sum over d' in C_j, d' != d of
             2 * T_d,d'(c_dp / D_DP, dp_messages)

DataP = max over stage groups C_j of DataP(C_j)

The factor two represents shard distribution followed by redistribution of the aggregated shards. This is an all-to-all sharded parameter-server model, not ring all-reduce.

For candidate stage groups C_j and C_k:

edge(C_j, C_k) = min over perfect matchings M of
                  max over (d, d') in M of T_d,d'(c_pp, pp_messages)

Pipeline = 2 * minimum Hamiltonian-path sum of edge(C_j, C_k)
Comm     = DataP + Pipeline

The pipeline factor two represents forward activations and equal-sized backward activation gradients. If the two directions have different payloads or message counts, the pipeline term must be extended instead of doubled.

The exact Held-Karp solver is limited to D_PP <= 16. Bottleneck matching is polynomial. The outer balanced partition remains heuristic regardless of the exact inner solutions.

How to use it

1. Measure the fabric

Measure the same transport, interface, MTU, routing path, and NCCL version used by training. A single large transfer reports effective throughput; it cannot separate startup latency alpha from serialization bandwidth beta. Fit several payload sizes to:

elapsed_seconds = alpha_seconds + payload_gigabits / beta_gbps

The following is an unexecuted GPU reference template. Run it on exactly two processes with exclusive access to one GPU each. Run once with SENDER=0 and once with SENDER=1 for every device pair. Adjust the sizes to bracket the workload's actual messages.

# Reference template: requires PyTorch, two CUDA GPUs, and a working NCCL path.
import os
import time

import torch
import torch.distributed as dist

SIZES_MIB = (1, 4, 16, 64, 256)
REPETITIONS = {1: 100, 4: 50, 16: 20, 64: 10, 256: 4}
sender = int(os.environ["SENDER"])
local_rank = int(os.environ["LOCAL_RANK"])

torch.cuda.set_device(local_rank)
dist.init_process_group("nccl")
rank = dist.get_rank()
if dist.get_world_size() != 2 or sender not in (0, 1):
    raise RuntimeError("run exactly two ranks and set SENDER to 0 or 1")
peer = 1 - rank
samples = []

for size_mib in SIZES_MIB:
    nbytes = size_mib * 1024 * 1024
    buffer = torch.empty(nbytes, dtype=torch.uint8, device="cuda")
    dist.barrier()
    for _ in range(3):
        dist.send(buffer, peer) if rank == sender else dist.recv(buffer, peer)
    torch.cuda.synchronize()
    dist.barrier()
    torch.cuda.synchronize()

    started = time.perf_counter()
    for _ in range(REPETITIONS[size_mib]):
        dist.send(buffer, peer) if rank == sender else dist.recv(buffer, peer)
    torch.cuda.synchronize()
    elapsed = (time.perf_counter() - started) / REPETITIONS[size_mib]
    if rank == sender:
        samples.append((nbytes * 8 / 1e9, elapsed))

if rank == sender:
    x_mean = sum(x for x, _ in samples) / len(samples)
    y_mean = sum(y for _, y in samples) / len(samples)
    denominator = sum((x - x_mean) ** 2 for x, _ in samples)
    slope = sum((x - x_mean) * (y - y_mean) for x, y in samples) / denominator
    intercept = y_mean - slope * x_mean
    residual = sum((y - (intercept + slope * x)) ** 2 for x, y in samples)
    total = sum((y - y_mean) ** 2 for _, y in samples)
    r_squared = 1.0 - residual / total
    if slope <= 0 or intercept < 0 or r_squared < 0.98:
        raise RuntimeError("poor linear fit; inspect congestion and message-size regimes")
    print(
        "sender=%d alpha_ms=%.3f beta_gbps=%.3f r2=%.5f"
        % (sender, intercept * 1e3, 1.0 / slope, r_squared)
    )

dist.destroy_process_group()

Example static launch on each of two hosts:

SENDER=0 torchrun \
  --nnodes=2 \
  --nproc-per-node=1 \
  --node-rank="$NODE_RANK" \
  --master-addr="$MASTER_ADDR" \
  --master-port=29500 \
  measure_pair.py

Repeat with SENDER=1. Build JSON with one row per physical GPU. Matrix indices identify devices; labels are locality labels used only for human-readable output and may repeat:

{
  "labels": ["Oregon", "Oregon", "Virginia", "Virginia"],
  "alpha_ms": [[0, 1, 35, 37], [1, 0, 36, 38], [34, 36, 0, 1], [37, 39, 1, 0]],
  "beta_gbps": [[0, 20, 1.2, 1.1], [20, 0, 1.1, 1.0], [1.3, 1.2, 0, 20], [1.2, 1.1, 20, 0]]
}

The script validates shape, finiteness, non-negative latency, and positive off-diagonal bandwidth. Zero bandwidth on the diagonal is valid.

2. Size the workload

Use measured transmitted bytes, not parameter counts alone.

  • c_pp is the aggregate activation payload crossing one stage boundary for one modelled data partition.
  • pp_messages is the number of runtime transfers that carry that aggregate activation payload.
  • c_dp is the actual per-stage buffer entering gradient aggregation. Its element type may differ from the optimizer or scheduler's assumed type.
  • dp_messages is the number of transfers to each peer in each of the two sharded exchange rounds.
  • Payloads are decimal GB because bandwidth is decimal Gbps.

Trace one representative step when the framework fuses, chunks, compresses, or overlaps communication. The ratio c_pp / (c_dp / D_DP) is diagnostic only; latency multiplicity, the all-peer sum, the worst-device maximum, and pipeline matching all affect the result.

3. Solve

python3 scripts/geo_placement.py solve \
  --matrix fabric.json \
  --pp 8 \
  --dp 8 \
  --c-pp "$C_PP_GB" \
  --c-dp "$C_DP_GB" \
  --pp-messages "$PP_MESSAGES" \
  --dp-messages "$DP_MESSAGES" \
  --baseline-samples 200 \
  --seed 7

The output includes the model score, the random-placement distribution, group locality, and:

rank_map=(...)

rank_map[i] is the global rank assigned to matrix row and device index i. Increase --budget or --restarts only after recording the seed and comparing repeated runs. A lower score is a better result under this model; it is not proof of global optimality.

The built-in demonstration is reproducible:

python3 scripts/geo_placement.py demo --baseline-samples 200
python3 scripts/geo_placement.py selftest

The world-wide example uses all 28 inter-region values from DT-FM Table 2 and the paper's 5 ms, 2 Gbps intra-region assumption. The two-datacenter example is synthetic and is labelled as such by the command.

4. Launch

This reference template assumes one participating GPU and one process per host. Under that restriction, node_rank equals global rank. For more than one participating GPU per host, the launcher must map local processes to ranks explicitly; permuting node_rank is insufficient.

Static torchrun rendezvous must run on the host assigned global rank 0. Derive that host after solving:

master_index=
for i in "${!rank_map[@]}"; do
  if [[ ${rank_map[$i]} -eq 0 ]]; then
    master_index=$i
    break
  fi
done
[[ -n $master_index ]] || { echo "rank_map has no rank 0" >&2; exit 1; }

MASTER_ADDR=${ips[$master_index]}
MASTER_PORT=29500

for i in "${!ips[@]}"; do
  rank=${rank_map[$i]}
  ssh "ubuntu@${ips[$i]}" \
    "export NCCL_SOCKET_IFNAME='=wg0'; \
     CUDA_VISIBLE_DEVICES=0 torchrun \
       --nnodes=${#ips[@]} \
       --nproc-per-node=1 \
       --node-rank=$rank \
       --master-addr=$MASTER_ADDR \
       --master-port=$MASTER_PORT \
       train.py" &
done
wait

NCCL_SOCKET_IFNAME='=wg0' selects the exact interface name. Replace wg0 with the production data interface. Set GLOO_SOCKET_IFNAME only if the application creates a Gloo process group. The rendezvous address must be reachable independently of NCCL interface selection.

How to develop with it

The following block was executed with NumPy and the repository script. It compares both exact solvers with brute force, distinguishes min-sum from min-max matching, verifies directional symmetrisation, rejects malformed input, and checks the Held-Karp boundary.

import itertools

import numpy as np

from scripts.geo_placement import Fabric, bottleneck_matching, open_loop_tsp

rng = np.random.default_rng(19)

for _ in range(100):
    n = int(rng.integers(2, 6))
    cost = rng.random((n, n))
    expected = min(
        max(cost[i, permutation[i]] for i in range(n))
        for permutation in itertools.permutations(range(n))
    )
    actual, matching = bottleneck_matching(cost)
    assert abs(actual - expected) < 1e-12
    assert abs(max(cost[i, matching[i]] for i in range(n)) - expected) < 1e-12
print("bottleneck matching: 100/100 equal to brute force")

for _ in range(100):
    n = int(rng.integers(2, 7))
    cost = rng.random((n, n))
    cost = (cost + cost.T) / 2
    np.fill_diagonal(cost, 0.0)
    expected = min(
        sum(cost[p[i], p[i + 1]] for i in range(n - 1))
        for p in itertools.permutations(range(n))
    )
    actual, path = open_loop_tsp(cost)
    assert abs(actual - expected) < 1e-9
    assert sorted(path) == list(range(n))
print("open-loop TSP: 100/100 equal to brute force")

probe = np.array([[0.0, 6.0], [6.0, 10.0]])
value, matching = bottleneck_matching(probe)
assert value == 6.0 and matching == [1, 0]
assert min(probe[0, 0] + probe[1, 1], probe[0, 1] + probe[1, 0]) == 10.0
print("min-sum bottleneck 10.0; min-max bottleneck 6.0")

fabric = Fabric(
    alpha_ms=np.array([[0.0, 10.0], [30.0, 0.0]]),
    beta_gbps=np.array([[0.0, 1.0], [3.0, 0.0]]),
    labels=["a", "b"],
)
directional_sum = 0.010 + 0.25 * 8 / 1.0 + 0.030 + 0.25 * 8 / 3.0
assert abs(2 * fabric.transfer_s(0.25)[0, 1] - directional_sum) < 1e-12
assert fabric.transfer_s(0.25)[0, 0] == 0.0
print("directional transfer-time sum preserved")

try:
    Fabric(
        np.array([[0.0, np.nan], [1.0, 0.0]]),
        np.array([[0.0, 1.0], [1.0, 0.0]]),
        ["a", "b"],
    )
except ValueError:
    print("non-finite matrix rejected")
else:
    raise AssertionError("non-finite matrix accepted")

try:
    open_loop_tsp(np.zeros((17, 17)))
except ValueError:
    print("Held-Karp boundary rejected at 17 stages")
else:
    raise AssertionError("17-stage exact TSP accepted")

Executed output:

bottleneck matching: 100/100 equal to brute force
open-loop TSP: 100/100 equal to brute force
min-sum bottleneck 10.0; min-max bottleneck 6.0
directional transfer-time sum preserved
non-finite matrix rejected
Held-Karp boundary rejected at 17 stages

Why textbook Kernighan-Lin gain is misaligned

Kernighan-Lin optimises an additive cut-weight gain. DT-FM's data-parallel term is a maximum over per-device peer sums. Improving a device that is not currently the slowest may leave the DataP term unchanged, although it can still change the pipeline term. Each pipeline edge is also a bottleneck matching cost that must be recomputed after a swap.

The paper makes the same qualitative observation and reports that its KL baseline still outperforms random placement, but underperforms its tailored local search. A single min-sum Hungarian assignment also solves the wrong matching objective. Hungarian assignment can still be used inside a threshold-feasibility method; the repository script uses augmenting paths instead.

How to maintain it

  • Timestamp matrices and retain the NCCL version, interface, MTU, congestion control, routing path, and measurement command.
  • Re-measure after topology, provider, kernel, overlay, or instance changes.
  • Recompute payload bytes and message counts after model, precision, batch, pipeline-depth, fusion, or collective changes.
  • Record the matrix hash, solver seed, budget, restart count, score, random distribution, and emitted rank map with every run.
  • Compare model ordering with measured step time on at least the best solution, an alternate seed, and a random map. A cost model is useful only if it ranks real placements correctly.
  • Run python3 scripts/geo_placement.py selftest after any solver change.

DT-FM source limits

The repository audit is pinned to commit cda948f02111ad194ce108d4c65b85258afd0c19, dated 2022-07-03. The remote has no newer commit or tag as of 2026-07-13.

  • The current scheduler uses hidden size 2024, while its launcher uses 2048, and divides activation bytes by 1024**3 while bandwidth is decimal Gbps. The demo corrects both and produces 0.512 decimal GB. This is a correction to a scheduler-derived proxy, not a published paper payload.
  • The scheduler models a 0.650 GB FP32 stage buffer. The shipped training path casts the model and flattened gradient buffer to FP16, which makes the same parameter-count approximation about 0.325 GB. The executed workload and scheduler proxy therefore disagree.
  • Equation 2 assumes a sharded parameter server. The shipped local launch script selects central_ps, the argument default is allreduce, and no shipped shell launcher selects sharded_ps.
  • Table 2 and current config.py disagree on three inter-region entries. The traffic-control generator adds another mismatch by using 10 ms within a world-wide region instead of the paper's 5 ms. The script's world-wide example uses Table 2 plus the paper's intra-region values.
  • The paper calls Table 2 values delay, not RTT. Neither the paper nor the repository establishes that they are round-trip measurements. Claims that DT-FM doubled an intended RTT are therefore unsupported.
  • The paper promises a polynomial-time bottleneck-matching proof in supplementary material. The proof is absent from both arXiv v4 and the official NeurIPS supplement.

The paper's Cases 3 to 5 are emulations built from geographic measurements. Its later FluidStack run uses 32 A40 GPUs across US Mid and US East without artificial network control. Reported peak-FLOPS fractions are 27.4% for GPT3-1.3B with 40 layers and a 4K batch, and 26.4% and 29.7% for GPT3-6.7B and GPT3-13B with batch 1024.

How to run it in production

  1. Establish authenticated, encrypted, mutually routable paths before measuring. DT-FM's checked-in strongSwan research configuration uses null encryption, a published pre-shared key, SHA-1/MODP-1024, and a 3DES fallback. It is historical measurement plumbing, not a production VPN template.
  2. Bind NCCL only when automatic interface selection is unsuitable. Exact-name selection uses a leading equals sign. Gloo can use a separate interface list, and rendezvous reachability is a separate requirement.
  3. Benchmark congestion control on the real path. Loss-based CUBIC can underperform on lossy or shallow-buffer WAN paths. If BBR is listed in net.ipv4.tcp_available_congestion_control, test it with fq as the preferred pacing qdisc. Current BBR can pace without fq, with higher resource cost.
  4. Tune NCCL socket parallelism from measurements, not a fixed WAN rule. Current NCCL documents NCCL_SOCKET_NTHREADS from 1 to 16; NCCL_SOCKET_NTHREADS * NCCL_NSOCKS_PERTHREAD must not exceed 64. More sockets can help when per-socket throughput is limited and consume additional CPU.
  5. Validate the selected map with end-to-end step time, NCCL logs, per-link counters, straggler timing, and checkpoint recovery. Reject the placement if model ranking and observed ranking disagree.

For symmetric netem tests, if the target number is known to be an RTT, apply half of it to egress at both endpoints. Applying the full RTT at both endpoints produces approximately twice that ping RTT. Shaping only one side reproduces ping RTT but creates asymmetric one-way delay and is unsuitable for this symmetric model. The tc-netem manual also warns that sender-egress placement can be distorted by TCP Small Queues; receiver-ingress placement is more realistic for TCP experiments.

The effect of any DT-FM emulation mismatch on relative scheduler speedup is indeterminate without rerunning all compared placements under corrected delays.

Failure modes

Symptom Cause Action
Solver barely beats the random median Fabric is nearly uniform, payloads are wrong, or the search budget is too small Validate bytes and message counts, compare seeds, then skip placement if the fabric is genuinely uniform
Model score improves but step time does not Collective pattern, overlap, contention, or message multiplicity differs from the objective Trace the real communication graph and re-derive the model
A poor device dominates every solution Every device belongs to both one pipeline and one data-parallel group Exclude the device or match it to its least-bad neighbours; the TSP cannot remove it
The job hangs at rendezvous MASTER_ADDR does not point to the host assigned rank 0 Derive the master host from rank_map
Multi-GPU hosts receive incorrect ranks The one-process-per-host launch template was reused Build an explicit per-local-process rank assignment
A node joins, leaves, or fails Membership and the rank map are static Checkpoint, re-measure if needed, solve again, and restart
Solver rejects D_PP > 16 Exact Held-Karp state grows exponentially Add an approximate Hamiltonian-path solver and validate it against exact small cases
Results vary across seeds The outer search is heuristic Record the distribution and choose only after real step-time validation

References

Related: Overlay and mesh networking · Pipeline parallelism · DiLoCo · Recipe: DiLoCo · NCCL collectives and algorithms · Comms-compute overlap · Distributed training