Pipeline Parallelism

Model parallelism murni menyerahkan separuh waktunya pada pengangguran: saat tahap awal bekerja, tahap akhir menunggu, dan sebaliknya. Pipeline parallelism mengisi waktu kosong itu dengan memecah satu batch menjadi banyak micro-batch, lalu mengalirkannya bertingkat — micro-batch berikutnya masuk tahap awal tepat ketika micro-batch sebelumnya pindah ke tahap akhir. Kedua perangkat akhirnya sibuk hampir bersamaan, meski tidak pernah 100%: tahap pengisian (fill) dan pengurasan (drain) tetap menyisakan gelembung (bubble) waktu mati. Chapter ini membangun pipeline mini 2 tahap × 4 micro-batch yang berjalan nyata di CPU, menghitung efisiensinya dengan rumus m/(m+s-1), lalu membandingkan keluarga jadwal GPipe dan 1F1B yang dipakai sistem produksi.

Dari Eksekusi Berurutan ke Jadwal Bertingkat #

Mari kita bandingkan dua cara memproses 4 micro-batch pada 2 tahap. Eksekusi berurutan (model parallelism klasik) menuntaskan satu micro-batch penuh sebelum lanjut: 8 slot kerja yang seluruhnya serial — setiap saat hanya satu tahap yang bekerja. Pipeline fill-drain menjadwalkan ulang pekerjaan yang sama per siklus: pada siklus c, tahap0 memroses micro-batch c, sementara tahap1 memroses micro-batch c-1 yang sudah dituntaskan tahap0 pada siklus sebelumnya:

Total siklus berubah dari m × s (berurutan) menjadi m + s - 1 (pipeline), sehingga efisiensinya m/(m+s-1) — makin banyak micro-batch, makin kecil porsi gelembung. Berkas pipeline_mini.py menjalankan kedua jadwal sungguhan dan memverifikasi hasilnya identik:

"""Pipeline parallelism mini: 2 tahap, 4 micro-batch, jadwal fill-drain."""

import torch

torch.manual_seed(33)

D_IN, D_H, D_OUT = 16, 64, 4
N_MIRO = 4  # jumlah micro-batch
B_MIRO = 16

tahap0 = torch.nn.Sequential(
    torch.nn.Linear(D_IN, D_H),
    torch.nn.GELU(),
    torch.nn.Linear(D_H, D_H),
)
tahap1 = torch.nn.Sequential(
    torch.nn.GELU(),
    torch.nn.Linear(D_H, D_H),
    torch.nn.GELU(),
    torch.nn.Linear(D_H, D_OUT),
)

data = [torch.randn(B_MIRO, D_IN) for _ in range(N_MIRO)]
target = [torch.randn(B_MIRO, D_OUT) for _ in range(N_MIRO)]


def kerja_tahap(tahap, x):
    """Satu unit kerja tahap (di perangkat nyata berjalan paralel)."""
    return tahap(x)


# ---------- 1. Eksekusi berurutan murni (model parallelism klasik) ----------
print("=== Eksekusi berurutan: satu micro-batch sampai tuntas ===")
hasil_sekuensial = []
slot_sibuk = 0
for i, mb in enumerate(data):
    h = kerja_tahap(tahap0, mb)  # cpu:0 bekerja, cpu:1 menunggu
    slot_sibuk += 1
    out = kerja_tahap(tahap1, h)  # cpu:1 bekerja, cpu:0 menunggu
    slot_sibuk += 1
    hasil_sekuensial.append(out)
    print(f"langkah {i}: tahap0 -> mb{i}, lalu tahap1 -> mb{i}")
print(f"total slot tahap terpakai: {slot_sibuk} (2 tahap x 4 micro-batch)")

# ---------- 2. Pipeline: jadwal fill-drain (gaya GPipe) ----------
print()
print("=== Pipeline 2 tahap x 4 micro-batch (fill-drain) ===")
antar = {}  # aktivasi perbatasan yang "sedang terbang"
hasil_pipeline = {}
grafik = []
for siklus in range(N_MIRO + 2 - 1):  # m + s - 1 = 5 siklus
    kerja0 = f"mb{siklus}" if siklus < N_MIRO else "-"
    j = siklus - 1
    kerja1 = f"mb{j}" if 0 <= j < N_MIRO else "-"
    grafik.append((siklus, kerja0, kerja1))
    # tahap1 memproses micro-batch siklus-1 (hasil tahap0 siklus sebelumnya)
    if 0 <= j < N_MIRO:
        hasil_pipeline[j] = kerja_tahap(tahap1, antar[j])
    if 0 <= j < N_MIRO:
        del antar[j]
    # tahap0 memproses micro-batch baru pada siklus ini
    if siklus < N_MIRO:
        antar[siklus] = kerja_tahap(tahap0, data[siklus])

print(f"{'siklus':>6s} {'tahap0':>7s} {'tahap1':>7s}")
for siklus, k0, k1 in grafik:
    print(f"{siklus:>6d} {k0:>7s} {k1:>7s}")

slot_total = (N_MIRO + 2 - 1) * 2
slot_idle = slot_total - slot_sibuk
efisiensi = N_MIRO / (N_MIRO + 2 - 1)
print(f"siklus total        : {N_MIRO + 2 - 1} (m + s - 1)")
print(f"efisiensi m/(m+s-1) : {efisiensi:.2f} (bubble {1 - efisiensi:.0%} = {slot_idle} slot idle)")

# ---------- 3. Verifikasi hasil dan tabel efisiensi ----------
sama = all(
    torch.equal(hasil_sekuensial[i], hasil_pipeline[i]) for i in range(N_MIRO)
)
print()
print("=== Verifikasi ===")
print(f"hasil pipeline identik dengan sekuensial: {sama}")
print()
print(f"{'m':>4s} {'s=2':>8s} {'s=4':>8s} {'s=8':>8s}")
for m in (1, 2, 4, 8, 16):
    baris = [f"{m / (m + s - 1):>7.0%}" for s in (2, 4, 8)]
    print(f"{m:>4d} {baris[0]:>8s} {baris[1]:>8s} {baris[2]:>8s}")

Output yang Dihasilkan:

=== Eksekusi berurutan: satu micro-batch sampai tuntas ===
langkah 0: tahap0 -> mb0, lalu tahap1 -> mb0
langkah 1: tahap0 -> mb1, lalu tahap1 -> mb1
langkah 2: tahap0 -> mb2, lalu tahap1 -> mb2
langkah 3: tahap0 -> mb3, lalu tahap1 -> mb3
total slot tahap terpakai: 8 (2 tahap x 4 micro-batch)

=== Pipeline 2 tahap x 4 micro-batch (fill-drain) ===
siklus  tahap0  tahap1
     0     mb0       -
     1     mb1     mb0
     2     mb2     mb1
     3     mb3     mb2
     4       -     mb3
siklus total        : 5 (m + s - 1)
efisiensi m/(m+s-1) : 0.80 (bubble 20% = 2 slot idle)

=== Verifikasi ===
hasil pipeline identik dengan sekuensial: True

   m      s=2      s=4      s=8
   1      50%      25%      12%
   2      67%      40%      22%
   4      80%      57%      36%
   8      89%      73%      53%
  16      94%      84%      70%

Mari kita telaah kode di atas baris per baris:

1. Baris 27-29 (kerja_tahap): satu unit kerja tahap dienkapsulasi menjadi fungsi — di perangkat nyata, pemanggilan dari dua tahap yang berbeda berjalan di dua device secara bersamaan; di CPU uji kita keduanya berbagi satu proses.
2. Baris 34-43 (eksekusi berurutan): keempat micro-batch dituntaskan satu per satu — 8 slot kerja, dan di setiap saat salah satu tahap pasti menganggur. Inilah profil waktu model parallelism murni dari chapter sebelumnya.
3. Baris 51-63 (jadwal fill-drain): loop berjalan m + s - 1 = 5 siklus. Kamus antar menyimpan aktivasi perbatasan “yang sedang terbang”: tahap1 memproses micro-batch j = siklus - 1, lalu referensinya dihapus agar memori aktivasi hanya menampung micro-batch yang benar-benar berada di dalam pipa — bukan seluruh batch.
4. Baris 65-73 (grafik dan efisiensi): tabel siklus memperlihatkan gelembung secara harafiah: siklus 0 (tahap1 kosong saat pipa terisi) dan siklus 4 (tahap0 kosong saat pipa terkuras). Dari 10 slot tersedia, hanya 8 terpakai — efisiensi 0,80, gelembung 20%.
5. Baris 76-86 (verifikasi dan tabel): torch.equal membuktikan keluaran keempat micro-batch identik bit-per-bit dengan eksekusi berurutan — penjadwalan ulang tidak mengubah matematika. Tabel terakhir menunjukkan hukum skalanya: dengan 2 tahap, 16 micro-batch mencapai 94%; dengan 8 tahap, dibutuhkan micro-batch yang jauh lebih banyak untuk efisiensi yang sama.

Keluarga Jadwal: GPipe dan 1F1B #

Rumus efisiensi yang sama berlaku untuk semuanya: m/(m+s-1). Perbedaannya ada di memori aktivasi — GPipe menyimpan aktivasi seluruh micro-batch hingga fase backward dimulai, sedangkan 1F1B segera memulai backward sehingga puncak memori perbatasan jauh lebih rendah; inilah jadwal yang dipakai sistem seperti Megatron-LM.

Pipeline mengisi waktu kosong antar tahap, tetapi setiap perangkat masih menyimpan parameter utuh milik tahapnya beserta seluruh state optimizer-nya. Bagaimana bila yang mau kita pangkas justru state optimizer itu sendiri — memecahnya merata ke semua perangkat? Itulah pokok Parameter Sharding dan keluarga ZeRO pada chapter berikutnya.

Rujukan resmi pola penjadwalan pipeline — termasuk implementasi bertingkat antar perangkat — tersedia di tutorial pytorch.org/tutorials/intermediate/model_parallel_tutorial.html .