PyTorch Distributed: Experiences on Accelerating
Data Parallel Training
Shen Li† Yanli Zhao† Rohan Varma† Omkar Salpekar†
Pieter Noordhuis∗ Teng Li† Adam Paszke‡†
Jeff Smith Brian Vaughan† Pritam Damania† Soumith Chintala†{shenli, yanlizhao, rvarm1, osalpekar}@fb.com,
pcnoordhuis@gmail.com, tengli@fb.com, adam.paszke@gmail.com,{jeffksmith, bvaughan, pritam.damania, soumith}@fb.com†

Facebook AI‡

University of Warsaw

ABSTRACT

1.

This paper presents the design, implementation, and evaluation of the PyTorch distributed data parallel module. PyTorch is a widely-adopted scientific computing package usedin deep learning research and applications. Recent advances
in deep learning argue for the value of large datasets andlarge models, which necessitates the ability to scale out
model training to more computational resources. Data parallelism has emerged as a popular solution for distributed
training thanks to its straightforward principle and broad
applicability. In general, the technique of distributed data
parallelism replicates the model on every computational resource to generate gradients independently and then communicates those gradients at each iteration to keep model
replicas consistent. Despite the conceptual simplicity ofthe technique, the subtle dependencies between computation and communication make it non-trivial to optimize the
distributed training efficiency. As of v1.5, PyTorch natively
provides several techniques to accelerate distributed data
parallel, including bucketing gradients, overlapping computation with communication, and skipping gradient synchronization. Evaluations show that, when configured appropriately, the PyTorch distributed data parallel module attains
near-linear scalability using 256 GPUs.

Deep Neural Networks (DNN) have powered a wide spectrum of applications, ranging from image recognition [20],language translation [15], anomaly detection [16], content
recommendation [38], to drug discovery [33], art generation [28], game play [18], and self-driving cars [13]. Many
applications pursue higher intelligence by optimizing larger
models using larger datasets, craving advances in distributed
training systems. Among existing solutions, distributed data
parallel is a dominant strategy due to its minimally intrusive nature. This paper presents the design, implementation, and evaluation of the distributed data parallel package
in PyTorch v1.5 [30].
Training a DNN model usually repeatedly conducts threesteps [26], the forward pass to compute loss, the backward
pass to compute gradients, and the optimizer step to updateparameters. The concept of data parallelism is universally
applicable to such frameworks. Applications can create multiple replicas of a model, with each model replica working on
a portion of training data and performing the forward andbackward passes independently. After that, model replicas
can synchronize either their gradients or updated parameters depending on the algorithm. It’s nominally possible tobuild a working version of data parallel purely on the application side, as it only requires inserting appropriate communications into every iteration. However, squeezing out
the last bit of performance takes an enormous amount of effort in design and tuning. Providing native distributed data
parallel APIs on the platform side would help application
developers focus on optimizing their models, while the platform developing team could continuously and transparently
improve the training speed. To provide a general distributed
data parallel package, the challenges are three-fold.

PVLDB Reference Format:
Shen Li, Yanli Zhao, Rohan Varma, Omkar Salpekar, Pieter Noordhuis, Teng Li, Adam Paszke, Jeff Smith, Brian Vaughan, Pritam Damania, Soumith Chintala. PyTorch Distributed: Experiences on Accelerating Data Parallel Training. PVLDB, 13(12):3005-3018, 2020.
DOI: https://doi.org/10.14778/3415478.3415530∗
This work was conducted when Pieter Noordhuis was an
employee at Facebook.

INTRODUCTION• Mathematical equivalence: The purpose of data
parallel is to speed up training on large datasets. Applications expect to harvest the same result model as if
all training had been performed locally without model
replication. This requires mathematical equivalence tolocal training despite its distributed nature.

This work is licensed under the Creative Commons AttributionNonCommercial-NoDerivatives 4.0 International License. To view a copyof this license, visit http://creativecommons.org/licenses/by-nc-nd/4.0/. For
any use beyond those covered by this license, obtain permission by emailing
info@vldb.org. Copyright is held by the owner/author(s). Publication rights
licensed to the VLDB Endowment.
Proceedings of the VLDB Endowment, Vol. 13, No. 12
ISSN 2150-8097.
DOI: https://doi.org/10.14778/3415478.3415530• Non-intrusive and interceptive API: Application
developments usually start from local models and thenscale out when necessary. To avoid the exorbitant3005

hurdles during the transition, the API must be nonintrusive in application code. On the other hand, theAPI needs to allow the internal implementation totimely intercept signals to carry out communications
and system optimizations.

2.

• High performance: Data parallel training is subject to subtle dependencies between computations andcommunications. The design and implementation haveto explore the solution space to efficiently convert moreresources into higher training throughput.

2.1

PyTorch provides distributed data parallel as an nn.Moduleclass, where applications provide their model at construction
time as a sub-module. To guarantee mathematical equivalence, all replicas start from the same initial values for model
parameters and synchronize gradients to keep parametersconsistent across training iterations. To minimize the intrusiveness, the implementation exposes the same forward [7]API as the user model, allowing applications to seamlessly
replace subsequent occurrences of a user model with the distributed data parallel model object with no additional codechanges. Several techniques are integrated into the design todeliver high-performance training, including bucketing gradients, overlapping communication with computation, andskipping synchronization.
Evaluations were conducted on an exclusive 32-GPU cluster and on 256 GPUs from a much larger shared entitlement.
We developed benchmarks to evaluate the distributed package across different scales to present an in-depth view ofthe performance implications of different optimization techniques and configurations. Experiments also cover the comparison between NCCL and Gloo communication libraries.
The results show that 1) communication is the dominant
training latency contributor, and its impact increases withmodel sizes; 2) bucket sizes considerably affect communication efficiency, which could lead to more than 2X speedup if
configured properly; 3) skipping synchronizations appropriately would significantly reduce amortized communication
overhead without noticeably degrading convergence speed.
Techniques described in this paper were first released in
PyTorch v1.1. During the past year, we have seen significant
adoption both internally and externally. Within Facebook,a workload study from 05/11/20 to 06/05/20 shows that
more than 60% of production GPU hours during that period
were spent on the PyTorch distributed data parallel package across a wide variety of applications, including speech,
vision, mobile vision, translation, etc. There are three main
contributions in this paper. First, this paper reveals thedesign and implementation of a widely adopted industrial
state-of-the-art distributed training solution. Second, thispaper highlights real-world caveats (e.g., due to pluralizedgraphs) that were overlooked by prior work. Third, we shareperformance tuning experiences collected from serving internal teams and open-source community users and summarized several directions for future improvements.
The remainder of the paper is organized as follows. Section 2 briefly introduces PyTorch and data parallelism. Section 3 elaborates the design for the PyTorch distributed data
parallel module. Implementations and evaluations are presented in Section 4 and Section 5 respectively. Then, Section 6 discusses lessons learned and opportunities for future
improvements, and Section 7 surveys related work. Finally,
Section 8 concludes the paper.

BACKGROUND

Before diving into distributed training, let us briefly discuss the implementation and execution of local model training using PyTorch. Then, we explain and justify the idea ofdata parallelism and describe communication primitives.

PyTorchPyTorch organizes values into Tensors which are genericn-dimensional arrays with a rich set of data manipulating
operations. A Module defines a transform from input values to output values, and its behavior during the forward
pass is specified by its forward member function. A Module
can contain Tensors as parameters. For example, a Linear
Module contains a weight parameter and a bias parameter,whose forward function generates the output by multiplying
the input with the weight and adding the bias. An application composes its own Module by stitching together native
Modules (e.g., linear, convolution, etc.) and Functions (e.g.,
relu, pool, etc.) in the custom forward function. A typical training iteration contains a forward pass to generate
losses using inputs and labels, a backward pass to compute
gradients for parameters, and an optimizer step to updateparameters using gradients. More specifically, during theforward pass, PyTorch builds an autograd graph to record
actions performed. Then, in the backward pass, it uses the
autograd graph to conduct backpropagation to generate gradients. Finally, the optimizer applies the gradients to updateparameters. The training process repeats these three steps
until the model converges.

2.2

Data Parallelism

PyTorch offers several tools to facilitate distributed training, including DataParallel for single-process multi-threaddata parallel training using multiple GPUs on the samemachine, DistributedDataParallel for multi-process data
parallel training across GPUs and machines, and RPC [6]for general distributed model parallel training (e.g., parameter server [27]). The remainder of this paper focuses onDistributedDataParallel. Data parallelism enables distributed training by communicating gradients before the optimizer step to make sure that parameters of all model replicas are updated using exactly the same set of gradients, andhence model replicas can stay consistent across iterations.
Parameter averaging is another popular technique to scaleout model training. Similarly, it can launch multiple processes across multiple machines, but instead of synchronizing gradients, parameter averaging directly computes theaverage of all model parameters. This occurs after the local optimizer step, meaning that parameter averaging canbe implemented completely as an auxiliary step and does
not need to interact with local training steps at all, which is
attractive as it can easily and cleanly decouple the code of
distributed training and local iterations. There are several
caveats with parameter averaging.
• Parameter averaging can produce vastly different results compared to local training, which, sometimes,
can be detrimental to model accuracy. The root cause
is that parameter averaging is not mathematically equivalent to processing all input data locally, especiallywhen the optimizer relies on past local gradients values (e.g., momentum). As different model replicas arelikely to see different gradients, the states in optimizers can gradually diverge, causing conflicting gradient3006

descent directions. This can result in inexplicable differences in performance when switching from locally
optimized models to large scale deployed models.

DistributedDataParallel
Python API• The structure of parameter averaging orchestrates computation (i.e., backward pass) and communication (i.e.,
computing average) into non-overlapping phases, using
optimizer step() functions as a hard separation point.
Regardless of how vigorously we optimize the computation or communication, one type of resource will stayidle at any given time instance, giving up a substantial
performance optimization opportunity.
Given the above fundamental pitfalls, we decided to implement distributed training using data parallelism to synchronize gradients instead of parameters. Note that, applications can still easily build parameter averaging using
PyTorch. In fact, the collective communication feature described in Section 3.3 is an appropriate solution for this use
case. Applications just need to explicitly launch AllReduce
operations to calculate averaged parameters accordingly.

2.3

Gradient Reduction
Collective Communication
NCCL

3.1

SYSTEM DESIGN

1
For simplicity, the rest of the paper uses the acronym DDP
to represent DistributedDataParallel henceforth.
2
For optimizers with intrinsic randomness, different processes can initialize their states using the same random seed.

API

When designing the API, we have defined two design goals
to achieve the necessary functionality.

AllReduce

PyTorch [30] provides a DistributedDataParallel (DDP1 )module to help easily parallelize training across multiple processes and machines. During distributed training, each process has its own local model replica and local optimizer.
In terms of correctness, distributed data parallel training
and local training must be mathematically equivalent. DDP
guarantees the correctness by making sure that all model
replicas start from the exact same model state, and see
the same parameter gradients after every backward pass.
Therefore, even though optimizers from different processes
are all independent, they should be able to bring their local
model replicas to the same state at the end of every iteration2 . Fig. 1 illustrates building blocks of DDP, which contains a Python API frontend, C++ gradient reduction corealgorithm, and employs the c10d collective communication

MPI

Figure 1: Building Blocks• Non-intrusive: The API must be non-intrusive toapplications. Application developers usually start fromwriting local training scripts, and scale out when hitting the resource limit on a single machine. At thatpoint, it is unacceptable to ask developers to rewrite
the entire application to enable distributed data parallel training. Instead, the developer should be able toreuse the local training script with minimal modifications.

AllReduce is the primitive communication API used byDistributedDataParallel to compute gradient summation
across all processes. It is supported by multiple communication libraries, including NCCL [2], Gloo [1], and MPI [4].
The AllReduce operation expects each participating process to provide an equally-sized tensor, collectively applies
a given arithmetic operation (e.g., sum, prod, min, max) to
input tensors from all processes, and returns the same result tensor to each participant. A naive implementation
could simply let every process broadcast its input tensor
to all peers and then apply the arithmetic operation independently. However, as AllReduce has significant impact on distributed training speed, communication libraries
have implemented more sophisticated and more efficient algorithms, such as ring-based AllReduce [2] and tree-based
AllReduce [23]. As one AllReduce operation cannot start
until all processes join, it is considered to be a synchronizedcommunication, as opposed to the P2P communication usedin parameter servers [27].

3.

Gloo

library. The following
sections are presented
in the top-down orderof this stack graph.
Section 3.1 presents
general principles thatdrive DDP API design. Section 3.2 explains gradient reduction techniques usedin the PyTorch distributed data parallel
package. Finally, Section 3.3 discusses thecollective communication backend options
for DDP.

• Interceptive: The API needs to allow the implementation to intercept various signals and trigger appropriate algorithms promptly. Distributed data parallel
aims at accelerating training by using more computational resources. This process requires subtle optimizations in both computations and communications
to achieve the best performance. Hence, the API must
expose as many optimization opportunities as possible
to the internal implementation.
Given the above requirements, we implemented distributed
data parallel as an nn.Module, which takes the local model asa constructor argument and transparently synchronizes gradients in the backward pass. The code snippet below shows
an example of using DDP module. This example uses an
nn.Linear layer to create a local model on line 10. Then, it
converts the local model into a distributed training model online 11 and sets up the optimizer on line 12. Line 14 through23 are typical forward pass, backward pass, and optimizer
step implementations. In this toy distributed training example, line 11 is the only difference that converts a local
training application into a distributed one, which satisfies
the non-intrusive requirement. It also fulfills the interceptive requirement. The constructor allows DDP to inspect themodel structure and parameters. After construction, the local model is replaced by the distributed one, which can theneasily intercept the forward() call to perform necessary actions accordingly. For the backward pass, DDP relies on backward hooks to trigger gradient reduction, which will be invoked by the autograd engine when executing backward()on the loss tensor.

3007

Total Gloo Execution Time (Sec)Total NCCL Execution Time (Sec)

10 1
10 2
10 3
10 4

0.25

100

10 1

10K
100K
1M
10M
Number of Parameters per AllReduce

1K

10K
100K
1M
10M
Number of Parameters per AllReduce(a) NCCL(b) GLOO

Measured Range
Median Time

Measured Range
Median TimeTime Elapsed in Backward on CPU (Sec)1KTime Elapsed in Backward on GPU (Sec)after the local backward pass and before updating local parameters. However, the API shown in Section 3.1 does notprovide an explicit entry point for this phase as there isnothing between backward() and step(). Fortunately, thePyTorch autograd engine accepts custom backward hooks.
DDP can register autograd hooks to trigger computation after
every backward pass. When fired, each hook scans throughall local model parameters, and retrieves the gradient tensor
from each parameter. Then, it uses the AllReduce collective communication call to calculate the average gradientson each parameter across all processes, and writes the result
back to the gradient tensor.
The naive solution works correctly, but there are two performance concerns.

101

100

0.20
0.15
0.10
0.05
0.00

6
5
4

• Collective communication performs poorly on small
tensors, which will be especially prominent on large
models with a massive number of small parameters.

3
2
1• Separating gradient computation and synchronization
forfeits the opportunity to overlap computation with
communication due to the hard boundary in between.

0
0

1

2
3
4
Number of Parameters

5

6

1e7

(c) GPU

0

1

2
3
4
Number of Parameters

5

6

1e7

(d) CPU

Figure 2: Communication vs Computation Delay

The following sections elucidates solutions to address theabove two concerns.

3.2.2
1
2
3
4

importimportimportimporttorch
torch . nn as nn
torch . nn . parallel as par
torch . optim as optim

5
6
7# initialize torch . distributed properly# with in i t_ p r o c e s s _ g r o u p

8
9
10
11
12

# setup model and optimizer
net = nn . Linear (10 , 10)
net = par .DistributedDataParallel( net )opt = optim . SGD ( net . parameters () , lr =0.01)

13
14
15
16
17

# run
inp =
exp =
out =

forward pass
torch . randn (20 , 10)
torch . randn (20 , 10)
net ( inp )

18
19
20

# run backward passnn . MSELoss () ( out , exp ) . backward ()

21
22
23

# update parameters
opt . step ()3.2

Gradient Reduction

The gradient reduction algorithm in DDP has evolved overthe past releases. To introduce the structure of the current
implementation, let us start from a naive solution, gradually
introduce more complexities, and land in the current version
as of today in PyTorch v1.5.0. This will also explain how
the same simple API described in Section 3.1 allows us toinstall various performance optimization algorithms.

3.2.1

A Naive Solution

As mentioned in the beginning of Section 3, DDP guarantees correctness by letting all training processes (1) startfrom the same model state and (2) consume the same gradients in every iteration. The former can be achieved bybroadcasting model states from one process to all others at
the construction time of DDP. To implement the latter, anaive solution can insert a gradient synchronization phase

Gradient Bucketing

The idea of gradient bucketing is motivated by the observation that collective communications are more efficienton large tensors. Fig. 2 (a) and (b) provide a quantitative
view, which show the total execution time to AllReduce
60M torch.float32 parameters with different numbers ofparameters per AllReduce. To maximize the bandwidth utilization, AllReduce operations are launched asynchronously
and block waiting on all of them together, mimicking DDP’s
gradient reduction algorithm. The experiments are conducted on an NVLink [3] enabled server with two NVIDIA
Quadro GP100 GPUs. NCCL [2] AllReduce runs on CUDA
input tensors directly, while Gloo [1] AllReduce runs on
CPU input tensors to eliminate the overhead of copying between CUDA memory to CPU memory when using Gloobackend. The total communication time clearly decreases
when using larger input tensors, for both NCCL and Gloo.
Gloo reaches pinnacle speed at around 500K parameters per
input tensor, while there is no clear saturation signal for
NCCL on NVLink with even 20M-parameter GPU tensors.
These experiments suggest that, instead of launching a
dedicated AllReduce immediately when each gradient tensor becomes available, DDP can achieve higher throughput
and lower latency if it waits for a short period of time andbuckets multiple gradients into one AllReduce operation.
This would be especially helpful for models with many small
parameters. However, DDP should not communicate all gradients in one single AllReduce, otherwise, no communication
can start before the computation is over. Fig. 2 (c) and (d)show the GPU and CPU backward computations time of a
ResNet152 [20] that contains roughly 60M parameters. The
X axis is the number of ready gradients and the Y axis thetime elapsed since the beginning of the backward pass. The
backward on GPU takes about 250ms to complete, which isin the same order of magnitude as NCCL on NVLink. Thisconclusion also applies to Gloo and CPU backward. These
measurements herald that, with relatively small bucket sizes,
DDP can launch AllReduce operations concurrently with the3008

Skipped
Gradient

Ready
Gradientt

Bucket

t

t

Ready
Time

AllReduce

g1

g1

g2

g2

g2

g3

g3

g3

g3

g4

g4

g4

g4

g1

g1

g2

Process 2

Process 1(a)t

Process 2

Process 1(b)

Figure 3: Gradient Synchronization Failures

backward pass to overlap communication with computation,
which would make a difference in per iteration latency.

3.2.3

Overlap Computation with Communication

The AllReduce operation on gradients can start before
the local backward pass finishes. With bucketing, DDP only
needs to wait for all contents in the same bucket before
launching communications. Under such settings, triggering AllReduce at the end of the backward pass is no longer
sufficient. It needs to react to more frequent signals andlaunches AllReduce more promptly. Therefore, DDP registers one autograd hook for each gradient accumulator. The
hook fires after its corresponding accumulator updating thegradients, and will inspect the bucket it pertains. If hooksof all gradients in the same buckets have fired, the last hookwill trigger an asynchronous AllReduce on that bucket.
Two caveats require caution. First, the reducing ordermust be the same across all processes, otherwise, AllReduce
contents might mismatch, resulting in incorrect reduction
result or program crash. However, PyTorch dynamicallybuilds the autograd graph in every forward pass, and different processes might not agree on the gradient ready order.
Fig. 3 (a) shows one example, where the two vertical axes
represent time and dotted lines indicate when a gradient isready. In process 1, the four gradients are computed in order, but the gradient g2 are computed after g3 and g4 onprocess 2. In this case, if all processes AllReduce buckets
as soon as they become ready, the AllReduce content would
mismatch. Therefore, all processes must use the same bucketing order, and no process can launch AllReduce on bucket
i+1 before embarking bucket i. If bucket 0 is the last one
that becomes ready, there is no way that communication canoverlap with computation. PyTorch v1.5.0 addresses thisproblem by using the reverse order of model.parameters()
as the bucketing order, assuming that, layers are likely registered according to the same order as they are invoked in theforward pass. Hence, the reverse order should approximately
represent the gradient computation order in the backward
pass. Admittedly, this is not a perfect solution, but is an approximation that we can rely on with minimum engineering
overhead.
Second, it is possible that one training iteration only involves a sub-graph in the model and the sub-graph can bedifferent from iteration to iteration, meaning that some gradients might be skipped in some iterations. However, as
gradient-to-bucket mapping is determined at the construction time, those absent gradients would leave some buckets
never seeing the final autograd hook and failing to mark thebucket as ready. As a result, the backward pass could hang.
Fig. 3 (b) shows an example, where the parameter corresponding to gradient g3 is skipped in one iteration, leading
to the absent of the ready signal for g3 . To address thisproblem, DDP traverses the autograd graph from the output
tensors of the forward pass to find all participating parameters. The readiness of those participating tensors is a sufficient signal to conclude the completion of the backward
pass. Therefore, DDP can avoid waiting for the rest of theparameter gradients by proactively marking them ready atthe end of the forward pass. Note that, this change does notprevent us from developing non-intrusive APIs, because application directly invokes the forward function on DDP andhence DDP can easily insert this step in its member function.
Algorithm 1: DistributedDataParallel
Input: Process rank r, bucket size cap c, local modelnet
1 Function constructor(net):2
if r=0 then3
broadcast net states to other processes
4
init buckets, allocate parameters to buckets in thereverse order of net.parameters()
5
for p in net.parameters() do
6acc ← p.grad accumulator
7
acc → add post hook(autograd hook)
8 Function forward(inp):
9out = net(inp)10
traverse autograd graph from out and markunused parameters as ready
return out
12 Function autograd hook(param index ):13
get bucket bi and bucket offset using param index
14
get parameter var using param index
15
view ← bi .narrow(offset, var.size())
16
view.copy (var.grad)
17
if all grads in bi are ready then18
mark bi as ready19
launch AllReduce on ready buckets in order
20
if all buckets are ready then21
block waiting for all AllReduce ops
11

Algorithm 1 presents the pseudo-code of DDP. The constructor contains two major steps, broadcasting model statesand installing autograd hooks. DDP’s forward function is a
simple wrapper of the local model’s forward. It traverses
the autograd graph to mark unused parameters accordingly.
The autograd hook takes the internal parameter index as input, which helps to find the parameter tensor and its belonging bucket. It writes the local gradient to the correct offset in
the bucket and then launches the asynchronous AllReduce
operation. There is an additional finalizing step omitted in
the pseudo-code that waits for AllReduce operations andwrites the value back to gradients at the end of the backward pass. Fig. 4 elucidates how DDP interacts with the local
model during the forward and backward passes.
The above solution works for most use cases. However, as
DDP always computes the average of all gradients and writes
them back to parameter .grad field, an optimizer cannot
distinguish whether a gradient has participated in the lastbackward pass or not. Due to the decoupled design of DDP
and the optimizer, there is no side channel for DDP to allude3009

w2 gw2

gb2

addmm2b2

gw1

allreduce2

gw1
gb1

gb1
gw2

allreduce1

gw2
gb2

gb2

bucket2b1

bucket1

gb1

addmm1bucket1

w1 gw1

DDP2

DDP1
bucket2

local model 1

local model 2

b1

gb1

gw1 w1

addmm1b2

gb2

will synchronize the accumulated gradients altogether. The
information of globally unused parameters also accumulates
in the bitmap, and serves when the next communication
takes place. Below is an example code snippet.

gw2 w2

1

addmm22
34

mse_loss

mse_loss

5
6

loss

Parameter

Process 1
Gradient

Autograd Edge

Copy

Communication

Figure 4: Distributed Gradient Reductionthat information to the optimizer. Without this information, the training process could suffer from regressions onmodel accuracy, e.g., when the optimizer uses gradient absence information to skip updating momentum values. To
tackle this problem, DDP should only touch gradients thatare indeed involved in the backward pass. Nevertheless, thisinformation cannot be extracted from the local autograd
graph alone, because locally absent gradients might still be
involved in the forward/backward pass in a peer DDP process.
Therefore, DDP uses a bitmap to keep track of local parameter participants and launches one additional AllReduce tocollect globally unused parameters. Unfortunately, DDP cannot coalesce this bitmap into other gradient AllReduce operations due to the potential mismatch in element types. Suchadditional overhead only materializes when the application
explicitly tells DDP to look for unused parameters, and hence
the price is only paid when necessary.

3.2.4

7

loss

Process 2

Gradient Accumulation

One common technique to speed up distributed data parallel training is to reduce gradient synchronization frequencies. Instead of launching AllReduce in every iteration, theapplication can conduct n local training iterations before
synchronizing gradients globally. This is also helpful if the
input batch is too large to fit into a device, where the application could split one input batch into multiple microbatches, run local forward and backward passes on every
micro-batch, and only launch gradient synchronization at
the boundaries of large batches. Theoretically, this should
produce the same results as if all data in the large batchis processed in one shot, as gradients will simply be accumulated to the same tensor. However, this conflicts withthe gradient reduction algorithm discussed in Section 3.2.3
to some degree. That algorithm would mark unused parameters as ready at the end of every forward pass, while
those unused parameters in one iteration still could participate in subsequent iterations. Moreover, DDP cannot distinguish whether the application plans to immediately invoke
optimizer.step() after backward or accumulate gradients
through multiple iterations. Therefore, we need to introduce
one additional interface (i.e., no sync) for this use case.
Under the hood, the implementation for no sync is very
simple. The context manager just toggles a flag on enteringand exiting the context, and the flag is consumed in theforward function of DDP. In no sync mode, all DDP hooks
are disabled, and the first backward pass out of the context

8

3.3

ddp = D i s t r i b u t e d D a t a P a r a l l e l ( net )with ddp . no_sync () :for inp , exp in zip ( inputs , e x p e c t e d _ outputs ) :# no synchronization , accumulate grads
loss_fn ( ddp ( inp ) , exp ) . backward ()# synchronize grads
loss_fn ( ddp ( another_inp ) , another_exp ) . backward ()opt . step ()Collective Communication

Distributed data parallel training uses a special communication pattern, where every participant provides an equallysized tensor and collects the global sum across all participants. This can certainly be implemented as a gather operator followed by local reductions on every participant using
point-to-point communication, but that would forfeit opportunities for performance optimizations [23]. DDP is built
on top of collective communication libraries, including three
options, NCCL [2], Gloo [1], and MPI [4]. 3 DDP takes theAPIs from the three libraries and wraps them into the same
ProcessGroup API. The name heralds that ProcessGroup
expects multiple processes to work collectively as a group.
All ProcessGroup instances construct at the same time byusing a rendezvous service, where the first arrival will blockwaiting until the last instance joins. For NCCL backend, theProcessGroup maintains a dedicated set of CUDA streams
for communication, so that communications will not blockthe computation in the default stream. As all communications are collective operations, subsequent operations on all
ProcessGroup instances must match in size and type andfollow the same order. Using the same ProcessGroup API
for all libraries allows us to experiment with different communication algorithms with the same DDP implementation.
For example, PyTorch v1.5 provides a composite roundrobin ProcessGroup implementation, which takes a list ofProcessGroup instances and dispatches collective communications to those ProcessGroup instances in a round-robinmanner. By using round-robin ProcessGroups, DDP can attain higher bandwidth utilization if a single NCCL, Gloo, or
MPI ProcessGroup is unable to saturate the link capacity.

4.

IMPLEMENTATION

The implementation of DDP has evolved several times in
the past few releases. This section focus on the current
status as of PyTorch v1.5.0. DDP implementation lives inboth Python and C++ files, with Python exposing the API
and composing non-performance-critical components, andC++ serving the core gradient reduction algorithm. The
Python API calls into C++ core through Pybind11 [5].

4.1

Python Front-end

The DDP nn.module is implemented in distributed.py,
which contains user-facing components, including the constructor, the forward function, and the no sync context
manager. Besides the general ideas highlighted in Section 3,
there are several implementation details in the Python frontend that shapes the behavior of DDP.
3
Please refer to documents of the three libraries for theirdesign and implementation.

3010

4.2

Core Gradient Reduction

Major development efforts are spent in gradient reduction
as it is the most performance-critical step in DDP. The implementation lives in reducer.cpp which consists of four main
components, namely, building parameter-to-bucket map, installing autograd hooks, launching bucket AllReduce, anddetecting globally unused parameters. This section expatiates on these four components.
Parameter-to-Bucket Mapping has a considerable impact on DDP speed. In every backward pass, tensors arecopied from all parameter gradients to buckets, and averaged gradients are copied back after AllReduce. To accelerate copy operations, buckets are always created on the samedevice as the parameters. If the model spans multiple devices, DDP takes device affinity into consideration to makesure that all parameters in the same bucket are on the samedevice. The order of AllReduce also makes a difference, as
it dictates how much communication can overlap with computation. DDP launches AllReduce in the reverse order ofmodel.parameters().
Autograd Hook is the entry point for DDP in the backward pass. During construction, DDP loops over all parameters in the model, finds the gradient accumulator on every
parameter, and installs the same post-hook function to every gradient accumulator. The gradient accumulator will
fire post hooks when the corresponding gradient is ready,and DDP will figure out when an entire bucket is ready tolaunch an AllReduce operation. However, as there is no
guarantee on the order of gradient readiness, DDP cannot selectively pick parameters to install hooks. In the current
implementation, each bucket keeps a count of pending gra-dients. Each post-hook function decrements the count, and
DDP marks a bucket as ready when that count reaches zero.
In the next forward pass, DDP replenishes the pending gradient count for every bucket.
Bucket AllReduce is the main source of communication
overhead in DDP. On one hand, packing more gradients into
the same bucket would reduce the amortized system overhead of communication. One the other hand, using a large
bucket size would result in longer lead time for reduction, as
each bucket needs to wait for more gradients. Hence, bucket
size is the key trade-off. By default, each bucket is 25MB in
size. Applications should measure their impact empiricallyand set it to the optimal value for their use cases.
Globally Unused Parameters’ gradients should stayintact during the forward and the backward passes. Detecting unused parameters requires global information, as one
parameter could be absent in one DDP process during one iteration, but participates training in the same iteration in another process. DDP maintains local unused parameter information in a bitmap, and launches an additional AllReduce
to gather a global bitmap. As the bitmap is much smaller
than tensor sizes, instead of creating per-bucket bitmaps,
all parameters in the model share the same bitmap. The
bitmap lives on CPU to avoid launching dedicated CUDA
kernels for each update. However, some ProcessGroup backends might not be able to run AllReduce on CPU tensors. For example, ProcessGroupNCCL only supports CUDA
tensors. Moreover, as DDP should work with any custom
ProcessGroup backend, it cannot make assumptions thatall backends support CPU tensors. To address this problem, DDP maintains another bitmap on the same device as
the first model parameter, and invokes a non-blocking copy
to move the CPU bitmap to the device bitmap for collective
communications.

5.

EVALUATIONThis section presents the evaluation results of PyTorch
DDP using an exclusive 32 GPU cluster and a shared entitlement. Fig. 5 shows the interconnection of the 8 GPUs
within the same server. In the exclusive cluster, the GPUs
are located on 4 servers, connected using Mellanox MT27700
ConnectX-4 100GB/s NIC. All 4 servers reside in the samerack, and each server is equipped with 8 NVIDIA Tesla V100
GPUs.
We only use
the shared entitlement
when a set of experNV2
NV1
NODE
iments require more than
7
32 GPUs. In the shared
6
entitlement, we submit
jobs to run on different
5numbers of GPUs where
4
different jobs can run
3
on different machines,
2and hence the hardwareand network connectiv1
ity can vary from job to0job. Although the dis0
1
2
3
4
5
6
7
GPUs
parity in the test environment can lead to different latency measures
Figure 5: GPU Topology
even for the same code,
we pack the same set of
GPUs

Configurable Knobs are exposed in the DDP constructor
API, including 1) process group to specify a process group
instance for DDP to run AllReduce, which helps to avoid
messing up with the default process group, 2) bucket cap mb
to control the AllReduce bucket size, where applications
should tune this knob to optimize training speed, and 3)
find unused parameters to toggle whether DDP should detect unused parameters by traversing the autograd graph.
Model Device Affinity in the local model also governs
DDP’s behavior, especially if the model spans multiple devices, which is common when the model is too large to fit
into a single device. For large models, applications can place
different layers of the model onto difference devices, and use
Tensor.to(device) API to move intermediate output fromone device to another. DDP also works with multi-device
models. As long as the device ids argument is None oran empty list, DDP will inspect the model, perform sanity
checks and apply configurations accordingly. Then, it treats
the multi-device model as one entirety.
Model Buffers are necessary when layers need to keep
track of states like the running variance and the runningmean (e.g., BatchNorm). DDP supports model buffers by letting the process with the rank 0 to take the authority. If themodel contains buffers, DDP will broadcast the buffer values
from rank 0 process to all other processes before starting
the forward pass on the local model. This behavior is also
compatible with the no sync mode. When no sync mode isenabled, it sets a flag in the forward pass properly to indicate whether it expects gradient reductions in the immediate
backward pass. If the communication takes place, DDP willthen broadcast buffers prior to the subsequent forward pass.

3011

Figure 6: Per Iteration Latency Breakdown
experiments into the same job, so that the trend shown inthe same curve is still meaningful.
We measure DDP per iteration latency and scalability using two popular models, ResNet50 [20] and BERT [15], torepresent typical vision and NLP applications. Most experiments use randomly generated synthetic inputs and labels, which are sufficient as the purpose is to compare periteration latency instead of model accuracy. Experiments
compute losses using the CrossEntropyLoss function andupdate parameters using the SGD optimizer. Configurations
for accuracy-related experiments will be explained in detailclose to their presentations.

5.1

Latency Breakdown

A typical training iteration contains three steps: forward
pass to compute loss, backward pass to compute gradients,and optimizer step to update parameters. With DDP, thebackward pass involves local computation and AllReduce
communication. To demonstrate the effectiveness of overlapping computation with communication, Fig. 6 plots thelatency breakdown when using NCCL and Gloo backends for
ResNet50 and BERT models respectively. All experiments
are conducted using 32 GPUs across 4 machines. To visually compare the speedup on different model and backend
combinations, we normalize the total latency to 1 for all nonoverlapping cases. The results demonstrate that the backward pass is the most time-consuming step with PyTorch
DDP training, as AllReduce communications (i.e., gradientsynchronization) are completed in this step. This observation justifies that the DDP backward pass deserves the mostefforts for improvements. Within the backward pass, thecommunication step takes more than half of the total delay
and this is exacerbated with the increase of the model size.
Between these two backends, NCCL is considerably fasterthan GLOO. The speedup is most effective when the computation and communication take roughly the same amountof time as they can overlap more. The overlapping approachhelps ResNet and BERT on NCCL attain 38.0% and 35.2%
speedup. With GLOO backend, the gain shrinks to 26.8%
and 21.5% respectively, as GLOO communication becomes
the dominating delay in the backward pass.

5.2

Bucket Size

To avoid launching an excessive number of AllReduce operations, DDP organizes small gradients into larger buckets
and synchronizes each bucket using an AllReduce operation. With this design, bucket size is an important configuration knob. DDP exposes this knob to applications throughbucket cap mb argument. No single bucket size can best
serve all applications. This value should be measured anddetermined empirically. The default value of bucket cap mb

is 25MB, which is our best effort estimation based experiences. The following experiments also confirm this is a reasonable choice for ResNet50 and BERT. This section compares per iteration latency across different bucket sizes using
16 GPUs on two machines. Zero bucket size means each gradient will be communicated on its own as soon as it is ready.
This serves as a baseline on one extreme of the bucket size
spectrum. The other extreme is communication all gradients in one short, which is skipped as results in Fig. 7 andFig. 8 clearly show the best option for both ResNet50 and
BERT is somewhere in the middle.
Fig. 7 (a) uses box-whisker to illustrate how bucket size
affects per iteration latency on ResNet50 with NCCL backend. The x-axis is the bucket size in MBs, and Y-axis periteration latency in seconds. The outliers are the tiny delay
spikes at 100 iteration boundaries caused by DDP instancere-construction and input data regeneration. Other thanthat, delays of most iterations concentrate in a very narrow time range, which also agrees with the results shownin Fig. 6 (a). The results show that the highest speed isachieved between 10MB and 25MB bucket sizes. Fig. 7 (b)presents the same measurements for Gloo backend. The results are different from NCCL backend in two ways, 1) per
iteration latency falls into a large range, 2) the 5MB bucket
size attains higher speed compared to 10MB and 25MB. The
first difference matches with Fig. 6 (b). To understand thesecond difference, let us revisit Fig. 2 (b) on Gloo AllReduce
latency across different tensor sizes. It’s clear that the total
AllReduce time fluctuates around the same level when thebucket size is larger than 512KB. Therefore, larger bucket
sizes beyond 512KB with Gloo backend would only mean
longer waiting time for gradients, which leads to longer per
iteration latency. Fig. 7 (c) and (d) show the measurements
for BERT model. As BERT model contains 15X more parameters compared to ResNet50, intuitively, it should benefit from larger buckets as larger communication overheads
would dwarf the waiting time for the first bucket. The results verified the intuition with NCCL backend, where 50MB
bucket size leads to the best performance. However, with
Gloo backend, 5MB bucket size still wins with the lowest
per iteration latency.
Fig. 8 presents the results of the same set of experiments
but on 32 GPUs. In this case, the outliers span a larger
range, which is not surprising as synchronizations usuallytake longer with more participants and the impact of strangler is more prominent. Fig. 8 (a) and (b) both suggestthat 0MB bucket size leads to obviously longer per iteration latency on 32 GPUs compared to 16 GPUs, as pergradient reductions on a larger cluster are expected to be
slower. However, when bucket size is set to above 5MB,
scaling from 16 GPUs to 32 GPUs does not lead to a noticeable speed regression. This is probably because although
individual AllReduce operations is expected to be slower,
asynchronous execution and parallelism could help to hide
the overall delay.

5.3

Scalability

To understand the scalability of DDP, we measure per iteration training latency of ResNet50 and BERT using NCCL
and Gloo backend on up to 256 GPUs in the shared entitlement. Results are presented in Fig. 9. The X-axis is thenumber of GPUs, and Y-axis the latency. Figure 9 (a) showsthat the per iteration latency steadily increases as it scales

3012

1.4

0.35

0.35

1.2

1.2

0.30
0.25
0.20
0.15
0.10

0

5

10
25
Bucket Size (MB)

0.30
0.25
0.20
0.15
0.10

50

(a) ResNet50 on NCCL

0

5

10
25
Bucket Size (MB)

1.0
0.8
0.6
0.4
0.2

50

Per Iteration Latency (Sec)1.4
Per Iteration Latency (Sec)0.40
Per Iteration Latency (Sec)Per Iteration Latency (Sec)0.40(b) ResNet50 on Gloo0

5

10 25 50
Bucket Size (MB)

100

1.0
0.8
0.6
0.4
0.2

200

(c) BERT on NCCL

0

5

10 25 50
Bucket Size (MB)

100

200

(d) BERT on Gloo

1.4

1.4

0.35

0.35

1.2

1.2

0.30
0.25
0.20
0.15
0.10

0

5

10
25
Bucket Size (MB)50(a) ResNet50 on NCCL

0.30
0.25
0.20
0.15
0.10

0

5

10
25
Bucket Size (MB)50

Per Iteration Latency (Sec)0.40

Per Iteration Latency (Sec)0.40
Per Iteration Latency (Sec)Per Iteration Latency (Sec)Figure 7: Per Iteration Latency vs Bucket Size on 16 GPUs

1.0
0.8
0.6
0.4
0.2

(b) ResNet50 on Gloo0

5

10 25 50
Bucket Size (MB)

100

200

1.0
0.8
0.6
0.4
0.2

(c) BERT on NCCL

0

5

10 25 50
Bucket Size (MB)

100

200

(d) BERT on Gloo

Figure 8: Per Iteration Latency vs Bucket Size on 32 GPUsout. Using 256 GPUs leads to 100% slow down in each iteration compared to local training, meaning that the real
scaling factor is 256 × 50% = 128. With the BERT model,
the per-iteration latency significantly increases due to thelarger model size. Another observation is that the 16-GPU
case suffers a longer per-iteration delay compared to the 32GPU case in Figure 9 (c). We suspect this is because either
the 16-GPU experiments were on a slow or congested linkor there are other workflows in the shared entitlement competing for resources with our job. Fig. 9 (b) and (d) show
the results for Gloo backend and the per-iteration slowdown
is about 3X for ResNet and 6X for BERT when using 256
GPUs. The deteriorated training speed with larger model
sizes indicates that the network is the bottleneck resource
when using Gloo backend in this experiment.
In general, scaling out to more GPUs slows down individual iterations. One option to mitigate the overhead isskipping gradient synchronizations, i.e., perform gradient
reduction every n iterations. This approach helps to considerably reduce the amortized latency. Fig. 10 depicts theaverage per iteration latency for conducting gradient reduction every 1, 2, 4, and 8 iterations. To visually compare
the effectiveness of this method, we consolidated different
skipping configurations for the same model and backend
combination into the same figure. ResNet50 on NCCL and
Gloo sees 38% and 57% speed up with 256 GPUs when conducting gradient sync every 8 iterations. There is a sudden
jump in delay with NCCL backend when scaling from 128
to 256 and this occurs to all experiments shown in this figure. We believe this is caused by slow or congested links
among some of those 256 nodes which are not included in
the 128-GPU experiments. Besides the per iteration latency,
it’s also crucial to measure the convergence speed to ver-ify if the acceleration might be erased by convergence slowdown. The experiments use MNIST [25] dataset to train the
ResNet. The learning rate is set to 0.02 and the batch sizeis 8. Results are plotted in Fig. 11 (a), which only contains
the measurements for NCCL backend as the communication layer does not change the convergence speed. X-axis is
the number of iterations and Y-axis the loss. Please notethat the goal of this experiment is not developing the bestmodel for MNIST, instead, it only aims to show the impact of skipping synchronization on the model convergence.
The raw loss data oscillate severely, which are presented by
the tiny dots. Directly connecting them into a line would
result in the last curve covering all previous drawn ones,
making them less visible. Therefore, we apply an order 3
low pass filter by using filtfilt from SciPy [8] and plot
the smoothed loss curve. The figure confirms that using
no sync in this case only leads to negligible exacerbation to
the convergence speed. However, we must emphasize thatthe impact of no sync could depend on the configuration.
Fig. 11 (b) shows similar measurements by replacing batch
size to 256 and learning rate to 0.06. As highlighted by the
red box in the right bottom corner, no sync hurts the final training loss. It is because large batch size and no sync
cause more gradients to be accumulated between consecutive communications and optimizer steps, which implicitly
requires using a smaller learning rate. In summary, whenskipping synchronizations properly, DDP attains near linear
scalability with negligible accuracy penalty.

5.4

Round-Robin Process Group

Another technique to speed up training is to use multiple
process groups to work around subtle intrinsic concurrency
limitations in process group backend implementations. The

3013

3.0

0.5

0.5

2.5

2.5

0.4
0.3
0.2
0.1

1

2

4

0.4
0.3
0.2
0.1

8 16 32 64 128 256
Number of GPUs(a) ResNet50 on NCCL

1

2

4

Per Iteration Latency (Sec)3.0
Per Iteration Latency (Sec)0.6
Per Iteration Latency (Sec)Per Iteration Latency (Sec)

0.6

2.0
1.5
1.0
0.5
0.0

8 16 32 64 128 256
Number of GPUs

1(b) ResNet50 on Gloo

2

4

2.0
1.5
1.0
0.5
0.0

8 16 32 64 128 256
Number of GPUs

1(c) BERT on NCCL

2

4

8 16 32 64 128 256
Number of GPUs(d) BERT on Gloo

0.6

1

2

4

2.0

1.8

0.2

1.6

1.6

2

4

0.11
0.10
0.09
1

2

4
8
16
Number of GPUs

2432(a) ResNet50 on NCCL

0.18
0.16
0.14
0.12
0.10
0.08

0

1

2

4
8
16
Number of GPUs

24

50

100 150 200 250 300 350
Number Iterations(b) Batch Size = 256

1.2

0.50
rr1
rr3
rr5

0.20

Medium Per Iteration Latency (Sec)Medium Per Iteration Latency (Sec)Medium Per Iteration Latency (Sec)0.12

100 150 200 250 300 350
Number Iterations

Figure 11: Accuracy with Skipping Synchronization

0.22

0.13

50

(a) Batch Size = 8

Figure 10: Skip Gradient Synchronization

0.08

0

8 16 32 64 128 256
Number of GPUs(b) ResNet50 on Gloo

0.14

2.0

1.8

1

nccl
no_sync_2
no_sync_4
no_sync_8

2.2

0.3

(a) ResNet50 on NCCL

rr1
rr3
rr5

nccl
no_sync_2
no_sync_4
no_sync_8

2.2

Loss

0.4

0.1

8 16 32 64 128 256
Number of GPUsgloo
no_sync_2
no_sync_4
no_sync_8

0.5

32

rr1
rr3
rr5

0.45
0.40

Medium Per Iteration Latency (Sec)nccl
no_sync_2
no_sync_4
no_sync_8

Loss

0.32
0.30
0.28
0.26
0.24
0.22
0.20
0.18
0.16
0.14

Average Per Iteration Latency (Sec)Average Per Iteration Latency (Sec)Figure 9: Scalability

0.35
0.30
0.25
0.20
0.15
0.10

(b) ResNet50 on Gloo

1

2

4
8
16
Number of GPUs

2432(c) BERT on NCCL

rr1
rr3
rr5

1.0
0.8
0.6
0.4
0.2
1

2

4
8
16
Number of GPUs

2432(d) BERT on Gloo

Figure 12: Round-Robin Process Group
concurrency limitations could come from NCCL streams or
Gloo threads, depending on the type of the backend, whichmight prevent one process group instance to fully utilize
all link bandwidth. The PyTorch distributed package supports composing a Round-Robin process group with multiple NCCL or Gloo process groups, which dispatches collective communications to different process group instances in
Robin-Robin order. Fig. 12 plots the per iteration latencyof Round-Robin process group using 1, 3, and 5 NCCL or
Gloo process groups, where rrx stands for Round-Robinwith x process group instances. ResNet50 on NCCL backend sees negligible differences with different amounts of process groups, meaning that for relatively small models like
ResNet50, bandwidth is not the bottleneck resource. Noticeable difference can be observed in ResNet50 on Gloo,
where rr3 consistently outperforms rr1. The most prominent acceleration occurs in BERT model with NCCL backend, where rr3 achieves 33% speedup compared to rr1 on
16 GPUs, revealing that one NCCL group is incompetent tosaturate the link capacity.

6.

DISCUSSION

This section discusses lessons learned from our experiments and past experiences. We then present several ideas
for future improvements.

6.1

Lessons Learned

Distributed data parallel training is a conceptually simple
but practically subtle framework. There are various techniques to improve the training speed, creating a complex
configuration space. Based on our observations, there isno single configuration that would work for all use cases,
as it would highly depend on the model size, model structure, network link bandwidth, etc. However, on individual
configuration dimensions, we summarized intuitions to help
application developers to quickly navigate to a small range
which likely contains the optimal solution for a given use
case. The specific value of the configuration needs to be determined using empirical measurements for every different
deployment.

3014

• Communication Backend: NCCL is considerably
faster than Gloo in most use cases. When available,
applications should seek to use NCCL as the primarycollective communication backend.
• Bucket Size: Both excessively small or large bucket
sizes are detrimental to communication performance.
The optimal value lives in between and depends on
the type of communication backend employed. The
optimal bucket sizes are likely to increase with the sizeof the model in a sub-linear manner.
• Resource Allocation: There is a significant slowdown with NCCL backend when scaling models acrossmachine boundaries, if the bandwidth across machines
is considerably lower than that between same-machine
GPUs. In such cases, it is recommended to keep the
DDP group within the same machine. If the training requires larger scale, developers can explore enabling no sync mode if it attains acceptable convergence speed.

6.2

Future Improvements

While we implement and maintain the DDP package, several ideas for improvements popped up. This section discusses the basic ideas behind those improvements.

6.2.1

instead of parameters to buckets and all processes skip thesame bucket in the same iteration. Both options require
extra coordination across all DDP processes, which can be
implemented by using the same random seed or having an
authority process to broadcast the plan.

6.2.3

Another potential improvement for DDP is to reduce thevolume of data for communication by compressing gradients.
The absolute value of gradients are usually small, whichmight not require float32 or float64 types. Current DDP
implementation always uses the parameter type as the gradient type that can become an overkill especially when themodel is approaching convergence. In this case, DDP would
benefit from adaptive compression levels by only communicating gradients with the necessary precision. Some recent
research work [34] even proposes more aggressive compression schemes, where by trading a tiny amount of model accuracy, applications can significantly accelerate distributed
training by communicating just 1 bit for each gradient.

7.

Gradient Order Prediction• Synchronous update vs Asynchronous update: With
the former, all model replicas can use AllReduce to collectively communicate gradients or parameters, while
the asynchronous scheme employs P2P communication
to update gradients or parameters independently.
• Cross-iteration vs Intra-iteration: Cross-iteration parallelism (e.g., pipeline parallelism) allows the lifetimeof multiple iterations to overlap with each other, while
intra-iteration scheme focuses on parallelizing training
within one iteration.
• Data parallel vs Model parallel: Data parallel training distributes input data to multiple model replicas,
while model parallelism divides the model into smaller
pieces, which is especially helpful when the model is
too large to fit in one device or machine.

Layer Dropping

One technique to accelerate training and avoid overfitting is to randomly drop layers during the forward pass [17].
This works well with local training. As every forward pass
would build a new autograd graph, those skipped layers will
not participate in the backward pass either. This idea also
works with DDP, because parameters in skipped layers can bemarked as ready in the forward pass and DDP will not wait for
their autograd hooks during the backward pass. Although
DDP would produce the correct result, this technique alone
is inadequate to accelerate distributed data parallel training
the same way as local training due to the fixed parameterto-bucket mapping. As AllReduce uses a bucket as the minimum granularity, it cannot judiciously react to vacancies inbuckets (i.e., skipped layers or parameters). Consequently,
regardless of how the forward pass skips layers, there is always the same amount of data to be communicated acrossthe wire during the backward pass. Besides, DDP cannot afford the luxury to adjust all buckets to cooperate with randomly skipped layers, as that would result in unacceptable
memory allocation overhead. To tackle this problem, one solution is to keep bucket buffers intact but modify parameterto-bucket mappings accordingly. Another option is to perform layer skips at the bucket level, i.e., DDP can map layers

RELATED WORK

Distributed training algorithms can be categorized into
different types from different perspectives. Below are three
popular categorizations.

Although DDP cannot deterministically detect the backward computation order on all parameters at construction
time, the order usually does not change that often in practice. One viable solution is to trace the backward order using
autograd hooks and update parameter to bucket mapping
accordingly. As bucket re-allocation will introduce noticeable overhead, it should be conducted infrequently. Given
the existing complexities in DDP, tracing overhead should be
negligible. Nevertheless, if there are disparities among tracing results from different iterations, additional complexities
will be necessary to reach a consensus.

6.2.2

Gradient Compression

Table 1 summarizes some recent distributed training solutions by marking which scheme they can support. Besides advances in training schemes, prior work has also explored different communication algorithms, including treebased AllReduce [23], heterogeneity-aware interconnection
structure [39], and AllReduce decomposition [14]. As thispaper focuses on DDP, the remainder of this section only elaborates and compares closely related techniques, i.e., Synchronous, Intra-iteration, and Data parallel training schemes.
The techniques presented in this paper were first implemented and released in PyTorch v1.1. Similar computationcommunication overlap techniques are also introduced inTensorFlow v2.2 as the Multi Worker Mirrored Strategy [10].
This technique is researched in academia as well. GradientFlow [37] combines bucketing AllReduce with skipping
parameter synchronizations. Compared to PyTorch DDP,
instead of skipping the entire synchronization step in one
iteration, GradientFlow selectively communicates a subset
of gradients. Although this strategy helps to reduce communication overhead for gradients, it requires an additional3015

communication phase to attain consensus on which gradients to synchronize. As a result, the overhead incurred toacquire consensus might overshadow the speedup achievedin gradient synchronizations, especially for small models orlarge network round-trip delays.
Another approach to speed up distributed training is preempting and prioritizing communications based on the order of downstream computations. Jayarajan et al. [22] proposed to prioritize gradient synchronizations and parameter
updates based on the forward order instead of the backward order, meaning that gradient buckets containing theinitial layers should receive higher priorities than those in
the final layers. Communications should still start from final layer gradients, as they will become ready earlier, buthigher priority gradients (i.e., in initial layers) can preempt
lower priority ones. This design allows the forward pass inthe next iteration to start sooner, even before finishing gradients communications in the previous iteration, creating more
opportunities to overlap computations and communications.
ByteScheduler [31] explored scheduling communications for
distributed data parallel training as well. However, insteadof binding with a single framework, ByteScheduler works for
multiple frameworks by inserting a common core scheduler
between framework APIs and framework engines and uses
per-engine plugins to intercept communication invocations.
To integrate with PyTorch, ByteScheduler builds on top of
Horovod [35] which launches communication in the optimizer. One downside of this approach is that, there is a hardbarrier between the backward pass and the optimizer step.
As a result, communication can only overlap with the next
forward pass instead of the current backward pass. With dynamic graphs, the next iteration might touch a different setof parameters, which would invalidate the schedule derived
from the previous iteration. PACE [12] computes the optimal communication schedule and implements preemption
by segmenting primitive AllReduce operations into smaller
pieces. Although segmenting can indeed mimic preemption,
it will on the other hand hurt the total communication time
as we have seen in Fig. 2. A more efficient approach would
be to natively support prioritization in the communication
libraries (e.g., NCCL and Gloo).

Table 1: Distributed Training Solutions: Six schemes
are Synchronous-Update vs Asynchronous-Update, Cross¯¯
Iteration
vs Intra-Iteration, D¯ata-Parallel vs Model-Parallel¯Scheme¯ S A C ¯I D M√√ √PT DDP [9]√ √ √ √ √PT RPC [6]√√ √TF Mirrored Worker [10]√√ √ √TF ParameterServer [11]√√ √ √Mesh TensorFlow [36]√√√GPipe [21]√√ √Horovod [35]√√ √GradientFlow [37]√√ √SlowMo [40]√√ √ √ √
PipeDream [29]√√ √ √ZeRO [32]√ √√ √ √Parallax [24]√√ √ √ByteScheduler [31]√√ √ √TicTac [19]√√ √PACE [12]duce considerable overhead. Hence, applications can choose
which techniques to use based on the size of the given modeland available resources. PipeDream [29] employs a different
approach where the model stack is decomposed into multiple
stages, where data parallelism is applied within one stage
and pipeline with model parallelisms govern the workload
across stages. One subtle detail is that to attain high training speed, PipeDream slightly sacrifices accuracy by using
the latest gradients from multiple concurrent passes. Although the gradient might not be derived from the current
parameter states, the authors show that this mismatch is tolerable in practice. Parallax [24] explored a hybrid structurethat combines parameter-server [27] and collective communications. Models are partitioned based on sparsity, where
dense parameters are communicated using AllReduce andsparse tensors are placed to parameter servers. This design
avoids densifying sparse tensors and communicating empty
values, which is especially helpful for NLP models.

8.
The mixture of different parallelism scheme fosters evenmore powerful training paradigms. Mesh-TensorFlow [36]combines data parallelism with model parallelism. It vertically divides some layers by dimensions and replicating otherlayers where the given dimension is absent. ZeRO [32] also
combines data parallelism with model parallelism, but withminimum model replication to support fast training on super large models. The authors observed that main memory
consumption contributors are input data, model parameters, gradients, optimizer states, and activations. Splitting
input data is trivial. However, model parameters and activations are compulsory ingredients for backward passes.
ZeRO addressed this problem by partitioning parameters,
gradients, and optimizer states on each DDP instance. Parameters are broadcast from the owner DDP instance to allothers when necessary. Activations are recomputed during
the backward pass. Compared to PyTorch DDP, ZeRO canscale to much larger models as each process only needs tomaintain a small partition of the model. The high scalability is achieved by sacrificing the training speed, as the additional re-computation, broadcast, and gather would intro-

CONCLUSION

This paper explained the design and implementation of
the distributed data parallel module in PyTorch v1.5, andconducted performance evaluations on NCCL and Gloo backend using ResNet50 and BERT models. DDP accelerates
training by aggregating gradients into buckets for communication, overlapping communication with computation, andskipping synchronizations. We also highlighted real-world
caveats in gradient synchronization which are important forbroad adoption. Results showed that DDP with NCCL backend can achieve near-linear scalability on 256 GPUs when
configured properly. The measurements also revealed thatthe backward pass in DDP is the most expensive step in training and requires efforts from both framework developers toenable optimization algorithms and application developers
to empirically configure the knobs. Based on our observations, we shared lessons learned from serving a variety
of application, discussed potential future improvements for
distributed data parallel training, and enthusiastically encourage open source community to experiment with morenovel ideas.

3016

9.

REFERENCES[1] Gloo: a collective communications library.
https://github.com/facebookincubator/gloo, 2019.
[2] NVIDIA Collective Communications Library (NCCL).
https://developer.nvidia.com/nccl, 2019.
[3] NVLINK AND NVSWITCH: The Building Blocks ofAdvanced Multi-GPU Communication. https://www.nvidia.com/en-us/data-center/nvlink/,
2019.
[4] Open MPI: A High Performance Message Passing
Library. https://www.open-mpi.org/, 2019.
[5] Pybind11: Seamless operability between C++11 andPython. https://pybind11.readthedocs.io/, 2019.
[6] PyTorch Distributed RPC Framework.
https://pytorch.org/docs/master/rpc.html, 2019.
[7] PyTorch Module forward Function.
https://pytorch.org/docs/stable/nn.html#torch.
nn.Module.forward, 2019.
[8] SciPy: open-source software for mathematics, science,and engineering. https://docs.scipy.org/, 2019.
[9] PyTorch DistributedDataParallel.
https://pytorch.org/docs/stable/nn.html#torch.
nn.parallel.DistributedDataParallel, 2020.
[10] TensorFlow Distributed Training
MultiWorkerMirroredStrategy.
https://www.tensorflow.org/guide/distributed_
training#multiworkermirroredstrategy, 2020.
[11] TensorFlow Distributed Training
ParameterServerStrategy.
https://www.tensorflow.org/guide/distributed_
training#parameterserverstrategy, 2020.
[12] Y. Bao, Y. Peng, Y. Chen, and C. Wu. Preemptive
all-reduce scheduling for expediting distributed dnn
training. In IEEE INFOCOM, 2020.
[13] M. Bojarski, D. Del Testa, D. Dworakowski, B. Firner,
B. Flepp, P. Goyal, L. D. Jackel, M. Monfort,
U. Muller, J. Zhang, et al. End to end learning forself-driving cars. arXiv preprint arXiv:1604.07316,
2016.
[14] M. Cho, U. Finkler, M. Serrano, D. Kung, andH. Hunter. Blueconnect: Decomposing all-reduce fordeep learning on heterogeneous network hierarchy.
IBM Journal of Research and Development, 63(6):1–1,
2019.
[15] J. Devlin, M.-W. Chang, K. Lee, and K. Toutanova.
Bert: Pre-training of deep bidirectional transformers
for language understanding. arXiv preprint
arXiv:1810.04805, 2018.
[16] M. Du, F. Li, G. Zheng, and V. Srikumar. Deeplog:
Anomaly detection and diagnosis from system logs
through deep learning. In Proceedings of the 2017
ACM SIGSAC Conference on Computer andCommunications Security, pages 1285–1298, 2017.
[17] A. Fan, E. Grave, and A. Joulin. Reducing
transformer depth on demand with structured
dropout. arXiv preprint arXiv:1909.11556, 2019.
[18] X. Guo, S. Singh, H. Lee, R. L. Lewis, and X. Wang.
Deep learning for real-time atari game play using
offline monte-carlo tree search planning. In Advancesin neural information processing systems, pages
3338–3346, 2014.

[19] S. H. Hashemi, S. A. Jyothi, and R. H. Campbell.
Tictac: Accelerating distributed deep learning with
communication scheduling. arXiv preprint
arXiv:1803.03288, 2018.
[20] K. He, X. Zhang, S. Ren, and J. Sun. Deep residual
learning for image recognition. In Proceedings of theIEEE conference on computer vision and patternrecognition, pages 770–778, 2016.
[21] Y. Huang, Y. Cheng, A. Bapna, O. Firat, D. Chen,
M. Chen, H. Lee, J. Ngiam, Q. V. Le, Y. Wu, et al.
Gpipe: Efficient training of giant neural networks
using pipeline parallelism. In Advances in Neural
Information Processing Systems, pages 103–112, 2019.
[22] A. Jayarajan, J. Wei, G. Gibson, A. Fedorova, and
G. Pekhimenko. Priority-based parameter propagation
for distributed dnn training. In Proceedings of Machine
Learning and Systems 2019, pages 132–145, 2019.
[23] S. Jeaugey. Massively Scale Your Deep Learning
Training with NCCL 2.4.
https://devblogs.nvidia.com/
massively-scale-deep-learning-training-nccl-2-4/,
February 2019.
[24] S. Kim, G.-I. Yu, H. Park, S. Cho, E. Jeong, H. Ha,
S. Lee, J. S. Jeong, and B.-G. Chun. Parallax:
Sparsity-aware data parallel training of deep neuralnetworks. In Proceedings of the Fourteenth EuroSys
Conference 2019, pages 1–15, 2019.
[25] Y. LeCun, C. Cortes, and C. Burges. The MNIST
Database. http://yann.lecun.com/exdb/mnist/,
1999.
[26] Y. LeCun, D. Touresky, G. Hinton, and T. Sejnowski.
A theoretical framework for back-propagation. In
Proceedings of the 1988 connectionist models summer
school, volume 1, pages 21–28. CMU, Pittsburgh, Pa:
Morgan Kaufmann, 1988.
[27] M. Li, D. G. Andersen, J. W. Park, A. J. Smola,
A. Ahmed, V. Josifovski, J. Long, E. J. Shekita, and
B.-Y. Su. Scaling distributed machine learning withthe parameter server. In 11th {USENIX} Symposium
on Operating Systems Design and Implementation({OSDI} 14), pages 583–598, 2014.
[28] H. Mao, M. Cheung, and J. She. Deepart: Learning
joint representations of visual arts. In Proceedings ofthe 25th ACM international conference on Multimedia,
pages 1183–1191, 2017.
[29] D. Narayanan, A. Harlap, A. Phanishayee,
V. Seshadri, N. R. Devanur, G. R. Ganger, P. B.
Gibbons, and M. Zaharia. Pipedream: generalized
pipeline parallelism for dnn training. In Proceedings ofthe 27th ACM Symposium on Operating Systems
Principles, pages 1–15, 2019.
[30] A. Paszke, S. Gross, F. Massa, A. Lerer, J. Bradbury,
G. Chanan, T. Killeen, Z. Lin, N. Gimelshein,
L. Antiga, A. Desmaison, A. Kopf, E. Yang,
Z. DeVito, M. Raison, A. Tejani, S. Chilamkurthy,
B. Steiner, L. Fang, J. Bai, and S. Chintala. Pytorch:An imperative style, high-performance deep learning
library. In Advances in Neural Information Processing
Systems 32, pages 8024–8035. Curran Associates, Inc.,
2019.
[31] Y. Peng, Y. Zhu, Y. Chen, Y. Bao, B. Yi, C. Lan,
C. Wu, and C. Guo. A generic communication

3017

scheduler for distributed dnn training acceleration. In
Proceedings of the 27th ACM Symposium on Operating
Systems Principles, pages 16–29, 2019.
[32] S. Rajbhandari, J. Rasley, O. Ruwase, and Y. He.
Zero: Memory optimization towards training a trillion
parameter models. arXiv preprint arXiv:1910.02054,
2019.
[33] B. Ramsundar, P. Eastman, P. Walters, and
V. Pande. Deep learning for the life sciences: applying
deep learning to genomics, microscopy, drug discovery,
and more. ” O’Reilly Media, Inc.”, 2019.
[34] F. Seide, H. Fu, J. Droppo, G. Li, and D. Yu. 1-bit
stochastic gradient descent and its application todata-parallel distributed training of speech dnns. In
Fifteenth Annual Conference of the International
Speech Communication Association, 2014.
[35] A. Sergeev and M. D. Balso. Horovod: fast and easy
distributed deep learning in TensorFlow. arXiv
preprint arXiv:1802.05799, 2018.
[36] N. Shazeer, Y. Cheng, N. Parmar, D. Tran,
A. Vaswani, P. Koanantakool, P. Hawkins, H. Lee,M. Hong, C. Young, et al. Mesh-tensorflow: Deeplearning for supercomputers. In Advances in Neural
Information Processing Systems, pages 10414–10423,
2018.
[37] P. Sun, Y. Wen, R. Han, W. Feng, and S. Yan.
Gradientflow: Optimizing network performance forlarge-scale distributed dnn training. IEEE
Transactions on Big Data, 2019.
[38] A. Van den Oord, S. Dieleman, and B. Schrauwen.
Deep content-based music recommendation. In
Advances in neural information processing systems,
pages 2643–2651, 2013.
[39] G. Wang, S. Venkataraman, A. Phanishayee,
J. Thelin, N. Devanur, and I. Stoica. Blink: Fast andgeneric collectives for distributed ml. arXiv preprint
arXiv:1910.04940, 2019.
[40] J. Wang, V. Tantia, N. Ballas, and M. Rabbat.
Slowmo: Improving communication-efficient
distributed sgd with slow momentum. arXiv preprint
arXiv:1910.00643, 2019.

3018