Horovod developer on retainer: DataLoader workers fork with same random.seed(42) across all ranks — augmentation masks identical on all 4 GPUs — gradient diversity equivalent to 1-rank run — BERT-large NER F1 plateaus at 0.78 — 3 weeks — Horovod developer on monthly retainer

horovod 0.28.1 • Python 3.11 • NVIDIA A100 40 GB × 4 • BERT-large NER fine-tuning • PyTorch 2.1.x • MPI backend • random.seed(42) set in main process • fork() copies parent RNG state to all 4×4 workers • collate_fn produces identical 15% token masks on all ranks • gradient all-reduce averages identical gradients • effective data diversity = 1 GPU • val F1 0.78 (same as 1-GPU baseline) • developer checks batch size, loss, Horovod config, NVLink — 3 weeks • fix: worker_init_fn seeded with hvd.rank() • val F1 0.78→0.84 • 1.5h retainer engagement

Eastfield Horovod Training Ltd. was fine-tuning BERT-large (340M parameters) on the CoNLL-2003 English NER dataset using 4 A100 40 GB GPUs with Horovod’s MPI backend. Their training pipeline used a custom collate_fn that applied MLM-style token masking (randomly masking 15% of tokens with probability controlled by random.random()) as a data augmentation strategy. The pipeline had been developed on a single GPU and then adapted for Horovod by adding hvd.init(), wrapping the optimizer with hvd.DistributedOptimizer, and pinning each process to its GPU with torch.cuda.set_device(hvd.local_rank()). Training throughput was correctly 4× faster than the single-GPU baseline (3,840 samples/second vs. 960 samples/second).

Validation F1, however, plateaued at 0.78 after 20 epochs, identical to the single-GPU baseline. The developer checked the effective batch size (correctly base_batch × 4 per step), the training loss curves (decreasing at a consistent rate), and the Horovod configuration (rank assignment and GPU pinning both correct). They verified NVLink bandwidth during training — at 100% utilization, suggesting gradient all-reduce was executing correctly. They tried augmentation probabilities of 0.10, 0.15, and 0.20. The validation F1 stayed at 0.78 for all variants. After three weeks the developer engaged a Horovod retainer developer.

The data loading and augmentation configuration at the time of the engagement:

# Eastfield Horovod Training Ltd. — BERT-large NER fine-tuning
# horovod 0.28.1, PyTorch 2.1.x, Python 3.11
# NVIDIA A100 40GB × 4, single node, MPI backend

import random
import numpy as np
import torch
import horovod.torch as hvd

# BUG: seed set in __main__ before hvd.init()
random.seed(42)
np.random.seed(42)
torch.manual_seed(42)

hvd.init()
torch.cuda.set_device(hvd.local_rank())

class NERDataset(torch.utils.data.Dataset):
    def __init__(self, examples):
        self.examples = examples

    def __getitem__(self, idx):
        return self.examples[idx]  # returns raw tokens + labels

    def __len__(self):
        return len(self.examples)

def collate_fn(batch):
    """Apply MLM-style token masking as augmentation."""
    input_ids_list, label_ids_list = [], []
    for tokens, labels in batch:
        masked_tokens = []
        for token in tokens:
            # BUG: random.random() uses Python's global random state
            # This state was set by random.seed(42) in __main__
            # fork() copies this state to ALL DataLoader workers on ALL ranks
            # All 4 ranks × 4 workers = 16 processes with identical RNG state
            if random.random() < 0.15:
                masked_tokens.append(103)  # [MASK] token ID
            else:
                masked_tokens.append(token)
        input_ids_list.append(torch.tensor(masked_tokens))
        label_ids_list.append(torch.tensor(labels))
    return torch.stack(input_ids_list), torch.stack(label_ids_list)

# DataLoader — no worker_init_fn to re-seed per worker
train_dataset = NERDataset(train_examples)
train_sampler = torch.utils.data.DistributedSampler(
    train_dataset,
    num_replicas=hvd.size(),
    rank=hvd.rank()
)
train_loader = torch.utils.data.DataLoader(
    train_dataset,
    batch_size=16,
    sampler=train_sampler,
    num_workers=4,     # 4 workers per rank; each forked from parent with seed=42
    collate_fn=collate_fn,
    # worker_init_fn=None  ← default; workers get parent's RNG state via fork()
)

# Result:
# Rank 0, worker 0: random sequence = [0.638, 0.025, 0.275, 0.223, ...]  (seed 42)
# Rank 1, worker 0: random sequence = [0.638, 0.025, 0.275, 0.223, ...]  (identical)
# Rank 2, worker 0: random sequence = [0.638, 0.025, 0.275, 0.223, ...]  (identical)
# Rank 3, worker 0: random sequence = [0.638, 0.025, 0.275, 0.223, ...]  (identical)
#
# For batch step N, all 4 ranks process different data shards (DistributedSampler
# correctly splits the dataset), BUT apply the SAME augmentation mask to their shard.
# The gradient computed by rank k at step N:
#   grad_k = grad(model, augmented_shard_k)
# Because augmentation is identical, relative token positions masked are proportional
# to the token ID distribution, not independently random. The gradient diversity
# that multi-GPU training is supposed to provide is absent.
#
# hvd.DistributedOptimizer all-reduce at step N:
#   mean_grad = (grad_0 + grad_1 + grad_2 + grad_3) / 4
# If augmentation sequences are identical, grad_0 ≈ grad_1 ≈ grad_2 ≈ grad_3
# (up to differences in the data shard), and mean_grad ≈ grad_0.
# The model receives essentially the same stochastic gradient it would see
# with 1-GPU training at the same effective batch size.

Why fork() copies the random state to all rank workers

Python’s random module maintains a global state object (a Mersenne Twister instance) in the Python interpreter. When random.seed(42) is called, this state is initialized to a deterministic sequence. PyTorch DataLoader with num_workers=4 creates worker processes using multiprocessing.Process with start_method='fork' (the default on Linux). The fork() system call creates child processes as byte-for-byte copies of the parent — including the entire Python heap, which contains the random module’s state object. Each worker starts with an exact copy of the parent’s random state at the moment it was forked.

Horovod uses MPI to launch one Python process per GPU. MPI launches all processes from the same executable in parallel, but the random.seed(42) call runs in each process independently before Horovod MPI coordination begins. Each Horovod rank’s main process has the same Python random state (seed 42). When each rank creates its DataLoader with num_workers=4, it forks 4 worker processes from that identical seed state. The result: all 16 workers (4 ranks × 4 workers) start with the same Mersenne Twister state and generate the same random number sequence when processing batches in the same order.

# Diagnostic: verify per-rank augmentation diversity

import horovod.torch as hvd
import random
import torch

def verify_augmentation_diversity(train_loader, num_batches=3):
    """
    Check whether augmentation masks differ across ranks.
    If all ranks produce identical masks, data diversity is absent.
    """
    for batch_idx, (input_ids, labels) in enumerate(train_loader):
        if batch_idx >= num_batches:
            break

        # Sample the augmentation mask: positions where tokens == 103 ([MASK])
        mask_positions = (input_ids == 103).float()   # (batch, seq_len)

        # Gather masks from all ranks using hvd.allgather
        all_masks = hvd.allgather(mask_positions)     # (num_gpus × batch, seq_len)

        if hvd.rank() == 0:
            rank_masks = all_masks.reshape(hvd.size(), -1, all_masks.shape[-1])
            # Compare rank 0 mask with rank 1 mask
            rank0 = rank_masks[0]  # (batch, seq_len)
            rank1 = rank_masks[1]  # (batch, seq_len)
            agreement = (rank0 == rank1).float().mean().item()
            print(f"Batch {batch_idx}: mask agreement rank0 vs rank1 = {agreement:.4f}")
            # If augmentation is DIVERSE (correct):  agreement ≈ 0.75-0.85
            # (positions agree by chance: p_agree = p_mask² + p_keep² = 0.15²+0.85² = 0.745)
            # If augmentation is IDENTICAL (buggy):  agreement = 1.0000

# Running the diagnostic on the buggy configuration:
# Batch 0: mask agreement rank0 vs rank1 = 1.0000   ← identical masks ✗
# Batch 1: mask agreement rank0 vs rank1 = 1.0000
# Batch 2: mask agreement rank0 vs rank1 = 1.0000
#
# Theoretical agreement under independent random masks (p=0.15):
# P(both masked) + P(both unmasked) = 0.15² + 0.85² = 0.0225 + 0.7225 = 0.745
# Observed agreement 1.0 >> 0.745 → masks are identical, not independent ✗

# Running the diagnostic AFTER the fix:
# Batch 0: mask agreement rank0 vs rank1 = 0.7441   ← near-theoretical ✓
# Batch 1: mask agreement rank0 vs rank1 = 0.7489   ✓
# Batch 2: mask agreement rank0 vs rank1 = 0.7452   ✓

The fix: seed each DataLoader worker with a rank-specific RNG

The fix is to pass a worker_init_fn to DataLoader that initializes the Python and NumPy random states using a seed that includes hvd.rank(). This function is called in each worker process after it is forked, overriding the parent’s copied state. With a rank-specific seed, each rank’s workers generate a distinct random sequence for augmentation, restoring the data diversity that multi-GPU training requires.

# Fixed configuration — Eastfield Horovod Training Ltd.
# horovod 0.28.1, PyTorch 2.1.x

import random
import numpy as np
import torch
import horovod.torch as hvd

hvd.init()
torch.cuda.set_device(hvd.local_rank())

# Base seed — same across all ranks for dataset determinism
BASE_SEED = 42

def worker_init_fn(worker_id):
    """
    Initialize each DataLoader worker with a rank- and worker-specific seed.
    Called in the worker process after fork(), overriding the parent's copied state.
    """
    # Combine rank and worker_id to get a unique seed per worker per rank
    seed = BASE_SEED + hvd.rank() * 10000 + worker_id
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    # Note: do NOT use torch.cuda.manual_seed here — DataLoader workers
    # are CPU processes; CUDA is not available inside workers.

train_loader = torch.utils.data.DataLoader(
    train_dataset,
    batch_size=16,
    sampler=train_sampler,
    num_workers=4,
    collate_fn=collate_fn,
    worker_init_fn=worker_init_fn,   # ← FIXED: unique seed per rank per worker
)

# After fix — seed values per worker:
# Rank 0: workers get seeds 42+0+0=42, 42+0+1=43, 42+0+2=44, 42+0+3=45
# Rank 1: workers get seeds 42+10000+0=10042, ...+1=10043, ...+2=10044, ...+3=10045
# Rank 2: workers get seeds 42+20000+0=20042, etc.
# Rank 3: workers get seeds 42+30000+0=30042, etc.
# All 16 workers have distinct seeds → distinct random sequences → diverse augmentation ✓

# Alternative using numpy Generator (preferred for new code):
def worker_init_fn_v2(worker_id):
    """Use numpy's new Generator API for better statistical properties."""
    # numpy.random.default_rng is not affected by numpy.random.seed()
    # so it's a cleaner approach than seeding the legacy numpy.random global state
    rng = np.random.default_rng(BASE_SEED + hvd.rank() * 10000 + worker_id)
    # Store rng in a module-level variable or pass via closure if collate_fn uses it
    import builtins
    builtins._worker_rng = rng
    # In collate_fn: use builtins._worker_rng.random() instead of random.random()

# Simplest alternative: use torch.initial_seed() for per-worker uniqueness
# PyTorch automatically sets a unique torch seed per worker based on the main
# process seed + worker_id. If your collate_fn can use torch.rand() instead of
# random.random(), you get worker-unique randomness automatically with no
# worker_init_fn required:
def collate_fn_torch_rng(batch):
    """Use torch.rand for augmentation — automatically unique per worker."""
    input_ids_list, label_ids_list = [], []
    for tokens, labels in batch:
        # torch.rand() uses PyTorch's per-worker seed (set automatically by DataLoader)
        mask = torch.rand(len(tokens)) < 0.15
        tokens_tensor = torch.tensor(tokens)
        tokens_tensor[mask] = 103  # [MASK]
        input_ids_list.append(tokens_tensor)
        label_ids_list.append(torch.tensor(labels))
    return torch.stack(input_ids_list), torch.stack(label_ids_list)

Validation

# Validation log — November 2026
# horovod 0.28.1, BERT-large (340M), 4× NVIDIA A100 40GB
# Task: CoNLL-2003 English NER, 20 epochs, batch_size=16 per GPU
# Augmentation: 15% token masking via collate_fn

# ---- Augmentation diversity diagnostic ----
# Before fix (worker_init_fn=None):
#   Batch 0: mask agreement rank0 vs rank1 = 1.0000  ← IDENTICAL ✗
#   Batch 1: mask agreement rank0 vs rank1 = 1.0000  ✗
#   Expected under independent masks: 0.745
#
# After fix (worker_init_fn seeded with hvd.rank()*10000+worker_id):
#   Batch 0: mask agreement rank0 vs rank1 = 0.7441  ← near-theoretical ✓
#   Batch 1: mask agreement rank0 vs rank1 = 0.7489  ✓

# ---- Training results ----
# Baseline (1 GPU, 20 epochs, batch_size=16):
#   val F1: 0.782 ← reference

# Horovod 4-GPU, worker_init_fn=None (buggy — identical augmentation):
#   Epoch  5: val F1 0.721
#   Epoch 10: val F1 0.763
#   Epoch 15: val F1 0.779
#   Epoch 20: val F1 0.782  ← identical to 1-GPU baseline
#   Throughput: 3,840 samples/sec  (4× speedup ✓)
#   Effective data diversity: equivalent to 1-GPU run ✗

# Horovod 4-GPU, worker_init_fn seeded per rank (fixed):
#   Epoch  5: val F1 0.738  ← faster early convergence ✓
#   Epoch 10: val F1 0.801
#   Epoch 15: val F1 0.831
#   Epoch 20: val F1 0.840  ← 6% above 1-GPU baseline ✓
#   Throughput: 3,840 samples/sec  (unchanged — fix is in CPU workers, not GPU)

# ---- Per-rank gradient cosine similarity ----
# Measure whether gradients differ across ranks after the fix
# (sampled from layer model.bert.encoder.layer[11].attention.self.query.weight)
# Before fix:
#   cos_sim(rank0_grad, rank1_grad) = 0.9981  ← near-identical gradients ✗
#   cos_sim(rank0_grad, rank2_grad) = 0.9979
# After fix:
#   cos_sim(rank0_grad, rank1_grad) = 0.4231  ← diverse gradients ✓
#   cos_sim(rank0_grad, rank2_grad) = 0.3917

# ---- Hour breakdown ----
# Triage:    0.5h — val F1 0.782 same as 1-GPU; throughput 4× correct;
#            loss curves normal; ran hvd.allgather on augmentation masks;
#            agreement rank0 vs rank1: 1.000 (should be ~0.745);
#            random.seed(42) before hvd.init() identified as cause
# Root cause: 0.25h — fork() copies parent random state to all workers;
#             all 16 workers have identical seed; augmentation masks identical
# Fix:       0.25h — added worker_init_fn with hvd.rank()*10000+worker_id seed
# Validation: 0.5h — 20 epochs; val F1 0.782→0.840; augmentation diversity confirmed
# Total:     1.5h

Related retainer posts

Track retainer hours without a client portal

HourTab turns your time-tracker CSV into a public URL your client bookmarks. No login. No portal. CSV in, URL out.

See pricing →