WorksheetsDistributed Systems
Total questions: 73
Worksheet time: 39mins
Primary motivation for parallel execution in ML training
To reduce model complexity
To speed up the model training process for large datasets
To minimize GPU memory usage
To simplify hyperparameter tuning
Training ResNet-50 on ImageNet-1K, single GPU
Minutes to hours
Hours to one day
Several days to two weeks
Over a month
Most popular form of in-parallel training
Model Parallelism
Pipeline Parallelism
Data Parallelism
Hybrid Parallelism
Key characteristic ImageNet vs CIFAR-10
Larger dataset size
More classes
Higher image resolution
Greater label complexity
Primary goal of parallel execution
To enable larger models
To reduce data storage costs
To speed up the training process
To improve model accuracy
Fundamental cause of long training times on single node
Limited CPU cores
GPU clock speed
Mismatch between data loading bandwidth and training bandwidth
Small batch sizes
Effect of higher resolution (ImageNet-1K)
Faster convergence
Larger activations → more memory → fewer images per batch
Reduced overfitting
Higher gradient precision
Why single-node training is "too slow"
CPU overheating
Bandwidth mismatch
Insufficient RAM
Software inefficiencies
Direct consequence of high-resolution activations
Increased model accuracy
Fewer images can be trained at once
Faster backward passes
Reduced need for synchronization
Effect of smaller activations (CIFAR-10)
Lower model capacity
Allows more input images at once → higher training bandwidth
Increased regularization
Slower convergence
Why perfect bandwidth matching removes need for parallelism
Eliminates data partitioning
Because distributed training adds overhead, which would dominate
Reduces model complexity
Enables larger batch sizes
How data parallelism addresses mismatch
By compressing data
By reducing model size
By splitting data loading across multiple accelerators
By using mixed precision
Factor besides number of images (ImageNet vs CIFAR)
Label diversity
Higher image resolution
Class imbalance
Augmentation complexity
GPT-3 288 years example illustrates
High hardware costs
Extreme necessity of distributed training for large models
Inefficiency of transformers
Need for model pruning
Why distributed training despite overheads
Speedup outweighs communication overhead
Simplifies debugging
Reduces energy consumption
Improves model generality
In data parallelism, each GPU/worker holds:
A shard of the model and full data
A full copy of the model and a partition of the data
Full data and no model
A partition of both model and data
Purpose of model synchronization:
Reduce memory usage
Ensure all workers' models share the same parameters after each iteration
Compress gradients
Balance workload
Why traditional Gradient Descent (GD) doesn't work in parallelism:
Requires excessive memory
Requires gradients over the full dataset, which no worker has
Too slow
Incompatible with GPUs
How input data is handled across workers:
Replicated identically
Partitioned into disjoint subsets
Randomly sampled
Augmented differently
Term for workers' models diverging:
Gradient explosion
Parameter drift
Model divergence
Synchronization failure
At start of training, each worker:
Has random model subsets
Holds a partitioned model
Holds a full copy of the model, initialized identically
Has no model
Why SGD is necessary in parallelism:
Low memory footprint
Allows updates using mini-batches (local data subsets)
Faster convergence
Better regularization
Two main actions in synchronization:
Aggregate losses & broadcast models
Shuffle data & reduce parameters
Collect gradients from workers & redistribute the aggregated result
Compress activations & synchronize caches
Key difference between training & inference:
Data augmentation
Synchronization is removed during inference
Batch normalization
Loss computation
Purpose of broadcasting aggregated gradients:
Reduce communication costs
Initialize next batch
Ensure all workers update models with identical gradients
Enable asynchronous updates
Why GD cannot be applied:
High variance
Slow convergence
No worker has full dataset for complete gradient computation
Requires double precision
Without synchronization:
Models converge faster
Models diverge → multiple inconsistent models
Gradients vanish
Data partitions overlap
Data parallelism at inference stage:
Increases accuracy
Synchronization phase is removed
Requires larger batches
Slows down prediction
Goal of synchronization (collect + distribute):
Minimize epoch count
Balance data partitions
Keep all workers' parameters identical
Reduce gradient variance
Why parameters diverge after iteration (w/o sync):
Random initialization
Each worker trained on different subsets
Learning rate differences
Hardware variability
Standard practice for initialization:
Different seeds per worker
No initialization
Fixed random seed to align all workers
Layer-wise initialization
Meaning of "on the same page":
Same data batch
Identical hardware
All GPUs maintain identical parameter values post-update
Synchronized clocks
Two primary roles in Parameter Server architecture:
Coordinator & Executor
Trainer & Validator
Parameter Server and Worker
Aggregator & Distributor
First step a worker takes in training iteration:
Compute gradients
Pull latest weights from the server
Load local data
Push gradients
Where final parameter update occurs:
On each worker
On the data shard
On the central parameter server(s)
During synchronization
Component aggregating model updates:
Worker pool
Parameter server
Gradient cache
Synchronization layer
Parameter Server architecture is:
Decentralized
Federated
A centralized model
Peer-to-peer
Primary worker responsibility:
Store global model
Compute gradients on local data partition
Update parameters
Aggregate gradients
Communication pattern server → workers (weights):
Fan-in
Peer-to-peer
Fan-out communication
Broadcast-reduce
Cause of bottleneck at "Push Gradients" stage:
Worker CPU limits
Server must receive gradients from all workers, dividing bandwidth
Network latency
Gradient size
Sharding in Parameter Server context:
Splitting data across workers
Splitting model parameters across multiple servers
Partitioning gradients
Dividing epochs
Communication pattern when all workers send gradients to server:
Scatter
All-gather
Fan-in
Reduce
Negative consequence of more workers (fixed cluster):
Higher per-iteration time
Increased accuracy
Less data processed per iteration
Faster convergence
Negative consequence of more parameter servers (fixed cluster):
Higher memory usage
Reduced model capacity
Fewer nodes compute gradients → lower throughput
Increased latency
Trade-off when choosing #servers vs #workers:
Accuracy vs speed
Communication bandwidth (servers) vs training throughput (workers)
Memory vs storage
Synchronization vs asynchrony
Shortcoming of Parameter Server design:
Low scalability
Poor fault tolerance
High coding complexity (explicit roles & protocols)
Slow convergence
In single server, N workers: why each worker only gets 1/N bandwidth?
Worker GPU limits
Server's bandwidth divided among N workers
Network switches
Gradient aggregation overhead
How sharding solves fan-out bottleneck:
Reduces gradient size
Each worker pulls from N servers concurrently, saturating its link
Compresses parameters
Uses lossy compression
Reason for decline of Parameter Server:
Low accuracy
Difficult to determine optimal server/worker ratio
High cost
Slow inference
How All-Reduce differs from Parameter Server in node roles:
Uses dedicated aggregators
It abandons the parameter server role, making all nodes equivalent workers
Requires more memory
Centralizes gradients
Collective communication primitive for synchronization:
Broadcast
Scatter
All-Reduce
Gather
Purpose of Broadcast collective operation:
Aggregate data
To distribute the same data from one node to all others
Shard parameters
Reduce memory
All-Reduce architecture paradigm:
Centralized
Hierarchical
Decentralized
Client-server
Primary advantage over server/worker ratio problem:
Lower memory
Eliminates the ratio issue since all nodes are workers
Faster data loading
Simpler initialization
Communication pattern of All-Reduce primitive:
One-to-all
All-to-one
All-to-all
Point-to-point
Global batch size in data parallel training:
Per-GPU batch size
Total number of training samples across all GPUs in one iteration
Number of epochs
Gradient accumulation steps
Rule of thumb for learning rate scaling:
Divide learning rate by N
Use constant learning rate
Multiply learning rate by N (number of GPUs)
Square root of N scaling
Term for batch size in single-node training (analogous to global batch):
Micro-batch
Gradient batch
Batch size or mini-batch
Epoch size
Why excessively large global batch size is detrimental:
Increases communication
It can harm model convergence
Causes divergence
Requires more servers
Architectural difference Parameter Server vs All-Reduce:
Data partitioning
Gradient aggregation
Parameter Server = centralized with distinct roles; All-Reduce = decentralized homogeneous roles
Learning rate scaling
Negative outcome of too small global batch size:
Slower convergence
Higher variance
Wastes distributed computational resources
Memory overflow
Why All-Reduce has lower coding complexity:
No gradients needed
Communication is abstracted away by libraries (vs explicit coding in Parameter Server)
Automatic batch sizing
Simplified initialization
Contrast of architectures (Parameter Server vs All-Reduce):
PS uses smaller batches
Parameter Server = centralized, All-Reduce = decentralized
PS faster for small clusters
All-Reduce requires more memory
Which describes location transparency in distributed systems?
Physical location is visible to users
The location of an object is hidden from the user
Objects are immovable
Location defines access rights
Which is NOT a goal of distributed systems design?
Scalability
Fault tolerance
Centralization
Resource sharing
Primary challenge of replication in distributed systems?
Storage cost
Maintaining global synchronization across replicas
Network latency
Data encoding
Why administrative scalability challenging?
Conflicting policies regarding usage, management, and security across domains
Limited IP addresses
Hardware heterogeneity
Software licenses
Example of a shared resource in a distributed system?
Local CPU cache
Private SSD
Cloud-based shared storage
GPU memory
Transparency hiding object movement during use?
Access transparency
Relocation transparency
Migration transparency
Concurrency transparency
"Layered architecture" refers to:
Physical network topology
Organization of components into hierarchical levels for specific functionalities
Data replication tiers
Security clearance levels
Main difference stateless vs stateful servers?
Stateless use UDP
Stateful servers keep track of client status, while stateless do not
Stateful are faster
Stateless scale better
Primary benefit of stateful servers?
Lower bandwidth
High performance through caching and prefetching
Easier debugging
Simpler load balancing
How sharding mitigates bottlenecks.
