Distributed Data Parallel

Data parallelism manual pada chapter sebelumnya mengajarkan isi perutnya teknik; DistributedDataParallel (DDP) membungkus mekanisme itu menjadi komponen siap pakai dengan desain yang berbeda: satu proses per perangkat, bukan satu proses yang mengatur semua perangkat. Setiap proses membangun modelnya sendiri, mengolah pecahan data dari DistributedSampler, dan backward() otomatis memicu penggabungan gradien antar proses lewat all-reduce — tanpa satu baris kode penggabungan manual. Dengan begitu GIL tidak lagi menjadi penghambat dan komunikasinya tumpang tindih dengan hitungan. Chapter ini membedah anatomi DDP, menjalankan inisialisasi grup proses sungguhan di backend CPU gloo, lalu menutup dengan launcher torchrun dan jebakan yang paling sering mengejutkan pemula.

Anatomi DDP: Proses, Rank, dan World Size #

Tiga istilah membentuk kosakata DDP. Rank adalah nomor identitas satu proses (mulai 0); world size adalah jumlah total proses; dan grup proses adalah kesepakatan komunikasi yang dibangun lewat init_process_group. Setiap proses memegang seluruh model (bedakan dengan model parallelism pada chapter berikutnya), tetapi data dibagi merata lewat sampler sehingga batch efektif = batch per proses × world size. Penggabungan gradien berlangsung lewat all-reduce — algoritma Ring yang lazim: setiap proses mengirim sebagian gradien ke tetangga dan menerima bagian lain secara bergilir, sehingga bandwidth terbagi rata tanpa ada satu perangkat yang menjadi pusat kemacetan:

Berkas ddp_cpu.py menjalankan seluruh pola itu secara nyata di CPU: inisialisasi backend gloo (backend komunikasi CPU yang juga dipakai lintas node), pembungkusan model DDP, all-reduce langsung, hingga pelatihan empat epoch dengan DistributedSampler:

"""DistributedDataParallel di jalur CPU: init_process_group gloo, world_size 1."""

import os

import torch
import torch.distributed as dist
from torch.utils.data import DataLoader, DistributedSampler, TensorDataset

torch.manual_seed(5)

N = 256
X = torch.randn(N, 6)
w_benar = torch.randn(6, 1)
Y = X @ w_benar + 0.1 * torch.randn(N, 1)


def main():
    os.environ.setdefault("MASTER_ADDR", "127.0.0.1")
    os.environ.setdefault("MASTER_PORT", "29613")
    dist.init_process_group(backend="gloo", rank=0, world_size=1)

    print("backend tersedia       :", dist.get_backend())
    print("world_size             :", dist.get_world_size())
    print("rank proses ini        :", dist.get_rank())

    # ---------- all-reduce gradien: kontrak inti DDP ----------
    grad = torch.tensor([1.0, 2.0, 3.0])
    dist.all_reduce(grad, op=dist.ReduceOp.SUM)
    print("all-reduce SUM (rank 1):", grad.tolist(), "(jumlah dari semua rank)")

    # ---------- model dibungkus DDP ----------
    model = torch.nn.Sequential(
        torch.nn.Linear(6, 48),
        torch.nn.Tanh(),
        torch.nn.Linear(48, 48),
        torch.nn.Tanh(),
        torch.nn.Linear(48, 1),
    )
    ddp = torch.nn.parallel.DistributedDataParallel(model)
    opt = torch.optim.SGD(ddp.parameters(), lr=0.03)

    dataset = TensorDataset(X, Y)
    sampler = DistributedSampler(dataset, num_replicas=1, rank=0, shuffle=True, seed=5)
    loader = DataLoader(dataset, batch_size=32, sampler=sampler)

    print()
    print("jumlah batch per epoch :", len(loader), f"({len(dataset)} sampel / 32)")
    indeks_awal = list(iter(sampler))[:8]
    print("indeks epoch-0 batch-1 :", indeks_awal, "...")
    for epoch in range(4):
        sampler.set_epoch(epoch)  # wajib: shuffle berbeda tiap epoch
        for x, y in loader:
            opt.zero_grad()
            loss = ((ddp(x) - y) ** 2).mean()
            loss.backward()
            opt.step()
        with torch.no_grad():
            mse = float(((model(X) - Y) ** 2).mean())
        print(f"epoch {epoch + 1} selesai, MSE seluruh data: {mse:.6f}")

    dist.destroy_process_group()
    print()
    print("destroy_process_group  : selesai")


if __name__ == "__main__":
    main()

Output yang Dihasilkan:

backend tersedia       : gloo
world_size             : 1
rank proses ini        : 0
all-reduce SUM (rank 1): [1.0, 2.0, 3.0] (jumlah dari semua rank)

jumlah batch per epoch : 8 (256 sampel / 32)
indeks epoch-0 batch-1 : [99, 70, 55, 138, 66, 19, 48, 54] ...
epoch 1 selesai, MSE seluruh data: 0.900974
epoch 2 selesai, MSE seluruh data: 0.136688
epoch 3 selesai, MSE seluruh data: 0.065360
epoch 4 selesai, MSE seluruh data: 0.062745

destroy_process_group  : selesai

Mari kita telaah kode di atas baris per baris:

1. Baris 19-20 (alamat master): dua variabel lingkungan MASTER_ADDR dan MASTER_PORT memberi tahu semua proses titik temu untuk berjabat tangan; di sini 127.0.0.1:29613 karena seluruh proses berada di satu mesin. Saat dipakai torchrun, variabel ini diisi otomatis.
2. Baris 20 (init_process_group): backend gloo berjalan di CPU — pilihan yang tepat untuk pengujian tanpa GPU dan untuk komunikasi lintas node berbasis CPU. Di GPU produksi, backend lazimnya nccl. Grup dibangun dengan rank=0, world_size=1: satu proses yang mengelola dirinya sendiri.
3. Baris 26-29 (all-reduce langsung): dist.all_reduce menghitung ReduceOp.SUM terhadap tensor di semua rank. Dengan world size 1 hasilnya tidak berubah — inilah bukti jujur bahwa pengujian kita berjalan pada satu proses; dengan N rank, tensor yang sama akan berisi jumlah dari N salinan.
4. Baris 39 (pembungkusan DDP): sekali dibungkus, ddp(x) menjalankan forward seperti biasa, tetapi loss.backward() otomatis mengantarkan gradien ke all-reduce sebelum opt.step() — kode penggabungan manual dari chapter sebelumnya tidak perlu ditulis lagi.
5. Baris 43-51 (DistributedSampler): sampler membagi data merata: num_replicas=1 berarti proses ini menerima seluruh 256 sampel; dengan 4 proses, tiap rank menerima 64 sampel tanpa tumpang tindih. Baris sampler.set_epoch(epoch) wajib dipanggil agar pengacakan berbeda tiap epoch — kelalaian paling umum pemakai DDP.
6. Baris 61 (destroy_process_group): grup proses wajib ditutup sebelum program selesai agar sambungan komunikasi terlepas dengan rapi.

Menjalankan Banyak Proses dengan torchrun #

Kode di atas sengaja diuji pada world_size=1 agar dapat diverifikasi penuh di mesin CPU ini. Pola yang sama dipakai utuh di GPU — yang berubah hanya launcher. Untuk menjalankan satu proses per GPU, perintahnya:

torchrun --nproc-per-node=2 ddp_cpu.py

Launcher torchrun menyalin skrip ke dua proses, mengatur MASTER_ADDR/MASTER_PORT, lalu memanggil init_process_group dengan LOCAL_RANK yang berbeda di tiap proses (pola lengkapnya memilih torch.device("cuda", int(os.environ["LOCAL_RANK"]))). Dua tabel ringkas memetakan konsekuensinya:

DDP membagi data ke banyak perangkat dengan model utuh di tiap proses — model tetap harus muat di memori satu perangkat. Bagaimana bila modelnya sendiri terlalu besar untuk satu GPU? Kita berpindah ke strategi yang memecah modelnya: Model Parallelism.

Rujukan resmi DDP — termasuk diagram all-reduce dan aturan sampler — tersedia di docs.pytorch.org/docs/stable/generated/torch.nn.parallel.DistributedDataParallel.html dan pytorch.org/tutorials/intermediate/ddp_tutorial.html .