AI System Scalability Questions
Covers designing and operating machine learning systems to handle growth in data volume, model complexity, and traffic. Topics include distributed training strategies such as data parallelism, model parallelism, and pipeline parallelism; coordination and orchestration approaches like parameter servers, gradient aggregation, and framework tools such as PyTorch distributed, Horovod, and TensorFlow strategies; data pipeline and I O considerations including sharding, efficient formats, preprocessing bottlenecks, streaming and batch ingestion; serving and inference scaling including model sharding, batching for throughput, autoscaling, request routing, caching, and latency versus throughput tradeoffs. Also includes monitoring, profiling, checkpointing and recovery, reproducibility, cost and resource optimization, and common bottleneck analysis across network, storage, CPU preprocessing, and accelerator utilization.
HardTechnical
50 practiced
Design a checkpoint format and protocol in pseudocode for sharded optimizers where each worker owns a disjoint slice of optimizer state and parameters. Include steps for saving consistent global checkpoints, metadata to locate slices, and how to restore to a different cluster size (elastic restore).
Sample Answer
Approach: Use a shard-centric checkpoint format where each worker writes its local parameter and optimizer-state slice to object store as an immutable file plus a global manifest (metadata) describing global layout, versioning, checksums, and mapping from global tensor ranges -> shard files. For consistency, coordinate via a short barrier/commit protocol: one coordinator (or leader via consensus) collects per-worker manifests and writes a consolidated manifest atomically. For elastic restore, manifest includes global shapes, shard-scheme, and provenance so a new cluster can remap slices (concatenate/redistribute) and perform lazy or eager shard re-sharding.Key points:- Immutable shard files + per-worker manifests make partial failures safe.- Atomic global_manifest + COMMITTED marker ensures consistent global checkpoint.- Manifest contains global tensor shapes/dtypes, shard ownership ranges, checksums, version/provenance to support backward compatibility.- Elastic restore remaps ranges; prefer range-reads from object store to avoid full downloads. Support lazy restore for very large models.Edge cases: overlapping/duplicate shards, missing shards, inconsistent manifests (reject commit), network failures during commit (use idempotent writes), format version changes (migrations). Alternatives: use Paxos/etcd for coordinator, CRDT-based manifests, or single monolithic archive (less scalable).
python
# Pseudocode for shard save/commit and elastic restore
# Local per-worker save (each worker i):
function WorkerSave(worker_id, epoch, local_params, local_opt_state, storage):
# compute per-tensor slices owned by this worker: list of (tensor_name, slice_range)
slices = compute_owned_slices(worker_id)
local_files = []
for (name, rng) in slices:
data = serialize(local_params[name][rng], local_opt_state[name][rng])
fname = f"{model_name}/epoch{epoch}/shard_w{worker_id}/{name}.{rng}.bin"
checksum = storage.put(fname, data) # upload immutable file
local_files.append({ "name": name, "range": rng, "file": fname, "checksum": checksum, "bytes": len(data)})
# local manifest: what this worker saved
local_manifest = {
"worker_id": worker_id,
"epoch": epoch,
"timestamp": now(),
"files": local_files,
"local_version": compute_version(local_params, local_opt_state)
}
storage.put(f"{model_name}/epoch{epoch}/shard_w{worker_id}/manifest.json", json.dumps(local_manifest))
# optionally notify coordinator (RPC or pubsub)
coordinator.notify_worker_manifest(worker_id, local_manifest)
# Coordinator commit (collect and write global manifest)
function CoordinatorCommit(epoch, expected_worker_count, storage, timeout=30s):
manifests = wait_for_manifests(expected_worker_count, timeout)
# validate no overlapping ownership, full coverage of global tensors
assert validate_coverage(manifests)
global_manifest = {
"model_name": model_name,
"epoch": epoch,
"version": format_version(),
"workers": [m.worker_id for m in manifests],
"shards": concat(m.files for m in manifests),
"global_tensors": list_global_tensor_shapes_and_dtypes(),
"timestamp": now(),
"checksums_ok": all(storage.verify(f.file, f.checksum) for f in shards)
}
# write atomically (write to temp then rename)
tmp = f"{model_name}/epoch{epoch}/global_manifest.json.tmp"
storage.put(tmp, json.dumps(global_manifest))
storage.rename(tmp, f"{model_name}/epoch{epoch}/global_manifest.json")
# publish commit marker
storage.put(f"{model_name}/epoch{epoch}/COMMITTED", b"1")
return global_manifest
# Elastic restore on new cluster with N' workers
function ElasticRestore(target_epoch, new_num_workers, storage):
manifest = storage.get_json(f"{model_name}/epoch{target_epoch}/global_manifest.json")
assert storage.exists(f"{model_name}/epoch{target_epoch}/COMMITTED")
# compute mapping from old shard ranges -> new worker assignments
global_tensors = manifest.global_tensors
shards = manifest.shards
mapping = remap_shards_to_new_workers(shards, new_num_workers, global_tensors)
# two modes: eager (repartition and materialize) or lazy (stream and assemble on-the-fly)
for new_worker_id in range(new_num_workers):
assigned_segments = mapping[new_worker_id] # list of (file, byte_range, target_range)
# fetch segments and reconstruct local params+opt state
local_state = {}
for seg in assigned_segments:
data = storage.get_range(seg.file, seg.byte_range) # range-read if supported
tensor_part = deserialize(data, seg.target_range)
merge_into(local_state, seg.tensor_name, tensor_part)
# optionally write new local shard files for faster future loads
write_local_shard_cache(new_worker_id, target_epoch, local_state)
return success
# Helper: remap_shards_to_new_workers
function remap_shards_to_new_workers(old_shards, new_num):
# For each global tensor, treat as contiguous linear index; compute new ranges by split
mapping = dict(new_worker_id -> [])
for tensor in global_tensors:
total_len = tensor.shape.num_elements()
for new_worker_id in range(new_num):
new_range = compute_range_for_worker(new_worker_id, new_num, total_len)
# find which old shard files overlap and emit (file, overlap_byte_range, target_range)
overlaps = find_overlaps(old_shards_for_tensor(tensor.name), new_range)
mapping[new_worker_id].extend(overlaps)
return mappingMediumTechnical
48 practiced
Compare Horovod and PyTorch Distributed Data Parallel (DDP). Focus on setup complexity, backend communication (NCCL/gloo), interoperability across frameworks, and their typical performance characteristics at scale.
Sample Answer
High-level summary: Horovod is a framework-agnostic distributed training library that uses allreduce primitives (MPI/gloo/NCCL) to synchronize gradients across workers. PyTorch DDP (DistributedDataParallel) is PyTorch-native, using torch.distributed with backends like NCCL or gloo and optimized for the single-process-per-GPU pattern.Setup complexity- Horovod: simpler if you need one API across TF/PyTorch/MXNet. Requires installing horovod (and optionally MPI), launching with mpirun/horovodrun/torchrun wrapper; tuning fusion buffer sizes may be needed. Multi-framework convenience reduces engineering work when supporting many frameworks.- DDP: Minimal if you already use PyTorch. Use torch.distributed.launch/torchrun or cluster schedulers. Best practice is one process per GPU (straightforward) but requires adapting code to use DistributedSampler, init_process_group, and proper checkpointing.Backend communication- NCCL: Preferred for GPU multi-node multi-GPU (high throughput, topology-aware, peer-to-peer, GPU-direct). Both Horovod and DDP can use NCCL.- gloo: CPU and fallback on smaller setups; used when NCCL unavailable.- MPI: Horovod can leverage MPI runtimes (mpirun) for process management and transport; DDP typically does not require MPI.Interoperability- Horovod: Framework-agnostic — same API pattern across TensorFlow/PyTorch/MXNet. Easier when standardizing pipelines across frameworks.- DDP: PyTorch-only but deep integration (autograd hooks, bucketization, native checkpointing). If you’re purely PyTorch, DDP integrates more naturally.Typical performance at scale- DDP (single-process-per-GPU) is often the fastest and most memory-efficient for PyTorch because it hooks directly into autograd, minimizes copies, and overlaps compute with NCCL allreduce. It scales predictably; NCCL collective optimizations and per-bucket tuning help.- Horovod performs very well, particularly when using NCCL and MPI; its ring/allreduce implementations and gradient fusion can be competitive. On heterogeneous or multi-framework clusters, Horovod simplifies ops while delivering near-DDP performance.- At very large scales, network topology, NCCL version, PCIe/NIC configuration, and surface area for overlapping compute/comm dominate. Both can be tuned (bucket sizes, gradient fusion, async options), but native DDP + torch.distributed + NCCL tends to have a slight edge for pure PyTorch workloads.Other operational notes- Fault tolerance: torch.elastic with DDP gives robust reconfiguration; Horovod has options but depends on MPI/launcher behavior.- Debugging and profiling: DDP benefits from PyTorch tooling (torch.profiler); Horovod integrates with framework profilers but can add launcher complexity.Recommendation- If you’re exclusively on PyTorch and want best-in-class integration and performance: DDP (single-process-per-GPU) + NCCL.- If you need a single distributed API across multiple frameworks or legacy MPI-based clusters: Horovod with NCCL/MPI is a strong, pragmatic choice.
HardTechnical
36 practiced
Discuss reproducibility challenges for distributed training at scale: nondeterministic ops (cuDNN), race conditions, network reordering, and asynchronous updates. Propose engineering patterns and system-level solutions to achieve bitwise reproducibility or 'statistically equivalent' runs, and analyze their trade-offs in performance and debuggability.
Sample Answer
Challenges:- Non-deterministic ops (cuDNN): algorithm selection, atomic reductions, and kernel scheduling produce different floating-point rounding and order of operations across runs/GPU drivers.- Race conditions & asynchronous updates: lock-free accumulators (e.g., Hogwild, async SGD) and parameter-server staleness cause different update orders and thus divergent trajectories.- Network reordering: packets arriving in different orders (especially with RDMA, multi-path) change timing of collective ops and asynchronous gradients.- Hardware/software stack variability: GPU clocks, NCCL collective algorithms, driver versions, and environment variables alter execution.Engineering patterns and system-level solutions:1. Enforce deterministic kernels - Pin deterministic cuDNN algorithms (e.g., set cudnn.deterministic=True) and disable autotuner. - Use deterministic primitives in framework (torch.use_deterministic_algorithms). Trade-off: sometimes slower or less memory-efficient; not all ops have deterministic variants.2. Control randomness sources - Seed all RNGs (OS, numpy, framework, cuRAND) and record seeds. - Use counter-based RNGs (Philox) for parallel reproducibility. Trade-off: reduces randomness for augmentation; careful sampling needed to avoid correlation.3. Synchronized, ordered updates - Use synchronous SGD with deterministic AllReduce (NCCL with configured algorithms or custom MPI) to fix gradient aggregation order. - For async systems, implement deterministic sequencing (logical clocks, total order using sequencer). Trade-off: sync hurts throughput/latency and can increase straggler sensitivity.4. Deterministic scheduling & isolation - Fix thread/core/GPU affinity, disable dynamic frequency scaling, isolate background processes. - Containerize and version-control drivers, libraries, and CUDA. Trade-off: operational complexity; reduced utilization.5. Reduce non-associativity effects - Use higher-precision accumulators (float32->float64) during accumulation or Kahan summation to make reductions less sensitive to ordering. Trade-off: higher memory/compute cost; may change training dynamics.6. Provide "statistical equivalence" mode - Accept bitwise non-reproducibility but test for distributional equivalence: compare loss/metric trajectories, seeds, and confidence intervals across runs; use seed ensembles. Trade-off: easier to scale and faster but harder to debug subtle bugs.7. Observability & reproducible debug traces - Log seeds, library versions, kernel selections, NCCL/MPI algorithms, and timestamps. - Capture and replay traces (deterministic replay systems) for failing runs. Trade-off: tracing overhead and storage.Analysis of trade-offs:- Bitwise reproducibility gives strongest debuggability (exact replay) but significantly constrains optimizer choices, hardware performance, and distributed throughput. Many modern stacks lack fully deterministic primitives, so achieving bitwise determinism often requires sacrificing speed, memory, and algorithmic advances.- Statistical equivalence accepts nondeterminism but requires robust metrics, larger experimental budgets, and probabilistic testing. It's more practical at scale: allows asynchronous and optimized kernels while ensuring models are functionally equivalent for deployment.Recommendation:- For development/debugging: enable deterministic mode with small-scale runs and rich tracing.- For large-scale production training: aim for statistical equivalence with thorough logging, seeded pipelines, reproducible data sharding, and periodic deterministic checkpoints/replays for critical failures.
MediumTechnical
33 practiced
Explain pipeline parallelism for model training. Describe micro-batching, pipeline stages, the bubble/idle overhead, and how schedules can be chosen to trade latency for throughput. Mention how pipeline interacts with data parallelism.
Sample Answer
Pipeline parallelism shards a model across multiple devices: each device (stage) holds a contiguous subset of layers and processes activations flowing through the pipeline. This lets very large models fit memory by splitting model state instead of data.Micro-batching: the training batch is split into smaller micro-batches which are fed sequentially into the pipeline. While stage 1 works on micro-batch 2, stage 2 can process micro-batch 1 — enabling concurrent device utilization.Bubble / idle overhead: when starting or finishing an iteration there are empty pipeline slots (bubbles) because downstream stages wait for earlier micro-batches (and vice versa on flush). For S stages and M micro-batches the ideal steady-state throughput starts after ~S−1 micro-batches (startup bubbles) and ends with shutdown bubbles, so small M increases relative overhead.Schedules & trade-offs: with few micro-batches you minimize end-to-end latency (faster single-batch forward/backward) but incur high bubble overhead (lower throughput). Increasing micro-batches improves throughput (amortizes bubbles) at cost of higher end-to-end latency and extra activation memory. Scheduling variants: naive lockstep, GPipe-style synchronous pipeline (simple, good throughput), and fine-grained schedules like PipeDream or 1F1B (one-forward-one-backward) that interleave forward/backward to reduce memory and stall time. Choice depends on whether you prioritize throughput (large M, aggressive interleaving) or low latency (small M, fewer concurrent micro-batches).Interaction with data parallelism: pipeline is orthogonal to data parallelism; you can replicate the whole pipeline across workers (model-parallel within replica, data-parallel across replicas). Combining both lets you scale to more devices: use pipeline to shard model and data-parallel to increase batch-parallel throughput and enable synchronous gradient averaging (e.g., all-reduce). Hybrid schemes must balance global batch size, learning-rate scaling, and communication costs (all-reduce after gradients aggregated across micro-batches).
HardTechnical
30 practiced
Given an AllReduce-based training cluster, design an experiment and list telemetry to determine whether performance is limited by network bandwidth/latency or by compute. Specify host-level, NIC-level, and GPU-level metrics to collect and how to interpret them.
Sample Answer
Experiment goal: determine whether AllReduce training is bottlenecked by network (bandwidth/latency) or by compute (GPU kernels/PCIe).Experiment design (steps):1. Run a controlled training job (fixed batch, model, data) across N nodes.2. Baseline: single-node multi-GPU training to measure pure compute limits.3. Multi-node AllReduce run while varying: a) message size (simulate gradients aggregation frequency), b) network link speed (if possible, throttle), c) number of GPUs per node.4. Collect synchronized telemetry during each run and compare to baseline.Telemetry to collect and how to interpret:Host-level- CPU utilization (per-core), load average: high CPU (near 100%) during communication indicates host-side software/serialization bottleneck.- Memory bandwidth and usage: saturation suggests host-side copy limits.- Context switches/interrupts: excessive values → OS overhead affecting network/GPU drivers.Interpretation: If host CPU/memory saturated while GPUs idle, host-bound.NIC-level- Link utilization (Gbps out/in) and packet rate (pps): sustained near link capacity → bandwidth-bound.- NIC transmit/receive queues, drops, retransmits, errors: drops/retransmits → effective throughput reduced; high latency.- RDMA metrics (if using RoCE/Infiniband): QP queue depths, S/W completions/sec, latency percentiles (p50/p95/p99).- NIC CPU interrupts or MSI-X steering: imbalanced interrupts → processing hotspot.Interpretation: If link utilization ~max and network latency/p99 high during AllReduce, network-bound. If high latency but low bandwidth, look at latency/packet issues.GPU-level- GPU utilization (SM utilization), GPU memory bandwidth, GPU power and temperature.- PCIe/NVLink throughput and PCIe transfer stalls.- Per-kernel timelines (start/end), kernel durations, and overlaps with NCCL communication events (NCCL traces).- GPU idle time and queue depths (CUDA streams).Interpretation: If GPUs show high SM utilization and kernels continuously executing with little idle time despite low network usage, compute-bound. If GPUs idle waiting for NCCL or data transfers and NCCL timelines show long communication phases, network-bound or PCIe-bound.Cross-layer signals and heuristics- Compare AllReduce time fraction: measure ratio of communication time vs computation time using distributed profiler (NCCL/allreduce profiling). If comms > 50% and network link near capacity → network-bound.- Scale test: increase number of nodes/GPU workers. If throughput per GPU drops and network metrics scale toward capacity, network-bound. If no change, compute-bound.- Artificially delay or throttle network: if training slows proportionally → network-bound.- Pinning test: run same workload with NCCL_SOCKET_IFNAME disabled to force intra-node only; if performance improves dramatically, inter-node network is culprit.Suggested thresholds/indicators- Link utilization > 80% sustained + comms dominating time → bandwidth limit.- p99 latency spikes during AllReduce + retransmits → latency-limited.- GPU SM utilization > 85% and PCIe transfers moderate → compute-limited.- High GPU idle time while NIC busy → network-bound.Final recommendation: use combined profiling (nvprof/nsight/NCCL trace + iperf/ibstat + sar/ethtool) to correlate timelines and make root-cause decision.
Unlock Full Question Bank
Get access to hundreds of AI System Scalability interview questions and detailed answers.
Sign in to ContinueJoin thousands of developers preparing for their dream job.