CUDA Streams, Events, and NCCL Fundamentals: The Foundation Layer of Multi-GPU Communication

Ders Haritası#

1. CUDA streams — paralel execution
2. Default stream vs explicit streams
3. CUDA events — fine-grained sync
4. Async memcpy — overlap data transfer
5. NCCL nedir, niye gerek?
6. Collective operations: allreduce, broadcast, all-gather, reduce-scatter
7. Ring allreduce algorithm
8. Communication overlap with computation
9. torch.distributed
   API
10. Production patterns: gradient bucketing, NVLink optimization

1. CUDA Streams — Paralel Execution#

Default behavior#

Kernel A → Kernel B → Kernel C → ...

Multiple streams = parallelism#

import torch

stream1 = torch.cuda.Stream()
stream2 = torch.cuda.Stream()

with torch.cuda.stream(stream1):
    a = torch.matmul(x1, y1)   # stream1'de

with torch.cuda.stream(stream2):
    b = torch.matmul(x2, y2)   # stream2'de paralel

torch.cuda.synchronize()       # iki stream tamamlanana kadar bekle

Niye performans kazandırıyor?#

LLM'de uygulama#

• Forward pass + previous gradient communication paralel
• Different layer'lar farklı stream'lerde
• Data loading + compute overlap (CPU→GPU async copy)

2. Default Stream vs Explicit Streams#

Default stream (stream 0)#

• Sync per-default
• Kernel A bitmeden Kernel B başlamaz
• Pratik için simple ama performance bırakıyor

Explicit stream'ler#

torch.cuda.current_stream()       # şu anki stream
torch.cuda.set_stream(stream)     # change current

Stream dependency#

event = torch.cuda.Event()

with torch.cuda.stream(stream1):
    a = compute_something()
    event.record(stream1)         # stream1'de bu noktayı işaretle

with torch.cuda.stream(stream2):
    event.wait(stream2)           # stream2 bekle, event'e ulaşana kadar
    b = use_a_result(a)

Common pattern: pipelining#

# Layer-wise pipelining
for layer in model.layers:
    with torch.cuda.stream(compute_stream):
        output = layer(input)

    with torch.cuda.stream(comm_stream):
        # Sync gradient'ler async olarak
        all_reduce(gradient)

Sınırlar#

• Memory stream'ler arası paylaşılıyor → race condition riski
• Synchronization doğru yapılmazsa data corruption
• Debug zor — stream'ler nondeterministic kompozisyon

3. CUDA Events — Fine-Grained Synchronization#

event_a = torch.cuda.Event()
event_b = torch.cuda.Event()

event_a.record()  # şu anki point'i kaydet
a = compute_something()  # stream çalışıyor

event_b.record()  # işlem bitiş point'i

# Wait
event_b.synchronize()  # CPU bu event'e kadar bekle

Timing#

start = torch.cuda.Event(enable_timing=True)
end = torch.cuda.Event(enable_timing=True)

start.record()
# ... workload ...
end.record()
torch.cuda.synchronize()

ms = start.elapsed_time(end)
print(f"Took {ms:.2f} ms")

Cross-stream synchronization#

producer_stream = torch.cuda.Stream()
consumer_stream = torch.cuda.Stream()

with torch.cuda.stream(producer_stream):
    a = produce_data()
    event = torch.cuda.Event()
    event.record(producer_stream)

with torch.cuda.stream(consumer_stream):
    event.wait(consumer_stream)   # producer bitene kadar bekle
    b = consume_data(a)

Use case'ler#

1. Pipelining: stream'ler arası dependency
2. Timing: precise kernel duration
3. Async copy + compute overlap
4. Multi-GPU sync: cross-device events

4. Async Memcpy — Data Transfer Overlap#

Sync copy (slow)#

data_cpu = torch.randn(1000, 1000)
data_gpu = data_cpu.cuda()       # sync: CPU bekler transfer bitsin
result = compute(data_gpu)

Async copy (fast)#

import torch.cuda

data_cpu = torch.randn(1000, 1000).pin_memory()  # pinned memory
data_gpu = torch.empty_like(data_cpu, device="cuda")

stream = torch.cuda.Stream()
with torch.cuda.stream(stream):
    data_gpu.copy_(data_cpu, non_blocking=True)  # async
    # While copy is happening, CPU can do other work
    do_cpu_work()

# Wait for copy
stream.synchronize()
result = compute(data_gpu)

Pinned memory neden?#

LLM training'de#

• Data loading: GPU eğitim yaparken sonraki batch yükleniyor
• PyTorch DataLoader
  num_workers > 0
  +
  pin_memory=True
  standart
• CPU offload: ZeRO Stage 3 optimizer states CPU'ya, async transfer

Pattern#

train_loader = DataLoader(
    dataset,
    batch_size=32,
    num_workers=4,        # paralel data loading
    pin_memory=True,      # pinned CPU memory
)

for batch in train_loader:
    batch = batch.cuda(non_blocking=True)  # async transfer
    output = model(batch)
    loss = criterion(output, target)
    loss.backward()

5. NCCL Nedir, Niye Gerek?#

Problem#

Her GPU CPU'ya gönder → CPU sumla → CPU'dan GPU'lara dağıt

NCCL çözüm#

• NVLink (within node): GPU-GPU direct, ~600 GB/s
• InfiniBand / RDMA (across nodes): GPU memory direct over network
• PCIe (fallback): GPU-CPU-GPU, slower

NCCL operations#

1. all_reduce: tüm GPU'ların değerini sumla + herkese dağıt
2. broadcast: bir GPU'dan diğerlerine
3. all_gather: her GPU'nun parçasını birleştir + herkese ver
4. reduce_scatter: sumla + parçalara ayrılıp dağıt
5. send/recv: point-to-point

NCCL vs alternatif#

• NCCL: NVIDIA GPUs, en hızlı
• MPI: CPU + GPU, daha generic, biraz yavaş
• Gloo: CPU primary, multi-platform, yavaş
• OpenSHMEM: scientific computing legacy

6. Collective Operations Detayda#

All-reduce#

GPU 0: [1, 2, 3]
GPU 1: [4, 5, 6]
GPU 2: [7, 8, 9]
GPU 3: [10, 11, 12]

After all_reduce (sum):
GPU 0: [22, 26, 30]
GPU 1: [22, 26, 30]
GPU 2: [22, 26, 30]
GPU 3: [22, 26, 30]

Broadcast#

GPU 0 (master): [1, 2, 3]
GPU 1: [0, 0, 0]
GPU 2: [0, 0, 0]

After broadcast from GPU 0:
GPU 0: [1, 2, 3]
GPU 1: [1, 2, 3]
GPU 2: [1, 2, 3]

All-gather#

GPU 0: [1]
GPU 1: [2]
GPU 2: [3]
GPU 3: [4]

After all_gather:
GPU 0: [1, 2, 3, 4]
GPU 1: [1, 2, 3, 4]
GPU 2: [1, 2, 3, 4]
GPU 3: [1, 2, 3, 4]

Reduce-scatter#

GPU 0: [1, 2, 3, 4]
GPU 1: [5, 6, 7, 8]
GPU 2: [9, 10, 11, 12]
GPU 3: [13, 14, 15, 16]

After reduce_scatter (sum):
GPU 0: [28]       (1+5+9+13)
GPU 1: [32]       (2+6+10+14)
GPU 2: [36]
GPU 3: [40]

Complexity#

7. Ring All-Reduce Algoritması#

Naive (kötü)#

Tüm GPU'lar → GPU 0 (master) → tüm GPU'lar

Ring all-reduce#

GPU 0 ↔ GPU 1 ↔ GPU 2 ↔ GPU 3 ↔ GPU 0

Phase 1: scatter-reduce

Phase 2: all-gather

Toplam communication#

2 × (N-1)/N × M bytes per GPU

Optimal niye?#

• Hiçbir GPU bottleneck değil — herkes paralel
• Bandwidth optimal: 2(N-1)/N tüm ring algorithm'ler için tight bound
• Scalable: N büyüdükçe latency artmıyor (constant)

Modern alternative: tree all-reduce#

NVIDIA InfiniBand SHARP#

8. Communication Overlap with Computation#

Naive (slow)#

Step 1: forward pass         (1 sec)
Step 2: backward pass        (2 sec)
Step 3: all_reduce gradient  (0.5 sec)
Step 4: optimizer step       (0.1 sec)

Total: 3.6 sec per iteration

Overlap (fast)#

Forward + previous all_reduce paralel
Backward + current all_reduce paralel (gradient ready oldukça)

Total: 3 sec per iteration  (~15% faster)

Implementation#

import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP

model = DDP(model, device_ids=[local_rank])
# DDP backward sırasında gradient'leri bucket'larla allreduce eder
# Compute continues, comm happens

Gradient bucketing#

Bucket 1 (25MB): allreduce start
Bucket 2 (25MB): backward continuing
Bucket 3 (25MB): backward continuing
...

Bucket size#

DDP(model, device_ids=[local_rank], bucket_cap_mb=50)

Profiling#

9.

torch.distributed

API#

Initialization#

import torch.distributed as dist
import os

# Environment variables (torchrun otomatik set eder):
# MASTER_ADDR, MASTER_PORT, RANK, WORLD_SIZE, LOCAL_RANK

dist.init_process_group(backend="nccl")
local_rank = int(os.environ["LOCAL_RANK"])
torch.cuda.set_device(local_rank)

# Cleanup
dist.destroy_process_group()

Launch#

# Single node, 8 GPU
torchrun --nproc_per_node=8 train.py

# Multi-node (örn. 2 node × 8 GPU)
torchrun --nnodes=2 --nproc_per_node=8 \
         --rdzv_id=foo --rdzv_backend=c10d \
         --rdzv_endpoint=master_addr:29500 \
         train.py

Basic ops#

# All-reduce
tensor = torch.tensor([1.0, 2.0]).cuda()
dist.all_reduce(tensor, op=dist.ReduceOp.SUM)
# Şimdi tüm rank'lar aynı sum'ı alır

# Broadcast
if dist.get_rank() == 0:
    tensor = torch.tensor([3.0, 4.0]).cuda()
else:
    tensor = torch.empty(2).cuda()
dist.broadcast(tensor, src=0)

# Gather (all_gather)
tensor = torch.tensor([dist.get_rank()]).cuda()
gathered = [torch.empty(1).cuda() for _ in range(dist.get_world_size())]
dist.all_gather(gathered, tensor)

DistributedSampler#

from torch.utils.data.distributed import DistributedSampler

sampler = DistributedSampler(dataset, shuffle=True)
loader = DataLoader(dataset, batch_size=32, sampler=sampler)

for epoch in range(epochs):
    sampler.set_epoch(epoch)  # different shuffle per epoch
    for batch in loader:
        ...

Common patterns#

10. Production Patterns: Gradient Bucketing, NVLink Optimization#

Pattern 1: Gradient bucketing#

# Llama 3 8B for 8x A100
DDP(model, device_ids=[local_rank], bucket_cap_mb=50, gradient_as_bucket_view=True)

• gradient_as_bucket_view=True
  : memory saving + faster
• bucket_cap_mb=50
  : orta-büyük bucket (overlap dengeli)

Pattern 2: NVLink topology awareness#

nvidia-smi topo -m

        GPU0    GPU1    GPU2 ...
GPU0     X     NV12   NV12  ...
GPU1   NV12     X     NV12  ...
...

Pattern 3: Cross-node InfiniBand#

ibstat              # InfiniBand status
ibping              # latency check
nccl-tests          # NCCL benchmark

Pattern 4: NCCL env tuning#

export NCCL_DEBUG=INFO              # debug logs
export NCCL_IB_DISABLE=0             # InfiniBand enable
export NCCL_NET_GDR_LEVEL=PHB        # GPU Direct RDMA optimization
export NCCL_TREE_THRESHOLD=0         # tree algorithm always
export NCCL_BUFFSIZE=8388608         # 8MB buffer

Pattern 5: Topology-aware sharding#

11. Mini Egzersizler#

1. Stream pattern: 2 GPU, 4 layer'lı model. Stream'lerle pipeline nasıl?
2. All-reduce bandwidth: 8 GPU, 1GB gradient. Naive vs ring all-reduce communication?
3. Communication overlap: 8 GPU, allreduce 0.5s, backward 1.5s. Overlap yoksa total? Overlap'le maksimum saving?
4. NCCL debug: Multi-node training'de NCCL hata veriyor. Investigation adımları?
5. NVLink vs PCIe: NVLink 600 GB/s, PCIe 64 GB/s. 1GB allreduce için fark?

Bu Derste Neler Öğrendik?#

Sıradaki Ders#