# ScaDLES: Scalable Deep Learning over Streaming data at the Edge

Sahil Tyagi  Department of Intelligent Systems Engineering
Luddy School of Informatics, Computing and Engineering
Indiana University Bloomington
Indiana, USA
<EMAIL_ADDRESS>Martin Swany  Department of Intelligent Systems Engineering
Luddy School of Informatics, Computing and Engineering
Indiana University Bloomington
Indiana, USA
<EMAIL_ADDRESS>
###### Abstract

Distributed deep learning (DDL) training systems are designed for cloud and
data-center environments that assumes homogeneous compute resources, high
network bandwidth, sufficient memory and storage, as well as independent and
identically distributed (IID) data across all nodes. However, these
assumptions don’t necessarily apply on the edge, especially when training
neural networks on streaming data in an online manner. Computing on the edge
suffers from both systems and statistical heterogeneity. Systems heterogeneity
is attributed to differences in compute resources and bandwidth specific to
each device, while statistical heterogeneity comes from unbalanced and skewed
data on the edge. Different streaming-rates among devices can be another
source of heterogeneity when dealing with streaming data. If the streaming
rate is lower than training batch-size, device needs to wait until enough
samples have streamed in before performing a single iteration of stochastic
gradient descent (SGD). Thus, low-volume streams act like stragglers slowing
down devices with high-volume streams in synchronous training. On the other
hand, data can accumulate quickly in the buffer if the streaming rate is too
high and the devices can’t train at line-rate. In this paper, we introduce
_ScaDLES_ to efficiently train on streaming data at the edge in an online
fashion, while also addressing the challenges of limited bandwidth and
training with non-IID data. We empirically show that _ScaDLES_ converges up to
$\mathbf{3.29\times}$ faster compared to conventional distributed SGD.

###### Index Terms:

Deep learning, Distributed training, Streaming data, Federated learning,
Adaptive compression

††publicationid: pubid: 978-1-6654-8045-1/22/$31.00 ©2022 IEEE

##  I Introduction

With the advent of big data and IoT, the number of smart devices has grown
exponentially over the years. These devices capture data across a wide range
of modalities, such as image/video in smartphones and surveillance camera
feeds, audio and speech from smart speakers, text/language on phone/tablet
keyboards etc. The data collected on the devices can either be moved to a
centralized server in the cloud or persist locally. Local storage is practical
when network bandwidth is limited and data privacy is a concern.

Locally storing data presents its own challenges due to limited capacity on
edge/fog devices. The problem is exacerbated on devices with high-inflow
streaming data. The data lifetime is also influenced by device streaming rates
as high-volume streams may require more frequent storage purge or handling via
other means. Commercial solutions offer data storage in the cloud for finite
time, but this violates data privacy, incurs high communication cost of data
movement and the subscription-based cost to store that data. As a result, data
residing on the edge ends up being flushed that could otherwise be used for
training models. It is thus prudent to develop online training systems to
leverage transient data with a limited lifespan.

_Distributed deep learning (DDL)_ typically assumes centralized data, where
each process/device samples training data in an IID fashion at every
iteration. However, this is not necessarily true for streaming data which can
be skewed not just in volume, but can be unbalanced and have non-IID
distribution as well. Another consequence of training on devices with varying
flow-rates is that high-inflow devices may have to wait on low-inflow ones
until they gather enough samples corresponding to the mini-batch set for
training. Thus, devices with low-volumes of streaming data can essentially be
perceived as _stragglers_ that slow down distributed training.

In DDL, gradients computed locally are aggregated into a global update which
is propagated back to the devices before proceeding to the next iteration. The
size of the gradients communicated is of the same scale as the number of
trainable parameters in the network, which can span over hundreds of millions
or even billions for modern language and vision models. Using single-precision
($32$-bit) floats to represent gradients means that hundreds of megabytes or
even gigabytes of data needs to be exchanged at every iteration. Thus,
heterogeneity in data inflow among devices, unbalanced-ness in device-local
data, finite memory/storage and limited bandwidth violate assumptions of
conventional distributed training designed for HPC and cloud.

In this paper, we build a streaming-based distributed training framework
cognizant of the aforementioned issues that we call _ScaDLES:_ {Sca}lable
{D}eep {L}earning over {S}treaming data at the {E}dge. _ScaDLES_ is designed
to train across devices with heterogeneous volumes of streaming data in an
online manner. Instead of waiting for all workers to accumulate enough samples
corresponding to the mini-batch, we choose a variable min-batch size for each
device based on its streaming rate. As a result, there is no additional wait-
time on account of low-volume devices. To aggregate gradients across workers,
we perform weighted aggregation such that a device with a larger batch size is
weighted more than those with smaller batches. We empirically show how this
weighted gradient aggregation approach converges faster than typical
distributed SGD.

To tackle the issues of limited memory, storage and expensive disk IO, we
compare two simple data storage policies: Stream Persistence and Truncation.
We simulate streams and implement these policies with Apache Kafka [1], a
popular distributed stream processing platform.

Lastly, to deal with limited bandwidth and high communication cost of gradient
reduction, we propose an adaptive compression technique where we scale the
compression ratio based on gradient variance and adjusting to critical regions
[4] in the training phase. We apply this adaptive method on Top-k gradient
sparsification [5]. _ScaDLES_ works in an online, black-box manner that we
validate by simulating streams with different degrees of heterogeneity, both
on IID and non-IID data, and compare performance with conventional distributed
SGD.

##  II Challenges in Streaming DL

DDL training on streaming data presents unique challenges that can severely
impact training time and/or convergence quality. Using data streams simulated
on Kafka and Pytorch’s [6] distributed data-parallel [7] module, we observe
the effects of heterogeneous streams, skewness in training data, limited
memory/storage and communication cost of synchronizing model updates on the
overall training time and model convergence.

###  II-A Heterogeneity in device streaming rates

In conventional DDL, multiple devices train a local model replica on
partitions sampled from the entire training dataset, and aggregate gradient
updates at the end of each iteration either via parameter servers [8] or
Allreduce using communication libraries like Open MPI [9] and NCCL [10]. With
distributed SGD, parameter update $\mathit{w}$ at iteration $(\mathit{t}+1)$
for $\mathit{N}$ devices optimizing loss function
$\mathit{\mathcal{L}(\cdot)}$ on a sample $\mathit{x_{i}}$ of size
$\mathit{b_{i}}$ from distribution $\mathcal{X}_{i}$ and learning rate
$\mathit{\eta}$ is given by Eqn. (1).

$w_{t+1}=w_{t}-\eta\dfrac{1}{N}\sum_{n=1}^{n=N}{\dfrac{1}{|b_{i}|}\sum_{i\in
b_{i}}\dfrac{\partial}{\partial w_{t}}\mathcal{L}(x_{(i,n)},w_{t})}$ (1)

Each device trains on the same mini-batch size $\mathit{b}$, making global
batch-size $\mathit{N\cdot b}$. However, when dealing with streaming data,
devices can have different streaming rates. Devices with high-volume streams
can readily collect $\mathit{b}$ samples, while those with sparse inflow rates
need to wait until samples equal to $\mathit{b}$ are collected. With an inflow
rate of $\mathit{p}$ samples/sec., a device would have to wait about
$\mathit{(b/p)}$ seconds before proceeding to perform forward-backward pass.
We consider such variances in streaming rates among devices as _streaming
heterogeneity_. Heterogeneity can be inter or intra-device as well; the
streaming rate on a device itself can vary based on traffic, usage, time of
day, etc. To understand how streaming heterogeneity can affect wall-clock time
due to latency incurred while gathering a mini-batch, we sample streaming
rates from different distributions and compute the latency incurred to collect
different batch sizes.

TABLE I: Devices sampled with varying streaming rates Distribution | Sample set | Mean | Std. Dev.
---|---|---|---
Uniform | $\mathit{S_{1}}$ | 38 | 24
$\mathit{S_{2}}$ | 300 | 112
Normal | $\mathit{S_{1}^{\prime}}$ | 64 | 24
$\mathit{S_{2}^{\prime}}$ | 256 | 28

In Table I, we use two sets each of uniform and normal distribution to sample
streaming rates for devices. These two distributions capture inflow
heterogeneity that we typically expect to see in real-world settings. Uniform
distribution samples evenly across a given range, thus giving more
heterogeneous streaming rates. On the other hand, rates sampled in normal
distribution are centered around the mean so it resembles a more homogeneous
setting w.r.t the device streaming rates. Sets $\mathit{S_{1}}$ and
$\mathit{S_{1}^{\prime}}$ have a smaller mean as well as variance, while
$\mathit{S_{2}}$ and $\mathit{S_{2}^{\prime}}$ represent a higher mean and
larger standard deviation. $[\mathit{S_{2}},\mathit{S_{2}^{\prime}}]$ denote
higher streaming rates compared to $[\mathit{S_{1}},\mathit{S_{1}^{\prime}}]$.

The batch size is an important _hyperparameter_ in deep learning, i.e., a
factor that influences convergence in neural networks. A small batch size
cannot be efficiently parallelized, while a very large batch size increases
generalization error [11]. For now, we don’t take these considerations into
account and only see the latency incurred to gather different batch sizes when
we sample streaming rates from the described distributions. Fig. 1 shows the
streaming latency across each set for different batches. Latency increases
with larger batches as more training samples need to be collected. Thus, the
device with the lowest streaming rate (and maximum latency) effectively
becomes a straggler in synchronous training as other devices wait on it to
gather a mini-batch, perform computation and send its local gradients for
reduction.

(a) Latency in $\mathit{S_{1}}$ and $\mathit{S_{2}}$

(b) Latency in $\mathit{S_{1}^{\prime}}$ and $\mathit{S_{2}^{\prime}}$

Figure 1: Streaming latency across batches when device stream-rates are
sampled from different distributions.

###  II-B Data skewness in deep learning

While collaboratively training models, data on a device can be skewed either
in volume, properties, or both. For unbalancedness due to volume, imagine a
traffic surveillance system where devices capture identically distributed data
like frames of individual vehicles (car, bike, trucks, etc.), but the volume
of data on each device varies with the traffic density on the route where the
camera is installed. Skewness due to data properties is introduced when the
distribution of device-local data varies significantly from the overall data
distribution. For example, a vehicle recognition model running on a video
surveillance system installed in a subway captures images of trains, while
devices installed near the airport runways covers flying vehicles only. Thus,
training data has non-IID distribution as it has partial labels only (like a
train or a plane). Privacy-sensitivity and large volumes of data on
constrained networks make it unfeasible to move it to a centralized location
like the cloud. We train two popular image classifers: ResNet152 [12] and
VGG19 [13] on skewed data to observe the impact of unbalanced and non-IID
distribution on convergence. We induce non-IID distribution of CIFAR10 and
CIFAR100 [14] by mapping a subset of labels to a unique device. We train on 10
devices for CIFAR10 such that a single label resides on one device, while we
train CIFAR100 on 25 devices by mapping 4 labels to a single device. Using
Top-5 test accuracy as the performance metric, Fig. 2a shows the result of
training ResNet152 on CIFAR10 and VGG19 on CIFAR100. For comparison, we also
show the corresponding performance of training with data partitioned in an IID
manner. The model quality degrades considerably on non-IID data for both
models and datasets.

(a) Training with IID and non-IID data

(b) Memory util. vs. batch-size

Figure 2: Data skewness and limited memory on the edge

###  II-C Limited Memory and storage

With high volumes of streaming data, further processing and storage can be
costly or even unfeasible due to limited physical resources. Limited memory
presents challenges even in data-center settings where GPU memory is
significantly lower than system memory. Training a neural network on a GPU
requires storing model parameters (a.k.a weights), gradients computed in
backward pass, activation maps as well as training batches. Fig. 2b and 3a
show how GPU memory utilization varies on NVIDIA V100 GPUs based on the mini-
batch size and the kind of optimization used. Keeping all other
hyperparameters fixed, memory usage increases in a near-exponential fashion
with batch size (Fig. 2b). From Fig. 3a, memory consumption also increases as
we move from mini-batch SGD to Nesterov’s momentum [15], and then to Adam
optimizer [16]. Nesterov’s momentum needs more memory than mini-batch SGD
since it keeps parameter updates from the previous timestep as well. Adam
optimizer consumes even more memory since it stores both first and second
order updates from previous timestep. Thus, such heavy resource requirements
make training at the edge even more challenging.

(a) Memory varies with SGD variant

(b) Queue size grows over time

Figure 3: Memory and storage constraints with DDL on streaming data

Since neural network training is compute-intensive, it is difficult to train
models on streaming data at _line-rate_. High synchronization overhead to
aggregate updates further inhibits linear scaling of DDL. As a result, _data
quickly accumulates if the streaming rate is higher than the processing rate_.

The size of streaming queues can quickly blow up in distributed training, be
it on-disk or in-memory. We formulate this as follows: suppose each device
$\mathit{D_{i}}$ among $[\mathit{D_{1}},\mathit{D_{2}},....\mathit{D_{n}}]$
devices has a fixed streaming rate of $\mathit{S^{(i)}}$ samples/second. The
devices collaboratively train a model with average batch-size $\mathit{b_{i}}$
(i.e., $\mathit{b_{i}}=\sum_{j=1}^{j=n}\mathit{b_{j}}/n$). Consider a scenario
where a devices’ streaming rate is larger than the training batch-size, i.e.,
$\mathit{S^{(i)}>b_{i}}$. A single iteration in distributed training involves
calculating loss, computing gradients, aggregate and apply updates; let’s
denote this time on device $\mathit{i}$ as $\mathit{t_{i}}$. At initial
timestep $\mathit{ts}=0$, $\mathit{S^{(i)}}$ samples arrive each second at
$\mathit{i}$ which then processes $\mathit{b_{i}}$ samples from it. In the
time $\mathit{t_{i}}$ that $\mathit{i}$ completes one training iteration,
about $\mathit{(t_{i}\cdot S^{(i)})}$ more samples arrive in addition to the
residual $\mathit{(S^{(i)}-b_{i})}$ that weren’t used. Thus, there are
$\mathit{(S^{(i)}-b_{i})}+\mathit{t_{i}}\cdot S^{(i)}$ samples enqueued in the
streaming buffer at timestep $\mathit{ts}=1$. At timestep $\mathit{ts}=2$,
there are $2\mathit{(t_{i}+1)S^{(i)}}-2\mathit{b_{i}}$ samples in the buffer.

The queue size increases over time on account of residual samples from
previous timesteps. We generalize the number of accumulated samples
$\mathit{Q_{i}}$ on device $\mathit{i}$ after $\mathit{T}$ timesteps in Eqn.
(2) and note that $\mathit{Q_{i}}$ scales linearly with $\mathit{T}$.

$\mathit{Q_{i}}=\mathit{(t_{i}\cdot S_{i}-b_{i})\cdot
T+S^{(i)}}\quad\forall\quad t_{i}\cdot S^{(i)}\geq b_{i}$ (2)

As a timestep corresponds to an iteration in DDL, buffer size can increase
dramatically when $\mathit{T}$ is large, which is typical for neural networks
to run for thousands of iterations. To limit $\mathit{Q_{i}}$ from blowing up,
one could argue to set $\mathit{b_{i}}$ to $\mathit{t_{i}\cdot S^{(i)}}$. In
that case, $\mathit{Q_{i}}$ is always equal to $\mathit{S^{(i)}}$ irrespective
of the value of $\mathit{T}$ . However, $\mathit{b_{i}}$ is a hyperparameter
that may require careful tuning. Using $\mathit{t_{i}\cdot S^{(i)}}$ as batch
size can be impractical if the streaming rate is too high or too low. A small
batch-size doesn’t leverage parallelism while a large $\mathit{b_{i}}$ would
hurt generalization performance.

_Assuming high streaming rates and considerable iteration times in DDL due to
limited compute and bandwidth at the edge, Eqn. ( 2) reduces to_

$\mathit{Q_{i}=(T\cdot t_{i}\cdot
S^{(i)}+S^{(i)}})\quad\text{if}\quad\mathit{(t_{i}\cdot S^{(i)}})\gg b_{i}$
(3)

We simulate how $\mathit{Q_{i}}$ increases with $\mathit{T}$ as stated in Eqn.
(3). The results are illustrated for different $\mathit{tS}$ values in Fig.
3b. The y-axis takes the log (with base $10$) of the samples accumulated. As
$\mathit{tS}$ increases, so does the corresponding buffer size. We note that
when $\mathit{tS\approx 0}$, the buffer only holds $\mathit{S}$ samples at any
given time. Such a setting describes a hypothetical system where the total
iteration time is negligible regardless of the streaming rate.

To gauge buffer requirements with streaming data in real-world settings, we
measure the space needed to store $[32\times 32]$ colored images for training
ResNet152 and VGG19. For mini-batch size $64$, the models have average
iteration times of $1.2$ and $1.6$ seconds respectively. Table II tabulates
how training samples accumulate after $1K,10K,100K$ timesteps for these
iteration times. As $\mathit{T}$ increases, so does the storage requirements
to hold the data. Optimized stream processing platforms like Kafka reduce
memory footprint by storing messages on-disk as partitions, and then deleting
the data based on some retention policy once the messages are successfully
consumed. However, persisting data on-disk becomes unfeasible especially on
the edge as data keeps accumulating with the iterations.

TABLE II: Data accumulated over streaming in DDL Model | $\mathbf{\mathit{t}}$ | $\mathit{S}$ | Data accumulated at $\mathit{T}$ steps (GB)
---|---|---|---
| (s) | (img/s) | $\mathit{T}=10^{3}$ | $\mathit{T}=10^{4}$ | $\mathit{T}=10^{5}$
ResNet152 | $1.2$ | 100 | 0.35 | 3.5 | 34.33
600 | 2.06 | 20.6 | 200.6
VGG19 | $1.6$ | 100 | 0.47 | 4.69 | 46.8
600 | 2.75 | 27.5 | 274.83

###  II-D Synchronization overhead

(a) Gradient synchronization time

(b) Relative throughput increase

Figure 4: Synchronization overhead and scaling limitations in DDL

Although training neural networks on GPUs can significantly reduce computation
time, DDL scaling is still limited by periodic gradient synchronization.
Communication time increases with model size, as shown in Fig. 4a for
Transformer [17], ResNet152 and VGG19. Training ResNet152 and VGG19 on 8
NVIDIA K80s takes about $80$-$90\%$ of the total iteration time in gradient
synchronization. Additionally, communication cost tends to increase with the
number of participating devices. Increasing the network bandwidth brings down
the synchronization cost to only a certain extent and saturating thereafter
[3]. In an ideal setting with no communication latency, scaling a job to
$\mathcal{P}$ workers should improve overall throughput by $\mathcal{P}$ as
well. However, Fig. 4b shows that application throughput does not increase
linearly, ResNet152 and VGG19 running on $16$ K80s observed only about
$5\times$ and $4\times$ more throughput relative to single GPU training.

##  III Background and Related Work

###  III-A Handling unbalanced and non-IID data

Neural network training under constraints like privacy-sensitivity, skewness
due to non-IID and unbalanced data has been well studied under the premise of
_federated learning_. In federated training, devices train on local, skewed
data while a global shared model is learned by periodically aggregating
updates from other devices. For example, FedAvg [20] collects updates only
from a fraction of total clients after certain local epochs to reduce frequent
communication cost. FedProx [23] extends FedAvg to include partial work (to
address systems heterogeneity) and adds a proximal term to the local objective
function to deal with statistical heterogeneity. [2] addresses systems
heterogeneity by sampling batch-size based on a worker’s compute capability.
Sparse Tenary Compression (STC) [19] reduces communication by combining
Terngrad [24] quantization with Top-k [25] sparsification. Zhao et al. [18]
account for data skewness due to weight divergence among devices’ local model
replica and measure it with earth mover’s distance (EMD) between device-local
and overall data distribution. Benchmarks like FedML [21] facilitate
development of federated learning systems and algorithms.

###  III-B Dealing with limited memory

As neural networks have grown in size over the years, so has their resource
requirements. The memory space to hold a model comprises of trainable
parameters and a computation graph to store gradients and activation maps
computed in forward-backward pass. Micikevicius, et al. [26] proposed
automatic mixed-precision (AMP) training with half-precision (16-bit) floating
points to reduce the memory footprint by half. Gradient checkpointing [27]
trades memory for computation by flushing intermediate data from the
computation graph to reduce memory utilization. This comes at the cost of
increased computation as flushed activation maps need to be recomputed
whenever needed. Memory consumption can be reduced by decreasing the training
batch size as well. However, this reduces parallelizability and increases the
overall training time. Deep learning frameworks like PyTorch implement
torchvision.datasets.DatasetFolder and ImageFolder to avoid loading entire
training data to memory by reading samples from disk one batch size at a time.
But these dataloaders adhere to a rigid format for specifying labels in the
underlying directory structure. Thus, training a model on a particular dataset
may require considerable preprocessing and designing custom dataloaders. For
massive datasets, on-disk storage can quickly blow up too as seen previously.

###  III-C Reducing communication cost

Algorithms like FedAvg and FedProx minimize communication overhead by choosing
a low-frequency, high-volume communication strategy with occasional gradient
synchronization. On the other hand, gradient compression either via
sparsification, quantization or low-rank approximations uses a high-frequency,
low-volume approach to reduce communication cost in DDL. Sparsification
techniques like Topk-k [25] and Deep Gradient Compression (DGC) [28] apply
sparse updates by sending only a subset of the gradients and setting remaining
values to 0. The bit-width of floating-point gradients is reduced with
quantization methods. Automatic mixed-precision (AMP) training described
earlier uses half-precision gradients to achieve $2\times$ compression.
Another quantization method called Quantized SGD (QSGD) [29] quantizes
gradients while balancing the trade-off between precision and accuracy. On the
other hand, Terngrad [24] limits gradients across three quantization levels
[-$1$,$0$,+$1$]. All these compression techniques use a fixed compression
ratio throughout training. Using a high compression ratio incurs higher
communication cost, while a small compression ratio may trim too much useful
information from the gradients. Accordian [31] dynamically compresses
gradients by detecting critical regions in training by tracking gradient
variance. We extend this further by developing an adaptive compression
strategy by comparing entropy loss between the original and compressed
gradients.

##  IV scadles

We propose _ScaDLES_ to address the challenges described in section II and
accelerate DDL training on heterogeneous streams in both IID and non-IID
settings. The bird’s eye view shown in Fig. 5 is described as follows:

Figure 5: Per-iteration routine in _ScaDLES_ for two devices

Heterogeneous streams: The approaches described w.r.t federated training in
section III consider either systems heterogeneity or statistical heterogeneity
due to skewed and non-identical data. Training neural networks synchronously
on multiple devices with heterogeneous data streams suffers from stragglers. A
device $\mathit{i}$ among $\mathit{n}$ devices with the lowest streaming rate
$\mathit{S^{(i)}\in[S^{(1)},S^{(2)},...S^{(n)}]}$ can become a bottleneck
depending on the mini-batch size since all other devices have to wait on
$\mathit{i}$ to gather enough training samples $\mathit{b_{i}}$ and proceed an
iteration. Additionally, streaming rates can vary at intra-device level at the
edge too, depending on factors like battery level, time of day, usage, etc.
This wait-time incurred due to streaming latency can thus slow down training.

To mitigate the impact of streams with lower inflows, we propose performing
variable computation where we set $\mathit{b_{i}\>\forall\>i\propto S^{(i)}}$.
Thus, we minimize streaming latency by setting device batch size to its
streaming rate. As a result, some devices with high volume streams train on a
large batch-size while the low volume devices use a smaller batch-size, and
wait-times due to streaming latency are avoided. Since the amount of work done
on each device is different, we perform weighted aggregation rather than a
simple average to get the shared global updates. At iteration $\mathit{t}$,
device $\mathit{i}$ trains with batch-size corresponding to its streaming rate
$\mathit{S_{i}^{(t)}}$ and scales the computed gradients
$\mathit{g_{i}^{(t)}}$ by factor $\mathit{r_{i}^{(t)}}$ and updates the
parameters as:

$r^{(i)}_{t}=\dfrac{S^{(i)}_{t}}{\sum_{j=1}^{n}S^{(j)}_{t}}\quad:\>\>\sum_{j=1}^{n}r^{(j)}_{t}=1.0$
(4a) $\tilde{g}_{t}=\sum_{j=1}^{n}r^{(j)}_{t}\cdot g^{(j)}_{t}$ (4b)
$w_{t+1}=w_{t}-\eta_{scaled}\cdot\tilde{g}_{t}$ (4c)

The global batch-size with weighted gradients from Eqn. 4a is
$\mathit{\sum_{j=1}^{n}S^{(j)}}$. As streaming rates can vary both inter and
intra-device, so does the global batch-size. To ensure extremely high
streaming rates don’t increase the global batch-size so much that it degrades
generalization performance, we add a _linear scaling rule_ as suggested in
[32, 33]. Essentially, linear scaling adjusts the learning rate in proportion
to the batch-size, i.e., learning rate is increased if the batch-size
increases, and vice versa. When the batch-size is multiplied by factor k,
multiply the base learning rate by k as well. If the base global batch-size is
$\mathit{B}$, then we scale the base learning rate as

$\eta_{scaled}=\gamma_{scaled}\cdot\eta\;\;:\>\gamma_{scaled}=\dfrac{\sum_{j=1}^{n}S_{j}}{B}$

Even with a linear-scaling rule for training with larger batches, model
quality still may suffer with extremely large batches in high volume streams.
Likewise, using a batch-size too small is not efficient from parallelization
perspective. Thus, we set $\mathit{b_{i}=S^{(i)}}$ as long as the device
batch-size is bounded in the range $\mathit{b_{min}\leq b_{i}\leq b_{max}}$,
else we use the corresponding min-max for training.

Limited memory and storage: The buffer size can grow quickly due to continuous
data streams and considerable iteration times at the edge. The accumulated
data can either reside in memory like a buffered queue, and reside on-disk to
reduce memory footprint like in Kafka. By default, we could keep all the data
streaming in and store it until processed successfully. We refer to this
policy as _Stream Persistence_. As seen from Table II, accumulated samples
keep increasing over the iterations depending on the stream-rate. Looking at
Eqn. 2, buffer size grows to $\mathcal{O}(\mathit{S^{(i)}T})$ after
$\mathit{T}$ iterations. Stream persistence makes sense especially when
devices have sufficient memory or storage to hold the data, like in data-
centers, cloud or high-capacity fog devices. In _Stream Truncation_ , we
discard the residual samples and hold just enough data corresponding to the
device’s streaming rate $\mathit{S^{(i)}}$. As a result, storage requirements
for stream truncation is $\mathcal{O}(\mathit{S^{(i)}})$ at any given time,
which is significantly smaller than stream persistence.

High communication cost: Federated algorithms reduce communication cost either
with low-frequency, high-volume or high-frequency, low-volume communication
strategy. We focus our efforts on the latter by looking at various gradient
compression techniques. Rather than using a static compression ratio
throughout training that can be detrimental to the final model accuracy, we
look into adaptive compression. Prior work keeps track of the moving average
of gradient variance to detect critical regions in training [30, 31, 34].
Gradients are large initially, but get smaller as the model evolves and
training continues. Thus, we can use low compression in the beginning and
higher compression later. We implement an adaptive compression strategy with
Top-k sparsification which compares entropy loss between compressed and
uncompressed gradients. Gradients compressed to top k% are used if the
variance between compressed and uncompressed tensors falls below threshold
$\delta$; original, uncompressed tensors are communicated otherwise. The
intuition is that if the top k% gradients have most of the information as the
uncompressed gradients within the margin of $\delta$, then remaining gradients
are relatively less meaningful that don’t greatly contribute towards model
update and can thus be ignored. We compute variance of both compressed and
uncompressed gradients at each iteration while keeping exponentially weighted
moving average (EWMA) to implement the communication rule for adaptive
compression on gradients $\mathit{g}$ for a device as follows:

$\text{send}(\text{Top}\mathit{k(g)})\;\;\text{if}\;\;\dfrac{||g|^{2}-|\text{Top}\mathit{k(g)}|^{2}|}{|g|^{2}}\leq\delta\;\;\text{else}\;\;\text{send}(g)$

_Compression threshold,_ denoted by $\delta$ determines the degree of
relaxation we impose on the compressed tensors to be eligible for
communication. A small $\delta$ penalizes compressed tensors more severely and
performs reduction only when the compressed data captures most of the relevant
gradients in the original tensor. Constraints are loose with a larger $\delta$
which allows more iterations to use compressed tensors for synchronization.

Unbalanced and Non-IID data: Fig. 2a demonstrates how model quality degrades
when training with non-identical data. This happens since each device contains
only a subset of training labels that are not representative of the entire
distribution. Thus, parameters learned by device-local model replicas are
skewed, and so is the aggregated model. To deal with non-IID data, we propose
randomized _data-injection_ where a fraction of the training devices share
partial training samples with other devices. Particularly, at every iteration
a device randomly chooses a subset $\alpha$ of the total devices $\mathit{D}$
to share fraction $\beta$ of its streaming data $\mathit{\beta S{(i)\in[\alpha
D]}}$. Together, ($\alpha,\beta$) determine what set of devices share how much
of their training samples with other devices in DDL. Data injection helps
improve the overall data distribution by making the device local data more
representative of the complete dataset. However, this implies a trade-off
between high model quality on account of better data distribution and privacy
concern arising from moving data away from the devices. Privacy violation is
greatly minimized by choosing only a subset of devices randomly (from
$\alpha$) and broadcasting only partial data (determined by $\beta$).

##  V Evaluation

###  V-A Cluster setup

We simulate streaming data with Kafka by sampling streaming rates from uniform
and normal distributions described in Table I. The hardware used to evaluate
_ScaDLES_ in our experiments comprises of $4$ servers each with with $48$-core
Intel Xeon E5-2650, $128$ GB system memory and $8$ NVIDIA K80 GPUs connected
with $5$ Gbps ethernet. We mimic CUDA-capable edge devices by spawning them as
nvidia-docker containers on CentOS linux $7.9.2009$ with docker engine
$20.10.17$. Each device running as a container is allocated $4$vCPUs, $12$ GB
system memory and $1$ K80 GPU running NVIDIA driver $465.19.01$ on CUDA $11.3$
and PyTorch $1.10.1$. We create a docker swarm network on the $5$ Gbps network
interface to facilitate communication for gradient synchronization among
containers.

###  V-B Data, models and hyperparameters

TABLE III: Cluster setup and evaluation on IID and non-IID data Model | Size | Data | Devices | Label/device
---|---|---|---|---
ResNet152 | $60.2M$ | IID Cifar10 | 16 | 10
nonIID Cifar10 | 10 | 1
VGG19 | $143.7M$ | IID Cifar100 | 16 | 100
nonIID Cifar100 | 25 | 4

We evaluate two popular neural networks across different streaming
distributions, training dataset and cluster configurations. ResNet152 uses SGD
optimizer with momentum $0.9$ and weight decay $0.0001$ while adopting a
learning rate schedule with initial lr $0.1$ that decays by $0.2$ after $75$,
$150$ and $225$ epochs. We also train VGG19 momentum SGD of $0.9$ and weight
decay $0.0005$ with an intial lr $0.01$ that decays by $0.3$ after $75$, $150$
and $200$ epochs. The model quality for both neural networks is measured by
the _Top-5 test accuracy_.

Table III shows the cluster size and label distribution for IID/non-IID
CIFAR10/100 on ResNet152 with $60.2$ million and VGG19 with $143.7$ million
parameters respectively. At the end of each epoch, we transform the training
data with a RandomCrop and RandomHorizontalFlip to imitate unique samples
streaming into a device.

###  V-C Streaming data for DDL

We use Apache Kafka v$3.1.0$ to spawn a docker container that runs a broker as
well as producers. The broker-producer container is allocated $16$vCPUs, $32$
GB system memory and _no_ GPU since it doesn’t participate in model training.
We configure the container with $8$ network threads, $4$ IO-threads and $1$
partition per topic. The sole purpose of this container is to host the Kafka
broker and launch multiple producer processes such that each process publishes
to a unique topic corresponding to a device. Thus, there are as many topics as
the number of training devices. Each producer process controls the streaming
rate corresponding to a device’s topic. As for data consumption, the training
devices have a kafka consumer running on them. The consumer implements a
custom PyTorch dataloader that batches the data and integrates into a typical
training loop that is common in deep learning training.

Since we run multiple producers from a single container (and not a separate
container for every producer), we measure the effective streaming rate
achieved with our proposed setup to ensure we meet the target stream-rates in
our experiments. We scale up the number of concurrent producers (i.e., topics)
and measure the observed streaming rate. For e.g., $32$ streams in Fig. 6a
imply $32$ producers publishing to $32$ topics at $100$ samples/sec. Fig. 6
shows the density estimates of observed streaming rates with targets of $100$
and $600$ samples/sec. For each target rate, we scale up the number of
concurrent producers to $1,4,8,16$ and $32$. We achieve nearly the same target
of $100$ samples/sec as shown in Fig. 6a. For the $600$ samples/sec target,
the effective streaming rate decreases noticeably beyond $16$ concurrent
streams. We could likely improve this by increasing the number of network
threads and partitions per topic, but this setup sufficed for the evaluations
we perform in this paper.

(a) 100 samples/sec

(b) 600 samples/sec

Figure 6: Effective streaming rates achieved when scaling to multiple topics.

###  V-D Weighted aggregation in heterogeneous streams

We use a batch-size corresponding to a devices’ streaming rate in _ScaDLES_ to
avoid wait-times on account of possible streaming latency. To enforce bounds
on the batch-size used, we set $\mathit{b_{min}}$ and $\mathit{b_{max}}$ to
$8$ and $1024$ respectively, although stream-rate for any device remains
within this range regardless of the streaming distribution. The waiting time
can especially be long in highly heterogeneous streams when a devices’
streaming rate is lower than the mini-batch size configured prior training.
There is no waiting time high-volume streams with low-batch size settings.
However, the buffer size can grow quickly over time in that case. We compare
_ScaDLES_ with conventional DDL training for batch-size $64$ irrespective of
the device streaming rates. We look at the convergence curves and buffer
buildup over training epochs to compare the two. The streaming rates for the
16 devices were sampled from the distributions outlined in section II. Uniform
distributions are more heterogeneous compared to normal distributions ($2/3$rd
values lie within $1$ standard deviation from the mean in the latter).

Fig. 7a shows results for _ScaDLES_ and conventional DDL using device stream-
rates sampled from $\mathit{S_{1}}$ that converges $3.33\times$ and
$1.92\times$ faster in _ScaDLES_. Conventional DDL converged with higher final
accuracy in $\mathit{S_{2}}$ due to large batches used for training by
_ScaDLES_ with this distribution; about $4.5$K in _ScaDLES_ compared to only
$1$K in DDL. Linearly scaling the learning rate at batches this large did not
significantly improve _ScaDLES_ ’ generalization performance. Devices sampled
from $\mathit{S_{1}^{\prime}}$ achieved around $3.6\times$ and $4\times$
speedup with _ScaDLES_ while still achieving higher final accuracy. Using
larger batches and linear scaling proved to be beneficial in this case.
Lastly, Fig. 7d uses $\mathit{S_{2}^{\prime}}$ distribution where ResNet152
performs similarly for both _ScaDLES_ and DDL, while VGG19 performs better
with our approach. Based on the streaming-rate distributions, _ScaDLES_
achieved same or better accuracy than conventional DDL when the mean stream-
rate or the number of training devices is low/medium. If the mean stream-rate
or the number of participating devices is very high, _ScaDLES_ observed some
accuracy drop on account of generalization gap due to large batch training.

(a) $\mathit{S_{1}}$ distribution

(b) $\mathit{S_{2}}$ distribution

(c) $\mathit{S_{1}^{\prime}}$ distribution

(d) $\mathit{S_{2}^{\prime}}$ distribution

Figure 7: Convergence in conventional DDL vs. _ScaDLES_ ’ weighted aggregation

###  V-E Managing limited memory and storage

We first look at how streaming data gets accumulated in a device with the
default persistence policy. For the same runs described in the previous
section, we plot how samples get accumulated over the iterations for different
sampling distributions in Fig. 8. We plot logarithm of the accumulated samples
with base $10$. The buffer size is smaller in _ScaDLES_ compared to DDL
training for the same persistence policy. This is because _ScaDLES_ uses
batch-size $\mathit{S^{(i)}}$ while we use a smaller batch-size $64$ in
conventional DDL. $\mathit{S_{2}}$ and $\mathit{S_{2}^{\prime}}$ have larger
buffer sizes since they represent higher volume streams compared to
$\mathit{S_{1}}$ and $\mathit{S_{1}^{\prime}}$. DDL occupies $2\times$ and
$3.5\times$ more space with ResNet152 and VGG19 in $\mathit{S_{1}}$. _ScaDLES_
holds $3.6\times$ and $641\times$ less data than DDL for $\mathit{S_{2}}$.
Comparing Fig. 7b and Fig. 8b, we see the lower buffer size in _ScaDLES_ came
at the cost of lower final accuracy due to large-batch training. _ScaDLES_ has
$4.7\times$ and $5\times$ smaller buffer in $\mathit{S_{1}^{\prime}}$, and
$3.9\times$ and $42\times$ lesser data with $\mathit{S_{2}^{\prime}}$
distribution.

(a) $\mathit{S_{1}}$ distribution

(b) $\mathit{S_{2}}$ distribution

(c) $\mathit{S_{1}^{\prime}}$ distribution

(d) $\mathit{S_{2}^{\prime}}$ distribution

Figure 8: Buffer size increases with training iterations

Although _ScaDLES_ accumulates lesser samples than conventional DDL, we can
further lower the buffer size for continuous data streams. With stream
truncation, data in buffer exceeding the samples that just streamed in is
simply discarded. The buffer size with truncation policy is constant as long
as the streaming rate is continuous. On the other hand, persistence policy
grows with each passing iteration. Table IV shows the final buffer size to
reach $95$% and $84$% accuracy on ResNet152 and VGG19. The table also reports
reduction with truncation relative to persistence policy. We observed buffer
reductions from $850\times$ to $9400\times$ depending on the distribution.

TABLE IV: Buffer-size reduction with truncation policy Dist. | Model | Persistence | Truncation | Reduction
---|---|---|---|---
$\mathit{S_{1}}$ | ResNet152 | $2.9\times 10^{5}$ | $129$ | $2238\times$
VGG19 | $1\times 10^{5}$ | $118$ | $848\times$
$\mathit{S_{2}}$ | ResNet152 | $4.36\times 10^{6}$ | $633$ | $6889\times$
VGG19 | $4\times 10^{6}$ | $523$ | $7830\times$
$\mathit{S_{1}^{\prime}}$ | ResNet152 | $6.2\times 10^{5}$ | $143$ | $4340\times$
VGG19 | $3.7\times 10^{5}$ | $129$ | $2861\times$
$\mathit{S_{2}^{\prime}}$ | ResNet152 | $3.6\times 10^{6}$ | $384$ | $9429\times$
VGG19 | $2.5\times 10^{6}$ | $360$ | $6956\times$

###  V-F Data-injection for non-IID and skewed data

(a) ResNet152

(b) VGG19

Figure 9: Different ($\alpha,\beta$) configurations for data-injection in non-
IID training.

Data-injection helps improve overall model quality when dealing with non-IID
data. The degree of data-injection is determined by ($\alpha,\beta$)
parameters that determine the subset of devices to send partial data. We
evaluate four ($\alpha,\beta$) sets in _ScaDLES_ : $(0.5,0.5)$, $(0.25,0.25)$,
$(0.1,0.1)$ and $(0.05,0.05)$. A value of $(0.5,0.5)$ means half of the
devices share half of the samples in their current batch. We plot the
convergence curves for different streaming distributions in Fig. 9 and note
significantly better performance than training merely with non-IID data. Some
additional networking cost is associated with data-injection as a subset of
devices send partial data to other devices. For CIFAR10 and CIFAR100 datasets,
each sample is an image $3$ Kilobytes in size. For different streaming
distributions and ($\alpha,\beta$) parameters, we look at data exchange among
the devices in Fig. 10. The overhead is minimal and ranges anywhere from $150$
to $2000$ kilobytes on average for each training iteration.

(a) $\mathit{S_{1}}$ distribution

(b) $\mathit{S_{2}}$ distribution

(c) $\mathit{S_{1}^{\prime}}$ distribution

(d) $\mathit{S_{2}^{\prime}}$ distribution

Figure 10: Data-injection overhead at each iteration with non-IID data.

###  V-G Adaptive compression

We look at reduction in the overall communication volume, i.e., cumulative
single-precision floats communicated achieving target accuracy to evaluate the
performance of adaptive compression in _ScaDLES_. Compression ratio (CR)
measures the degree of compression by comparing the tensor size of compressed
gradients to that of the original tensors. CR of $0.1$ means compressed
tensors are $1/10$-th the size of uncompressed gradients. The update size of
ResNet152 ($230$ MB) and VGG19 ($548$ MB) at this CR reduces to just $23$ and
$55$ MB respectively.

Using the adaptive compression rule described in section IV, we measure the
usage of compressed gradients for communication with Compression-to-No-
Compression (CNC) ratio. CNC ratio compares iterations using compressed
gradients for communication to the total iterations used throughout training.
The latter includes the iterations that use compression as well iterations
that use the original, uncompressed gradients for communication:

$\text{CNC ratio}=\frac{T_{compressed}}{T_{compressed}+T_{uncompressed}}$

CNC ratio of $0$ means that compressed tensors _were not_ used even for a
single iteration, while CNC of $1.0$ implies all training iterations used
_only_ the compressed tensors for exchange since there are no iterations that
used the original gradients for communication. To measure the impact of
adaptive compression on model convergence, we tabulate the CNC ratio, accuracy
and overall reduction in communication volume for different (CR, $\delta$)
configurations in Table V. ResNet152 running with CR $0.1$ and any $\delta$
beyond $0.2$ is faster as it converges by exchanging fewer floats. A $\delta$
of $0.1$ barely used any compression for the two CRs in either of the models.
The most communication efficient configuration for ResNet152 used CR $0.01$
and $\delta$ $0.4$, although it results in slightly lower test accuracy. The
pattern of low communication overhead accompanied with degradation in final
accuracy was also observed in VGG19 for CR $0.1$ and $\delta$ $0.4$. The CNC
ratio is high in both cases implying compression is enabled for most
iterations and thus, accuracy drop can be attributed to model degradation
commonly associated with compression. An interesting observation in VGG19
using CR $0.01$ is the communication volume is same across all $\delta$ values
and the total floats exchanged is the same as training without any
compression. This means the adaptive strategy is not using compression in this
configuration, which is further corroborated by the CNC ratio that is $0$
across all $\delta$.

TABLE V: communication reduction in adaptive compression Model | CR | $\delta$ | CNC | Accuracy | Floats sent
---|---|---|---|---|---
ResNet152 | $0.1$ | $0.1$ | $0.29$ | 97.55% | $4.43\times 10^{11}$
$0.2$ | $0.99$ | 96.81% | $0.56\times 10^{11}$
$0.3$ | $1.0$ | 98.41% | $0.4\times 10^{11}$
$0.4$ | $1.0$ | 98.57% | $0.4\times 10^{11}$
$0.01$ | $0.1$ | $0$ | 97.39% | $6.02\times 10^{11}$
$0.2$ | $0.17$ | 97.47% | $4.99\times 10^{11}$
$0.3$ | $0.43$ | 96.72% | $2.56\times 10^{11}$
$0.4$ | $0.99$ | 94.97% | $6.32\times 10^{8}$
VGG19 | $0.1$ | $0.1$ | $0$ | 85.45% | $1.3\times 10^{12}$
$0.2$ | $0.08$ | 84.74% | $1.19\times 10^{12}$
$0.3$ | $1.0$ | 81.91% | $1.3\times 10^{10}$
$0.04$ | $1.0$ | 81.78% | $1.3\times 10^{10}$
$0.01$ | $0.1$ | $0$ | 84.68% | $1.3\times 10^{12}$
$0.2$ | $0$ | 83.98% | $1.3\times 10^{12}$
$0.3$ | $0$ | 83.94% | $1.3\times 10^{12}$
$0.4$ | $0.004$ | 84.39% | $1.29\times 10^{12}$

###  V-H Overall performance of ScaDLES

Last, we look at the overall performance gains in _ScaDLES_ by combining
weighted aggregation in heterogeneous streams, data-injection for non-IID
data, buffer reduction with stream truncation and reducing communication with
adaptive compression (using CR $0.1$ and $\delta$ of $0.3$ in our final
evaluation). We compare against conventional DDL with fixed batch-size $64$,
persistence policy and the same training schedule as _ScaDLES_ described in
section V.

For the same streaming distribution, _ScaDLES’_ performance is measured
relative to conventional DDL in terms of drop in test accuracy, reduction in
buffer size using truncation policy (in Gigabytes) and overall training
speedup to convergence w.r.t wall-clock time. A negative accuracy drop means
_ScaDLES_ achieved lower accuracy by that margin. Table VI shows the results
for IID training. ResNet152 trains on _ScaDLES_ with a maximum of drop of
$0.32\%$ in final model accuracy compared to conventional DDL with stream-
rates sampled from $\mathit{S_{2}}$. Training with _ScaDLES_ is also much
faster; ranging from $1.15\times$ to $3.29\times$ faster than DDL. For high-
volume streams $\mathit{S_{2}}$ and $\mathit{S_{2}^{\prime}}$, truncation
policy saves up to $5.9$ GB in the occupied buffer-size. _ScaDLES_ achieved
lower final accuracy in VGG19 by as much as $4.18\%$. We observed that VGG19
is more sensitive to the combination of _ScaDLES_ ’ large-batch training and
adaptive compression than ResNet152. However, VGG19 still reduces buffer-size
by up to $3.91$ GB and reduces wall-clock training time by $1.56\times$ to
$2.83\times$ over conventional DDL. As for training with non-IID data,
conventional DDL was unable to reach the same convergence targets as _ScaDLES_
’ data-injection strategy for either of the neural networks. Conventional DDL
with non-IID data saturated ResNet152 at $56\%$ test accuracy, while VGG19 did
not improve beyond $35\%$. For the same non-IID training, _ScaDLES_ achieved
at least $93.6\%$ and $77.8\%$ accuracy for ResNet152 and VGG19 across the
four stream-rate distributions.

TABLE VI: scadles’ performance gains over conventional ddl Model | Dist. | Acc. drop | Buffer red. (GB) | Speedup
---|---|---|---|---
ResNet152 | $\mathit{S_{1}}$ |  $-0.06$% | $0.6$ | $\mathbf{1.89\times}$
$\mathit{S_{2}}$ |  $-0.32$% | $5.9$ | $\mathbf{1.15\times}$
$\mathit{S_{1}^{\prime}}$ |  $-0.13$% | $0.8$ | $\mathbf{3.29\times}$
$\mathit{S_{2}^{\prime}}$ |  $-0.21$% | $4.03$ | $\mathbf{1.42\times}$
VGG19 | $\mathit{S_{1}}$ |  $-1.93$% | $0.26$ | $\mathbf{1.56\times}$
$\mathit{S_{2}}$ |  $-4.18$% | $3.91$ | $\mathbf{2.83\times}$
$\mathit{S_{1}^{\prime}}$ |  $-2.03$% | $0.35$ | $\mathbf{2.06\times}$
$\mathit{S_{2}^{\prime}}$ |  $-1.59$% | $2.58$ | $\mathbf{2.13\times}$

##  VI conclusion

This paper presents the notion of distributed training over streaming data at
the edge, which is challenged by parallel and statistical heterogeneity.
_ScaDLES_ addresses the problem of heterogeneous streaming rate among devices
with weighted aggregation where each device trains on the samples accumulated
in the stream and avoids wait-time or large buffer accumulation. Since it is
difficult to perform training at line-rate, samples streaming into the device
can accumulate over time. _ScaDLES_ uses a simplistic truncation policy to
keep the buffer size in check. Devices on the edge commonly have non-IID and
unbalanced data. Data-injection strategy improves model convergence
significantly in such scenarios. Lastly, we propose an adaptive compression
technique to deal with limited bandwidth on the edge and high communication
cost in large deep learning models. We simulate different degrees of streaming
heterogeneity by sampling from both uniform and normal distributions and
evaluate popular image classifiers. Our empirical evaluation shows that
_ScaDLES_ can converge anywhere from $1.15\times$ to $3.29\times$ faster than
DDL. At the same time, _ScaDLES_ reduces the number of accumulated samples in
the buffer by $848\times$ to $9429\times$.

## References

  * [1] J. Kreps, N. Narkhede and J. Rao, Kafka: a Distributed Messaging System for Log Processing.
  * [2] Tyagi S., Sharma P. , Taming Resource Heterogeneity In Distributed ML Training With Dynamic Batching. 2020 IEEE International Conference on Autonomic Computing and Self-Organizing Systems (ACSOS).
  * [3] Zhen Z., Chaokun C., Haibin L., Yida W., Raman A., and Xin J. Is Network the Bottleneck of Distributed Training? In Proceedings of the Workshop on Network Meets AI and ML (NetAI ’20).
  * [4] Achille A., Rovere M., Soatto S. (2017), Critical Learning Periods in Deep Neural Networks. ArXiv, abs/1711.08856.
  * [5] Dan A., Torsten H., Mikael J., Sarit K., Nikola K., and Cédric R.. 2018. The convergence of sparsified gradient methods. In Proceedings of the 32nd International Conference on Neural Information Processing Systems (NIPS’18).
  * [6] Adam P., Sam G., Francisco M., Adam L., James B., Gregory C., Trevor K., Zeming L., Natalia G., Luca A., Alban D., Andreas K., Edward Y., Zach D., Martin R., Alykhan T., Sasank C., Benoit S., Lu Fang, Junjie B., and Soumith C.. 2019. PyTorch: an imperative style, high-performance deep learning library. Proceedings of the 33rd International Conference on Neural Information Processing Systems.
  * [7] Shen L., Yanli Z., Rohan V., Omkar S., Pieter N., Teng L., Adam P., Jeff S., Brian V., Pritam D., and Soumith C.. 2020. PyTorch distributed: experiences on accelerating data parallel training. Proc. VLDB Endow.
  * [8] Li M., Andersen D.G., Park J.W., Smola A., Ahmed A., Josifovski V., Long J., Shekita E.J. and Su B. (2014). Scaling Distributed Machine Learning with the Parameter Server. BigDataScience ’14.
  * [9] Gabriel E., Fagg G.E., Bosilca G., Angskun T., Dongarra J.J., Squyres J.M., Sahay V., Kambadur P., Barrett B.W., Lumsdaine A., Castain R.H., Daniel D.J., Graham R.L., and Woodall T.S. (2004). Open MPI: Goals, Concept, and Design of a Next Generation MPI Implementation. PVM/MPI.
  * [10] NCCL: NVIDIA Collective Communications Library. Available here: https://docs.nvidia.com/deeplearning/nccl/user-guide/docs/index.html.
  * [11] Keskar N. S., Mudigere D., Nocedal J., Smelyanskiy M. and Tang P. T. B., On Large-Batch Training for Deep Learning: Generalization Gap and Sharp Minima, ICLR, 2017.
  * [12] He K., Zhang X., Ren S. and Sun J. Deep Residual Learning for Image Recognition. 2016 IEEE Conference on Computer Vision and Pattern Recognition (CVPR), 770-778.
  * [13] Simonyan K. and Zisserman A. (2015). Very Deep Convolutional Networks for Large-Scale Image Recognition. CoRR, abs/1409.1556.
  * [14] CIFAR-10 and CIFAR-100: https://www.cs.toronto.edu/ kriz/cifar.html.
  * [15] Nesterov Y. (1983). A method for solving the convex programming problem with convergence rate $O(1/k^{2})$. Proceedings of the USSR Academy of Sciences, 269, 543-547.
  * [16] Kingma D.P., and Ba J. (2015). Adam: A Method for Stochastic Optimization. CoRR, abs/1412.6980.
  * [17] Vaswani A., Shazeer N.M., Parmar N., Uszkoreit J., Jones L., Gomez A.N., Kaiser L. and Polosukhin I. (2017). Attention is All you Need. ArXiv, abs/1706.03762.
  * [18] Zhao Y., Li M., Lai L., Suda N., Civin D. and Chandra V. (2018). Federated Learning with Non-IID Data. ArXiv, abs/1806.00582.
  * [19] Sattler F., Wiedemann S., Müller K. and Samek W. (2020). Robust and Communication-Efficient Federated Learning From Non-i.i.d. Data. IEEE Transactions on Neural Networks and Learning Systems, 31, 3400-3413.
  * [20] McMahan H.B., Moore E., Ramage D., Hampson S. and Arcas B.A. (2017). Communication-Efficient Learning of Deep Networks from Decentralized Data. AISTATS.
  * [21] He C., Li S., So J., Zhang M., Wang H., Wang X., Vepakomma P., Singh A., Qiu H., Shen L., Zhao P., Kang Y., Liu Y., Raskar R., Yang Q., Annavaram M. and Avestimehr S. (2020). FedML: A Research Library and Benchmark for Federated Machine Learning. ArXiv, abs/2007.13518.
  * [22] Wang H., Kaplan Z., Niu D. and Li B. (2020). Optimizing Federated Learning on Non-IID Data with Reinforcement Learning. IEEE INFOCOM 2020 - IEEE Conference on Computer Communications, 1698-1707.
  * [23] Sahu A., Li T., Sanjabi M., Zaheer M., Talwalkar A.S. and Smith V. (2020). Federated Optimization in Heterogeneous Networks. arXiv: Learning.
  * [24] Wen W., Xu C., Yan F., Wu C., Wang Y., Chen Y. and Li H.H. (2017). TernGrad: Ternary Gradients to Reduce Communication in Distributed Deep Learning. ArXiv, abs/1705.07878.
  * [25] Alistarh D., Hoefler T., Johansson M., Khirirat S., Konstantinov N. Renggli C. (2018). The Convergence of Sparsified Gradient Methods. NeurIPS.
  * [26] Micikevicius P., Narang S., Alben J., Diamos G.F., Elsen E., García D., Ginsburg B., Houston M., Kuchaiev O., Venkatesh G. and Wu H. (2018). Mixed Precision Training. ArXiv, abs/1710.03740.
  * [27] Griewank A. and Walther A. (2000). Algorithm 799: revolve: an implementation of checkpointing for the reverse or adjoint mode of computational differentiation. ACM Trans. Math. Softw., 26, 19-45.
  * [28] Lin Y., Han S., Mao H., Wang Y. and Dally W.J. (2018). Deep Gradient Compression: Reducing the Communication Bandwidth for Distributed Training. ArXiv, abs/1712.01887.
  * [29] Alistarh D., Grubic D., Li J., Tomioka R. and Vojnovic M. (2016). QSGD: Communication-Optimal Stochastic Gradient Descent, with Applications to Training Neural Networks.
  * [30] Achille A., Rovere M. and Soatto S. (2017). Critical Learning Periods in Deep Neural Networks. ArXiv, abs/1711.08856.
  * [31] Agarwal S., Wang H., Lee K., Venkataraman S. and Papailiopoulos D. (2021). Accordion: Adaptive Gradient Communication via Critical Learning Regime Identification. ArXiv, abs/2010.16248.
  * [32] Goyal P., Dollár P., Girshick R.B., Noordhuis P., Wesolowski L., Kyrola A., Tulloch A., Jia Y. and He K. (2017). Accurate, Large Minibatch SGD: Training ImageNet in 1 Hour. ArXiv, abs/1706.02677.
  * [33] Smith S.L., Kindermans P. and Le Q.V. (2018). Don’t Decay the Learning Rate, Increase the Batch Size. ArXiv, abs/1711.00489.
  * [34] Frankle J., Schwab D.J. and Morcos A.S. (2020). The Early Phase of Neural Network Training. ArXiv, abs/2002.10365.