netcl wiki
tutorials

Tutorial: Data-Parallel Multi-GPU Training

Tutorial: Data-Parallel Multi-GPU Training

This tutorial guides you through scaling deep learning training across multiple GPUs on a single workstation using netcl's data-parallel engine.

1. The Theory: What is Data Parallelism?

Data parallelism is the most efficient scaling technique when a neural network model comfortably fits inside the VRAM of a single GPU, but you want to decrease training wall-clock time by distributing mini-batch computations across $N$ devices.

flowchart TD GB["Global Batch (Size = B)"] --> S0["Shard 0 (Size = B / 2)
GPU 0 (Device 0)"] GB --> S1["Shard 1 (Size = B / 2)
GPU 1 (Device 1)"] S0 --> F0["Forward Pass & Tape Backward"] --> G0["Local Gradient ∇L₀"] S1 --> F1["Forward Pass & Tape Backward"] --> G1["Local Gradient ∇L₁"] G0 --> AR["all_reduce(mean)
∇L = (∇L₀ + ∇L₁) / 2"] G1 --> AR AR --> U0["GPU 0: Optimizer Update"] AR --> U1["GPU 1: Optimizer Update"]

Mathematical Formulation

Given a global batch $\mathcal{B}$ split into $N$ disjoint worker shards $\mathcal{B}_1, \dots, \mathcal{B}_N$, each worker $i$ independently computes its local loss gradient:

$$\mathbf{g}_i = \frac{1}{|\mathcal{B}_i|} \sum_{x \in \mathcal{B}_i} \nabla_\theta \ell(f(x; \theta), y)$$

Before applying the optimizer update, workers synchronize gradients using a mean reduction:

$$\bar{\mathbf{g}} = \frac{1}{N} \sum_{i=1}^N \mathbf{g}_i$$

Because each device applies identical averaged gradients $\bar{\mathbf{g}}$ to identical initial weights $\theta_t$, model weights remain mathematically synchronized across all GPUs throughout training:

$$\theta_{t+1} = \theta_t - \eta \cdot \text{Update}(\bar{\mathbf{g}})$$

2. Setting Up the Distributed Device Manager

Unlike single-device scripts, data-parallel scripts instantiate netcl.distributed.DeviceManager to discover and allocate queues across available GPUs:

import numpy as np
import netcl.autograd as ag
from netcl.core.device import manager
from netcl.distributed import DeviceManager, prepare_replicas, sync_grads, broadcast_params, shard_batch
from netcl.nn import Linear, ReLU, Dropout, Sequential
from netcl.optim import AdamW

# 1. Discover all physical GPUs on the host
dist_mgr = DeviceManager()
queues = dist_mgr.get_queues()
n_devices = len(queues)

print(f"Allocated {n_devices} device queues:")
for i, q in enumerate(queues):
    print(f"  Replica {i}: {q.device.name}")

3. Preparing Model Replicas

The prepare_replicas function clones model parameters across each target device queue:

# Create primary model architecture on default device
primary_queue = queues[0]
master_model = Sequential(
    Linear(primary_queue, 784, 256), ReLU(), Dropout(p=0.1),
    Linear(primary_queue, 256, 128), ReLU(), Dropout(p=0.1),
    Linear(primary_queue, 128, 10),
)

# Replicate parameters onto all device queues
param_replicas = prepare_replicas(master_model.parameters(), queues=queues)

# Create an independent optimizer for each replica
optimizers = [
    AdamW(params, lr=1e-3, weight_decay=1e-4)
    for params in param_replicas
]

Each replica holds an independent copy of model parameters in its local GPU memory buffer, ready for concurrent kernel execution.

4. The Multi-GPU Training Loop

Below is the complete data-parallel training iteration:

# Helper to evaluate forward pass for a specific parameter replica
def replica_forward(params, x):
    # params: [W0, b0, W1, b1, W2, b2]
    h = ag.add(ag.matmul(x, ag.transpose(params[0])), params[1])
    h = ag.relu(h)
    h = ag.add(ag.matmul(h, ag.transpose(params[2])), params[3])
    h = ag.relu(h)
    return ag.add(ag.matmul(h, ag.transpose(params[4])), params[5])

# Synthetic training batch (batch size = 128, features = 784)
batch_x = np.random.randn(128, 784).astype(np.float32)
batch_y = np.random.randint(0, 10, size=(128,)).astype(np.int32)

# Step 1: Shard inputs evenly across replicas
x_shards = shard_batch(batch_x, num_shards=n_devices)
y_shards = shard_batch(batch_y, num_shards=n_devices)

losses = []

# Step 2: Concurrent forward and backward passes per replica
for r_idx, (q, params) in enumerate(zip(queues, param_replicas)):
    # Transfer shard to local device memory
    x_dev = Tensor.from_host(q, x_shards[r_idx])
    y_dev = Tensor.from_host(q, y_shards[r_idx])

    with ag.Tape() as tape:
        logits = replica_forward(params, ag.tensor(x_dev))
        loss = ag.cross_entropy(logits, ag.tensor(y_dev))

    # Compute local gradients on replica device
    tape.backward(loss)
    losses.append(loss.value.to_host()[0])

# Step 3: Mean-reduce gradients across all replicas
sync_grads(param_replicas)

# Step 4: Step each replica's optimizer
for opt in optimizers:
    opt.step()
    opt.zero_grad()

# Step 5: Broadcast parameters to guarantee zero numerical divergence
broadcast_params(src_params=param_replicas[0], dst_param_groups=param_replicas, root=0)

mean_loss = float(np.mean(losses))
print(f"Parallel step completed. Mean Batch Loss: {mean_loss:.4f}")

5. Using the High-Level data_parallel_step Orchestrator

For production scripts, netcl provides data_parallel_step which wraps sharding, forward-backward execution, gradient reduction, and optimizer updates into a single call:

from netcl.distributed import data_parallel_step

def step_forward(queue, xb, yb, params):
    with ag.Tape() as tape:
        logits = replica_forward(params, ag.tensor(xb))
        loss = ag.cross_entropy(logits, ag.tensor(yb))
    return loss, tape

# Train across epochs
for epoch in range(5):
    for batch_x, batch_y in dataloader:
        loss = data_parallel_step(
            step_forward,
            param_replicas,
            optimizers,
            (batch_x, batch_y),
        )
        broadcast_params(src_params=param_replicas[0], dst_param_groups=param_replicas, root=0)

6. Best Practices for Workstation Multi-GPU

  1. Scale the Learning Rate: When training with $N$ devices and constant per-device batch size $B$, the effective global batch size increases to $N \times B$. Under the Linear Scaling Rule, consider scaling learning rate $\eta' = N \cdot \eta$ with a brief learning rate warmup.
  2. Pinned Host Memory: Ensure data loading utilizes netcl.data.DataLoader with pinned memory buffers so batch transfers do not stall GPU computation.
  3. Always Broadcast After Step: Call broadcast_params periodically or after each step to prevent floating-point non-associativity drift across different GPU hardware architectures.