Distributed training

launchhelm.torch handles DDP under torchrun/mp.spawn. Launch the example with two CPU ranks:

LAUNCHHELM_URL=... LAUNCHHELM_API_KEY=... \
LAUNCHHELM_ORGANIZATION=... LAUNCHHELM_PROJECT=... \
torchrun --master_addr=127.0.0.1 --master_port=29500 --nproc_per_node=2 torch_ddp.py

torchrun --standalone also works on hosts where the machine hostname resolves.

def main() -> None:
    dist.init_process_group("gloo")
    rank = dist.get_rank()
    torch.manual_seed(0)

    model = DistributedDataParallel(
        torch.nn.Sequential(
            torch.nn.Linear(8, 16), torch.nn.ReLU(), torch.nn.Linear(16, 1)
        )
    )
    optimizer = torch.optim.SGD(model.parameters(), lr=0.01)
    x = torch.randn(64, 8)
    y = x.sum(dim=1, keepdim=True)

    tr = lht.init("quickstart", name="ddp-run", system_metrics=True)
    tr.watch(model, optimizer)

    for step in range(20):
        optimizer.zero_grad()
        loss = torch.nn.functional.mse_loss(model(x), y)
        loss.backward()
        optimizer.step()
        # step() is the collective control point: call it on every rank at
        # the same cadence.
        tr.step(step=step, loss=loss.item())

    tr.finish()
    dist.destroy_process_group()
    if rank == 0:
        print("run finished:", tr.run.run_id)

python-torch_ddp.py · verbatim from sdks/python/examples/torch_ddp.py

Call lht.init on every rank. Rank 0 creates (or resumes) the run and each rank joins it as its own participant rank-<r> — every rank shows up in LaunchHelm with its rank, host and world size.

Control is collective

Call tr.step on every rank at the same cadence. A control issued from the run's Controls tab applies on every rank at the same step, so DDP replicas stay identical. If any rank fails to apply it, every rank raises DivergedStateError and the command records outcome_unknown — a partially applied control is surfaced, not silently absorbed. Pass on_partial_failure="continue" to lht.init to let the run keep training instead.

Every result also reports how many ranks applied the control (applied_on, for example 511 of 512). The count comes from the ranks' own outcomes, gathered by rank 0, so the Controls tab shows "on all 512 processes" only when all 512 reported success. A custom collective handler reports its own count with run.runtime.report_applied_on(count, of) from inside the handler.

Every rank also reports its step with its heartbeat (every 15 seconds), so the run's processes table shows a rank that stalls or falls behind even though only rank 0 logs metrics.

Peers wait for each other at the control point; a rank that stalls more than 30 minutes trips the collective timeout.

log_ranks="zero" (default) logs metrics only on rank 0; every rank still registers as a participant. Pass log_ranks="all" to log from every rank. system_metrics=True runs the collector on local_rank 0 of each node.

FSDP

save_checkpoint expects a full state dict — for FSDP models, configure a full state dict before saving.