title: Pollux: Co-adaptive Cluster Scheduling for Goodput-Optimized Deep Learning
authors: Aurick Qiao Sang Keun Choe Suhas Jayaram Subramanya Willie Neiswanger Qirong Ho Hao Zhang Gregory R. Ganger Eric P. Xing
text: 
  
  
  
  
  
  
  
  
  
  
  
  
   
  
  
  
  
  : Co-adaptive Cluster Scheduling for Goodput-Optimized Deep Learning
      
  Aurick Qiao1,2
      Sang Keun Choe2
      Suhas Jayaram Subramanya2
      Willie Neiswanger1,2
  
  
  Qirong Ho1
      Hao Zhang1,3
      Gregory R. Ganger2
      Eric P. Xing4,1,2
  
  
  
  
  1Petuum, Inc.
      2Carnegie Mellon University
      3UC Berkeley
      4MBZUAI
  
      
  =========================================================================================================================================================================================================================================================
  
  
  
  
  
  
   improves scheduling performance in deep learning (DL) clusters by adaptively co-optimizing inter-dependent factors both at the per-job level and at the cluster-wide level. Most existing schedulers expect users to specify the number of resources for each job, often leading to inefficient resource use. Some recent schedulers choose job resources for users, but do so without awareness of how DL training can be re-optimized to better utilize the provided resources.
  
   simultaneously considers both aspects.
  By monitoring the status of each job during training,  models how their goodput (a metric we introduce to combine system throughput with statistical efficiency) would change by adding or removing resources.
  
   dynamically (re-)assigns resources to improve cluster-wide goodput, while respecting fairness and continually optimizing each DL job to better utilize those resources.
  
  In experiments with real DL jobs and with trace-driven simulations,  reduces average job completion times by 37–50% relative to state-of-the-art DL schedulers, even when they are provided with ideal resource and training configurations for every job.
  
  
   promotes fairness among DL jobs competing for resources,
  based on a more meaningful measure of useful job progress, 
  and reveals a new opportunity for reducing DL cost in cloud environments.
   is implemented and publicly available as part of an open-source project at <https://github.com/petuum/adaptdl>.
  
  
  Deep learning (DL) continues to rise as a common workload in today's shared resource environments such as clouds and datacenters.
  
  By observing specific characteristics of DL training, specialized cluster schedulers can tailor resource allocations to the needs of each job, mediate resource sharing between multiple jobs, and maximize cluster-wide training performance and resource utilization.
  
  Existing DL schedulers are at most one-way adaptive—they may adapt resource allocations for DL jobs, but each DL job runs with fixed training parameters (e.g. batch size and learning rate).
  However, the optimal choice of training parameters often depends on the specific allocation of resources, and should also be adapted for different resource allocations.
  It can be a significant challenge for users to select these training parameters, leading to sub-optimal utilization of the allocated resources and longer job completion time.
  The difficulty of this task is multiplied if the scheduler is elastic, and can change each job's resource allocations during run-time.
  
  We present , a co-adaptive scheduler which not only elastically allocates resources for each DL job in a cluster, but also automatically adapts their training parameters to best utilize those resources. With co-adaptivity,  can exploit an additional property of DL—that the scalability of a DL job may change over time according to its underlying statistical trends during training. We show that  outperforms current state-of-the-art DL schedulers by up to 2× in terms of job completion time, and reveals a new opportunity to reduce the cost of training large models in cloud environments. (Greg) Should bring out both perfectly-tuned jobs and realistic ones.
  Mention/define goodput in abstract, if it'll be in title?
  
  
  
  
  
  § INTRODUCTION
  
  
  Deep learning (DL) training has rapidly become a dominant workload in many shared resource environments such as datacenters and the cloud.
  DL jobs are resource-intensive and long-running, often demanding distributed execution using expensive hardware devices (eg. GPUs or TPUs) in order to complete within reasonable amounts of time. To meet this resource demand, dedicated clusters are often provisioned for deep learning <cit.>, with a scheduler that mediates resource sharing between many competing DL jobs. 
  
  
  
  
  
  
  Existing schedulers require users to manually configure their jobs, 
  which if done improperly, can greatly degrade training performance and resource efficiency. For example, allocating too many GPUs may result in long queuing times and inefficient resource usage, while allocating too few GPUs may result in long runtimes and unused resources. 
  Such decisions are especially difficult to make in a shared-cluster setting, since optimal choices are dynamic and depend on the cluster load while a job is running.
  
  Even though recent elastic schedulers can automatically select an appropriate amount of resources for each job, they do so blindly to inter-dependent training-related configurations that are just as important.
  For example, the batch size and learning rate of a DL job influence the amount of computation 
  
  needed to train its model. Their optimal choices vary between different DL tasks and model architectures, and they have strong dependence on the job's allocation of resources.
  
  The amount of resources, batch size, and learning rate are difficult to configure appropriately without expert knowledge about both the cluster hardware performance and DL model architecture.
  Due to the inter-dependence between their optimal values, they should be configured jointly with each other. Due to the dynamic nature of shared clusters, their optimal values may change over time. This creates a complex web of considerations a user must make in order to configure their job for efficient execution and resource utilization.
  
  The above paragraph, and really most of the intro from there to "This paper presents" paragraph, kinda struggles with "does the user pick the GPU count or does the scheduler?".  If the user specifies it, then most of what is discussed is much easier... just tune for that number of GPUs.  But, the cluster efficiency may be very poor, because the user can't know what will be available over the job's lifetime and likely even at submission time.  In part as a result, it focuses a lot of the training parameter effects and little on the dynamic GPU availability and where to assign them aspect.  Leaving this note here for now... if folks want, I can try to wade into it Wed evening/night.re-written according to feedback
  
  How can a cluster scheduler help to automatically configure user-submitted DL jobs? Fundamentally, a properly-configured DL job strikes a balance between two often opposing desires:
  (1) system throughput, the number of training examples processed per wall-clock time, and
  (2) statistical efficiency, the amount of progress made per training example processed.
  
  System throughput can be increased by increasing the batch size, as illustrated in Fig. <ref>.
  A larger batch size enables higher utilization of more compute resources (e.g., more GPUs). 
  
  
  But, even with an optimally-retuned learning rate, increasing the batch size often results in  a decreased statistical efficiency <cit.>. For every distinct allocation of GPUs, there is potentially a different batch size that best balances increasing system throughput with decreasing statistical efficiency, as illustrated in Fig. <ref>. Furthermore, how quickly the statistical efficiency decreases with respect to the batch size depends on the current training progress. A job in a later stage of training can potentially tolerate 10x or larger batch sizes without degrading statistical efficiency, than earlier during training <cit.>.
  
  
  
  
  
  Modified the following to bring up more concrete things  tunes.
  
  Guided by these insights, this paper presents , a hybrid resource scheduler that co-adaptively allocates resources and tunes the batch size and learning rate for all DL jobs in a shared cluster. 
   achieves this by jointly managing several system-level and training-related parameters, including the number of GPUs, co-location of workers, per-GPU batch size, gradient accumulation, and learning rate scaling. In particular:
  
  
  
   We propose a formulation of goodput for DL jobs, which is a holistic measure of training performance that takes into account both system throughput and statistical efficiency.
  
   We show that a model of a DL job's goodput can be learned by observing its throughput and statistical behavior during training, and used for predicting the performance given different resource allocations and batch sizes.
  
  
   We design and implement a scheduling architecture that uses such models to configure the right combination of resource allocation and training parameters for each pending and running DL job.  This includes locally tuning system-level and training-related parameters for each DL job, and globally optimizing cluster-wide resource allocations. The local and global components actively communicate and cooperate with each other, operating based on the common goal of goodput maximization. 
  
   We evaluate  on a cluster testbed using a workload derived from a Microsoft cluster trace.
  Compared with recent DL schedulers, Tiresias <cit.> and Optimus <cit.>,
  
   reduces the average job completion time by up to 73%. Even when all jobs are manually tuned beforehand,  reduces the average job completion time by 37%–50%.
  At the same time,  improves finish-time fairness <cit.> by 1.5×–5.4×.
  
   
  
  
  We show that, in cloud environments, using goodput-driven auto-scaling based on  can potentially reduce the cost of training large models by 25%.
  
  
  I'm not sure what to do with the cloud thing here.Can maybe remove to save space?
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  § BACKGROUND: DISTRIBUTED DL TRAINING
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  Training a deep learning model typically involves minimizing a loss function of the form
  
      ℒ(w) = 1/|X|∑_x_i ∈ Xℓ(w, x_i),
  
  where w ∈ℝ^d are the model parameters to be optimized, X is the training dataset, x_i is an individual sample in X, and ℓ is the loss evaluated at a single sample.
  
  The loss function can be minimized using stochastic gradient descent (SGD) or its variants like AdaGrad <cit.> and Adam <cit.>. For the purpose of explaining system throughput and statistical efficiency, we will use SGD as the running example. SGD repeatedly applies the following update until the loss converges to a stable value:
  
      w^(t+1) = w^(t) - ηĝ^(t).
      
  
  η is known as the learning rate, which is a scalar that
  controls the magnitude of each update, and 
  ĝ^(t) is a stochastic gradient estimate of the loss function
  ℒ, evaluated using a random mini-batch
  ℳ^(t)⊂ X of the training data:
  
      ĝ^(t) = 1/M∑_x_i ∈ℳ^(t)∇ℓ(w^(t), x_i).
  
  The learning rate η and batch size M=|ℳ^(t)| are training parameters which are typically chosen by the user.
  
  
  
  
   §.§ System Throughput
  
  The system throughput of DL training can be defined as the
  number of training samples processed per unit of wall-clock time.
  When a DL job is distributed across several nodes, its system throughput
  is determined by several factors, including (1) the allocation and placement
  of resources (e.g. GPUs) assigned to the job, (2) the method of distributed execution
  and synchronization, and (3) the batch size.
  
  
  Data-parallel execution.
  Synchronous data-parallelism is a popular method of distributed execution for DL training.
  The model parameters w^(t) are replicated across a set of distributed GPUs 1, …, K, and each mini-batch ℳ^(t) is divided into equal-sized partitions per node, ℳ^(t)_1, …, ℳ^(t)_K.
  Each GPU k computes a local gradient estimate ĝ^(t)_k using its own partition:
  
      ĝ^(t)_k = 1/m∑_x_i ∈ℳ^(t)_k∇ℓ(w^(t), x_i),
  
  where m = |ℳ^(t)_k| is the per-GPU batch size. These local gradient estimates are then averaged across all GPUs to obtain the desired ĝ^(t). 
  Finally, each node applies the same update using ĝ^(t) to obtain the new model parameters w^(t+1). 
  
  
  
  
  
  
  
  
  
  
  
  
  The run-time of each training iteration is determined by two main components.
  First, the time spent computing each ĝ^(t)_k, which we denote by T_grad.
  
  Second, the time spent averaging ĝ^(t)_k (e.g. using collective all-reduce <cit.>) and/or synchronizing w^(t) (e.g. using parameter servers <cit.>) across all GPUs, which we denote by T_sync.
  T_sync
  is influenced by the size of the gradients, performance of the network, and is typically shorter
  when the GPUs are co-located within the same physical node or rack.
  
  
  Limitations due to the batch size. 
  When the number of GPUs is increased, T_grad decreases due to a smaller per-GPU batch size.
  On the other hand, T_sync, which is typically independent of the batch size, remains unchanged.
  By Amdahl's Law, no matter how many GPUs are used, the run-time
  of each training iteration is lower bounded by T_sync.
  To overcome this scalability limitation, a common strategy is to
  increase the batch size. Doing so causes the 
  
  local gradient estimates to be computed over more training examples and thereby increasing the ratio of T_grad to T_sync.
  
  As a result, using a larger batch size enables higher system throughput
  when scaling to more GPUs in the synchronous data-parallel setting.
  
  
  
  
   §.§ Statistical Efficiency
  
  
  
  
  
  
  The statistical efficiency of DL training can be defined as the amount of
  training progress made per unit of training data processed, influenced by parameters such as batch size or learning rate; for example, a larger batch size normally decreases the statistical efficiency.
  
  The ability to predict statistical efficiency is key to
  
  improving said statistical efficiency, because we can use the predictions to better adapt the batch sizes and learning rates.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  Gradient noise scale.
  
  
  Previous work <cit.> relate the statistical efficiency
  
  of DL training to the gradient noise scale (GNS), which measures the noise-to-signal ratio of the stochastic gradient.
  
  
  
  
  
  
  
  
  A larger GNS means that training parameters such as the
  batch size and learning rate can be increased to higher values
  with relatively less reduction of the statistical efficiency.
  
  
  
  The GNS can vary greatly between different
  DL models <cit.>.
  It is also non-constant and tends to gradually increase during training,
  by up to 10 × or more <cit.>.
  Thus, it is possible to attain significantly better statistical efficiency for large batch sizes
  later on during training.
  
  The gradient noise scale mathematically captures an intuitive explanation of how the batch size affects statistical efficiency. When the stochastic gradient has low noise, adding more training examples to each mini-batch does not significantly improve each gradient estimate, which lowers statistical efficiency. When the  stochastic gradient has high noise, adding more training examples to each mini-batch reduces the noise of each gradient estimate, which maintains high statistical efficiency. Near convergence, the stochastic gradients have relatively lower signal than noise, and so larger batch sizes can be more useful later in training.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  Learning rate scaling.
  
  When training with an increased total batch size M, the learning rate η should also be increased, otherwise the final trained model quality/accuracy can be significantly worse <cit.>.
  How to increase the learning rate varies between different models and training algorithms (e.g. SGD, Adam <cit.>, AdamW <cit.>), and several well-established scaling rules may be used. 
  For example, the linear scaling rule <cit.>, which prescribes that η be scaled proportionally with M, or the square-root scaling rule <cit.> (commonly used with Adam), which prescribes that η be scaled proportionally with √(M).
  More recent scaling rules such as AdaScale <cit.> may scale the learning rate adaptively during training.
  
  In addition to decreasing statistical efficiency, using large batch sizes may also degrade the final model quality in terms of validation performance <cit.>, although the reasons behind this effect are not completely understood at the time of this paper. However, for each of the learning rate scaling rules mentioned above, there is usually a problem-dependent range of batch sizes that achieve similar validation performances. Within these ranges, the batch size may be chosen more freely without significantly degrading the final model quality.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
   §.§ Existing DL Schedulers
  
  
  
  
  
  
  
  
  
  We broadly group existing DL schedulers into two categories, to put  in context. 
  First, non-scale-adaptive schedulers are agnostic to the performance scalability of DL jobs with respect to the amount of allocated resources. 
  For example, Tiresias <cit.> requires users to specify the number of GPUs at the time of job submission, which will be fixed for the lifetime of the job. 
  
  Gandiva <cit.> 
  
  also 
  requires users to specify number of GPUs, but enhances resource utilization through fine-grained time sharing and job packing. 
  Although Gandiva may dynamically change the number of GPUs used by a job, it does so opportunistically and not based on knowledge of job scalability.
  
  Second, scale-adaptive schedulers automatically decide the amount of resources allocated to each job based on how well they can be utilized to speed up the job. 
  For example, Optimus <cit.> learns a predictive model for the system throughput of each job given various amounts of resources, and optimizes cluster-wide resource allocations to minimize the average job completion time. 
  SLAQ <cit.>, which was not evaluated on DL, uses a similar technique to minimize the average loss values for training general ML models.
  Gavel <cit.> goes further by scheduling based on a throughput metric that is comparable across different accelerator types.['s current throughput model does not consider accelerator heterogeneity.  We believe that extending with Gavel's metric would allow  to co-adapt for goodput in heterogeneous DL clusters.]
  AntMan <cit.> uses dynamic scaling and fine-grained GPU sharing to improve cluster utilization, resource fairness, and job completion times.
  Themis <cit.> introduces the notion of finish-time fairness, and promotes fairness between multiple DL applications with a two-level scheduling architecture.
  
  
  Crucially, existing schedulers are agnostic to the statistical efficiency of DL training and the inter-dependence of resource decisions and training parameters.
   explicitly co-adapts these inter-dependent values to improve goodput for DL jobs.
  
  
  
  
  
  
  
  
  
  § THE GOODPUT OF DL TRAINING AND 
  
  
  
  In this section, we define the goodput[Our notion of goodput for DL is analogous to the traditional definition of goodput in computer networks, ie. the useful portion of throughput as benchmarked by training progress per unit of wall-clock time.] of DL jobs, which is a measure of training performance that takes into account both system throughput and statistical efficiency. We then describe how the goodput can be measured during training and used as a predictive model, which is leveraged by  to jointly optimize cluster-wide resource allocations and batch sizes.
  
  
  
  (Goodput)
  The goodput of a DL training job at iteration t is the product between its system throughput and its statistical efficiency at iteration t,
  
      𝙶𝙾𝙾𝙳𝙿𝚄𝚃_t(⋆) = 𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃(⋆) ×𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t(M(⋆)),
  
  where ⋆ represents any configuration parameters that jointly influence the throughput and batch size during training, and M is the total batch size summed across all allocated GPUs.
  
  
  While the above definition is general across many training systems, we focus on three configuration parameters of particular impact in the context of efficient resource scheduling, i.e. ⋆ = (a, m, s), where:
  
  
      
    * a ∈ℤ^N: the allocation vector, where a_n is the number of GPUs allocated from node n.
      
    * m ∈ℤ: the per-GPU batch size.
      
    * s ∈ℤ: number of gradient accumulation steps (<ref>).
  
  The total batch size is then defined as
  
      M(a, m, s) = 𝚂𝚄𝙼(a) × m × (s + 1).
  
  
  's approach.
  An initial batch size M_0 and learning rate (LR) η_0 are selected by the user when submitting their job.  will start each job using a single GPU, m = M = M_0, s = 0, and η = η_0.
  As the job runs,  profiles its execution to learn and refine predictive models for both 𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃 (<ref>) and 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈 (<ref>).
  Using these predictive models,  periodically re-tunes (a, m, s) for each job, according to cluster-wide resource availability and performance (<ref>).
  
  𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t is measured relative to the initial batch size M_0 and learning rate η_0, and  only considers batch sizes that are at least the initial batch size, ie. M ≥ M_0. In this scenario, 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t(M) is a fraction (between 0 and 1) relative to 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t(M_0). Therefore, goodput can be interpreted as the portion of the throughput that is useful for training progress, being equal to the throughput if and only if perfect statistical efficiency is achieved.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  Plug-in Learning Rate Scaling.
  
  Recall from <ref> that different training jobs may require different learning rate scaling rules to adjust η in response to changes in M. In order to support a wide variety of LR scaling rules, including state-of-the-art rules such as AdaScale <cit.>,  provides a plug-in interface that can be implemented using a function signature
  
      𝚂𝙲𝙰𝙻𝙴_𝙻𝚁(M_0, M) ⟶λ.
  
  𝚂𝙲𝙰𝙻𝙴_𝙻𝚁 is called before every model update step,
  
  and λ is used by  to scale the learning rate. The implementation of 𝚂𝙲𝙰𝙻𝙴_𝙻𝚁 can utilize metrics collected during training, such as the gradient noise scale. Using this interface, one can implement rules including AdaScale, square-root scaling <cit.>,
  
  linear scaling <cit.> and LEGW <cit.>.
  
  
  
  
  
  
   §.§ Modeling Statistical Efficiency
  
  
  
  We model 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t(M) as the amount of progress made per training example using M, relative to using M_0. For SGD-based training, this quantity can be expressed in terms of the gradient noise scale (GNS) <cit.>.
  
  
  
  
  To support popular adaptive variants of SGD like Adam <cit.> and AdaGrad <cit.>, we use the pre-conditioned gradient noise scale (PGNS), derived by closely following the original derivation of the GNS (“simple” noise scale in <cit.>) starting from pre-conditioned SGD[Pre-conditioned SGD optimizes ℒ(Pw) instead of ℒ(w), where P is known as a pre-conditioning matrix. Adaptive variants of SGD such as Adam and AdaGrad may be viewed as vanilla SGD (with momentum) applied together with a particular pre-conditioning matrix P.] rather than vanilla SGD. The PGNS, which we denote by φ_t, is expressed as
  
      φ_t = tr(PΣ P^T)/|Pg|^2,
  
  where g is the true gradient, P is the pre-conditioning matrix of the adaptive SGD algorithm, and Σ is the covariance matrix of per-example stochastic gradients. The PGNS is a generalization of the GNS and is mathematically equivalent to the GNS for the special case of vanilla SGD.
  
  Similar to the GNS (Appendix D of <cit.>), it takes 1 + φ_t/M training iterations to make a similar amount of training progress across different batch sizes M. Therefore, we can use the PGNS φ_t to define a concrete expression for 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t(M) as
  
      𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t(M) = φ_t + M_0/φ_t + M.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  Intuitively, Eqn. <ref> measures the contribution from each training example to the overall progress. If 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t(M) = E, then (1) 0 < E ≤ 1, and (2) training using batch size M will need to process 1/E times as many training examples to make the same progress as using batch size M_0.
  
  During training,  estimates the value of
  φ_t,
  then uses Eqn <ref> to predict the
  𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t at different batch sizes. The measured
  value of φ_t varies
  according to the training progress at iteration t, thus
  𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t(M) reflects the lifetime-dependent trends
  exhibited by the true statistical efficiency.
  
  Fig. <ref> (TOP) shows the validation metrics on a held-out dataset for a variety of DL training tasks (details in Table  <ref>) versus their training progress. “Statistical epochs”[Similar to the notion of “scale-invariant iterations” defined in <cit.>.] is the number of training iterations normalized by 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t so that each statistical epoch makes theoretically, as projected by our model, the same training progress across different batch sizes. Thus, the degree of similarity between validation curves at different batch sizes is an indicator for the accuracy of 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t as a predictor of actual training progress.
  
  Although there are differences in the validation curves for several DL tasks (especially in earlier epochs), they achieve similar best values across the different batch sizes we evaluated (± 1% relative difference for all tasks except DeepSpeech2 at ± 4%). We note that these margins are within the plateau of high-quality models expected from large-batch training <cit.>.
  
  Fig. <ref> (MIDDLE and BOTTOM) show the measured and predicted 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t during training and for a range of different batch sizes. In general, larger batch sizes have lower 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t early in training, but close the gap later on in training. The exceptions being BERT, which is a fine-tuning task starting from an already pre-trained model, and recommendation, which uses a much smaller and shallower model architecture than the others. How 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t changes during training varies from task to task, and depends on specific properties like the learning rate schedule. For example, 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t for ImageNet, which uses step-based learning rate annealing, experiences sharp increases whenever the learning rate is annealed.
  
  Finally, we note that the 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t function (which is supplied with estimates of φ_t by ) is able to accurately model observed values at a range of different batch sizes.
  
  This means that φ_t measured using batch size M can be used by  to predict the value of 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t at a different batch size M^' without needing to train using M^' ahead of time.
  
  
  
  Upper batch size limit. In some cases, as the batch size increases, the chosen LR scaling rule may break down before the statistical efficiency decreases, which degrades the final model quality. To address these cases, the application may define a maximum batch size limit that will be respected by . Nevertheless, we find that a batch size up to 32× larger works well in most cases. Furthermore, limits for common models are well-studied for popular LR scaling rules <cit.>. As better LR scaling rules are developed, they may be incorporated into  using its plug-in interface (<ref>).
  
  
  
  
  
  Estimating φ_t. The PGNS φ_t can be estimated in a similar fashion as the GNS by following Appendix A.1 of <cit.>, except using the pre-conditioned gradient Pg instead of the gradient g. This can be done efficiently when there are multiple data-parallel processes by using the different values of ĝ_k^(t) already available on each GPU k. 
  However, this method doesn't work when there is only a single GPU (and gradient accumulation is off, i.e. s=0). In this particular situation,  switches to a differenced variance estimator <cit.> which uses consecutive gradient estimates ĝ^(t-1) and ĝ^(t).
  
  
  
  
  
  
  
   §.§ Modeling System Throughput
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  To model and predict the system throughput for data-parallel DL, we aim to predict the time spent per training iteration, T_iter, and then calculate the throughput as
  
      𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃(a, m, s) = M(a, m, s) / T_iter(a, m, s).
  
  We start by separately modeling T_grad, the time in each iteration spent computing local gradient estimates, and T_sync, the time in each iteration spent averaging gradient estimates and synchronizing model parameters across all GPUs. We also start by assuming no gradient accumulation, i.e. s = 0.
  
  Modeling T_grad. 
  The local gradient estimates are computed using back-propagation, whose run-time scales linearly with the per-GPU batch size m. Thus, we model T_grad as
  
      T_grad(m) = α_grad + β_grad· m,
  
  where α_grad, β_grad are fittable parameters.
  
  Modeling T_sync.
  When allocated a single GPU, no synchronization is needed and T_sync = 0.
  Otherwise, we model T_sync as a linear function of the number of GPUs since in data-parallelism, the amount of data sent and received from each replica is typically only dependent on the size of the gradients and/or parameters. We include a linear factor to account for performance retrogressions associated with using three or more GPUs, such as increasing likelihood of stragglers or network delays.
  
  
  
  Co-location of GPUs on the same node reduces network communication, which can improve T_sync. Thus, we use different parameters depending on GPU placement.
  Letting K = 𝚂𝚄𝙼(a) be the number of allocated GPUs,
  
      T_sync(a, m) =
          
              0    if  K = 1 
      α_sync^local + β_sync^local· (K-2)    if  N = 1,  K ≥ 2 
      α_sync^node + β_sync^node· (K-2)    otherwise,
  
  where N is the number of physical nodes occupied by at least one replica. α_sync^local and β_sync^local are the constant and retrogression parameters for when all processes are co-located onto the same node. α_sync^node and β_sync^node are the analogous parameters for when at least two process are located on different nodes. Note that our model for T_sync can be extended to account for rack-level locality by adding a third pair of parameters.
  
  Combining T_grad and T_sync. Modern DL frameworks can partially overlap T_grad and T_sync by overlapping gradient computation with network communication <cit.>. The degree of this overlap depends on structures in the specific DL model being trained, like the ordering and sizes of its layers.
  
  Assuming no overlap, then T_iter = T_grad + T_sync. Assuming perfect overlap, then T_iter = max(T_grad, T_sync). A realistic value of T_iter is somewhere in between these two extremes. To capture the overlap between T_grad and T_sync, we model T_iter as
  
      T_iter(a, m, 0) = ( T_grad(a, m)^γ + T_sync(a)^γ)^1/γ,
  
  where γ≥ 1 is a learnable parameter. Eqn. <ref> has the property that T_iter = T_grad + T_sync when γ = 1, and smoothly transitions towards T_iter = max(T_grad, T_sync) as γ→∞.
  
  
  
  Gradient Accumulation. In data-parallelism, GPU memory limits the per-GPU batch size, and many DL models hit this limit before the batch size is large enough for T_grad to overcome T_sync (or experience diminishing statistical efficiency), resulting in suboptimal scalability. Several techniques exist for overcoming the GPU memory limit <cit.>; we focus on gradient accumulation, which is easily implemented using popular DL frameworks. Per-GPU gradients are aggregated locally over s forward-backward passes before being synchronized across all GPUs during the (s+1)^th pass, achieving a larger total batch size. Thus, one iteration of SGD spans s accumulation steps followed by one synchronization step, modeled as
  
      T_iter(a, m, s) = s × T_grad(a, m)
          + ( T_grad(a, m)^γ + T_sync(a)^γ)^1/γ.
  
  
  Throughput model validation. 
  Fig. <ref> shows an example of our 𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃 function fit to measured throughput values for a range of resource allocations and batch sizes. Each DL task was implemented using PyTorch <cit.>, which overlaps the backward pass' computation and communication. Gradients are synchronized with NCCL 2.7.8, which uses either ring all-reduce or tree all-reduce depending on the detected GPUs and their placements and its own internal performance estimates.
  Overall, we find that our model can represent the observed data closely, while varying both the amount of resources as well as the batch size. In particular, all models we measured except ImageNet exhibited high sensitivity to inter-node synchronization, indicating that they benefit from co-location of GPUs. Furthermore, YOLOv3 and BERT benefit from using gradient accumulation to increase their total batch sizes. These detailed characteristics are well-represented by our 𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃 function, and can be optimized for by .
  
  In addition to the configurations in Fig. <ref>, we fitted the 𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃 function on a diverse set of GPU placements and batch sizes in a 64-GPU cluster. Across all DL tasks, the average error of the fitted model was at most 10%, indicating that it represents the observed throughput measurements well.
  
  Limits of the throughput model.  models data-parallel training throughput only in the dimensions it cares about, i.e. number and co-locality of GPUs, batch size, and gradient accumulation steps. The simple linear assumptions made in Eqn. <ref>, although sufficiently accurate for the settings we tested, may diverge from reality for specialized hardware <cit.>, sophisticated synchronization algorithms <cit.>, different parallelization strategies <cit.>, at larger scales <cit.>, or hidden resource contention not related to network used for gradient synchronization. Rather than attempting to cover all scenarios with a single throughput model, we designed 𝙶𝙾𝙾𝙳𝙿𝚄𝚃_t (Eqn. <ref>) to be modular so that different equations for 𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃 may be easily plugged in without interfering with the core functionalities provided by .
  
  
  
  
  
  §  DESIGN AND ARCHITECTURE
  
  
  
  
  
  
  
   adapts DL job execution at two distinct granularities. 
  First, at a job-level granularity,  dynamically tunes the batch size and learning rate for best utilization of the allocated resources. 
  Second, at the cluster-wide granularity,  dynamically (re-)allocates resources, driven by the goodput of all jobs sharing the cluster combined with cluster-level goals including fairness and job-completion time. 
  To achieve this co-adaptivity in a scalable way, 's design consists of two primary components, as illustrated in Fig. <ref>.
  
  First, a Agent runs together with each job. It fits the 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t and 𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃 functions for that job, and tunes its batch size and learning rate for efficient utilization of its current allocated resources. Agent periodically reports the goodput function of its job to the Sched.
  
  Second, the Sched periodically optimizes the resource allocations for all jobs in the cluster, taking into account the current goodput function for each job and cluster-wide resource contention. Scheduling decisions made by Sched also account for the overhead associated with resource re-allocations, slowdowns due to network interference between multiple jobs, and resource fairness.
  
  Agent and Sched co-adapt to each other. While Agent adapts each training job to make efficient use of its allocated resources, Sched dynamically re-allocates each job's resources, taking into account the Agent's ability to tune its job.
  
  
  
  
  
   §.§ Agent: Job-level Optimization
  
  
  
  An instance of Agent is started with each training job. During training, it continually measures the job's gradient noise scale and system throughput, and it reports them to Sched at a fixed interval. It also uses this information to determine the most efficient batch size for its job given its current resource allocations, and adapts its job's learning rate to this batch size using the appropriate plug-in LR scaling rule (e.g. AdaScale for SGD or square-root scaling for Adam).
  
  Online model fitting.
  In <ref>, we defined the system throughput parameters of a training job as the 7-tuple
  
      θ_sys = (α_grad, β_grad, α_sync^local, β_sync^local, α_sync^node, β_sync^node, γ),
  
  which are required to construct the 𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃 function. Together with the PGNS φ_t (for predicting 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t) and initial batch size M_0, the triple ( θ_sys, φ_t, M_0 ) specifies the 𝙶𝙾𝙾𝙳𝙿𝚄𝚃 function. While M_0 is a constant configuration provided by the user, and φ_t can be computed according to <ref>, θ_sys is estimated by fitting the 𝚃𝙷𝚁𝙾𝚄𝙶𝙷𝙿𝚄𝚃 function to observed throughput values collected about the job during training.
  
  
  
  Agent measures the time taken per iteration, T_iter, and records the tuple (a, m, s, T_iter) for all combinations of resource allocations a, per-GPU batch size m, and gradient accumulation steps s encountered during its lifetime.
  Periodically, Agent fits the parameters θ_sys to all of the throughput data collected so far. Specifically, we minimize the root mean squared logarithmic error (RMSLE) between Eqn. <ref> and the collected data triples, using L-BFGS-B <cit.>. We set constraints for each α and β parameter to be non-negative, and γ to be in the range [1, 10]. Agent then reports the updated values of θ_sys and  φ_t to Sched.
  
  Prior-driven exploration.
  At the beginning of each job, throughput values have not yet been collected.
  
  To ensure that  finds efficient resource allocations through systematic exploration,
  we impose several priors which bias θ_sys towards
  
  the belief that throughput scales perfectly with more resources, until such resource configurations are explored.
  
  In particular, we set α_sync^local = 0 while the job had not used more than one GPU, α_sync^local = β_sync^local = 0 while the job had not used more than one node, and β_sync^local = β_sync^node = 0 while the job had not used more than two GPUs.
  
  This creates the following behavior: each job starts with a single GPU and is initially assumed to scale perfectly to more GPUs. Sched is then encouraged to allocate more GPUs and/or nodes to the job, naturally as part of its resource optimization (<ref>), until the Agent can estimate θ_sys more accurately. Finally, to prevent a job from being immediately scaled out to arbitrarily many GPUs, we restrict the maximum number of GPUs that can be allocated to at most twice the maximum number of GPUs the job has been allocated in its lifetime.
  
  Although other principled approaches to exploration can be applied (e.g., Bayesian optimization), we find that this simple prior-driven strategy is sufficient in our experiments. Sec. <ref> shows that prior-driven exploration performs close (within 2-5%) to an idealized scenario in which the model is fitted offline for each job before being submitted to the cluster.
  
  
  
  
  
  Training job tuning.
  With θ_sys, φ_t, and M_0, which fully specify the DL job's 𝙶𝙾𝙾𝙳𝙿𝚄𝚃 function at its current training progress, Agent determines the most efficient per-GPU batch size and gradient accumulation steps,
  
      (m^*,   s^*) = max_m,s 𝙶𝙾𝙾𝙳𝙿𝚄𝚃(a, m, s),
  
  where a is the job's current resource allocation.
  
  Once a new configuration is found, the job will use it for its subsequent training iterations, using the plug-in LR scaling rule to adapt its learning rate appropriately. As the job's 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t function changes over time, Agent will periodically re-evaluate the most efficient configuration.
  
  
  
   §.§ Sched: Cluster-wide Optimization
  
  
  
  The Sched periodically allocates (and re-allocates) resources for every job in the cluster. To determine a set of efficient cluster-wide resource allocations, it maximizes a fitness function that is defined as a generalized (power) mean across speedups for each job:
  
      𝙵𝙸𝚃𝙽𝙴𝚂𝚂_p(A) = (1/J∑_j=1^J𝚂𝙿𝙴𝙴𝙳𝚄𝙿_j(A_j)^p)^1/p.
  
  A is an allocation matrix with each row A_j being the allocation vector for a job j, thus A_jn is the number of GPUs on node n allocated to job j, and J is the total number of running and pending jobs sharing the cluster.
  We define the speedup of each job as the factor of goodput improvement using a given resource allocation over using a fair-resource allocation, ie.
  
      𝚂𝙿𝙴𝙴𝙳𝚄𝙿_j(A_j) = max_m,s𝙶𝙾𝙾𝙳𝙿𝚄𝚃_j(A_j, m, s)/max_m,s𝙶𝙾𝙾𝙳𝙿𝚄𝚃_j(a_f, m, s),
  
  where 𝙶𝙾𝙾𝙳𝙿𝚄𝚃_j is the goodput of job j at its current training iteration, and a_f is a fair resource allocation for the job, defined to be an exclusive 1/J share of the cluster.[We note that 𝚂𝙿𝙴𝙴𝙳𝚄𝙿 has similarities with finish-time fairness <cit.>. But, 𝚂𝙿𝙴𝙴𝙳𝚄𝙿 is related to training performance at a moment in time, whereas finish-time fairness is related to end-to-end job completion time.]
  
  
  In <ref>, we described how the 𝙶𝙾𝙾𝙳𝙿𝚄𝚃 function can be fitted to observed metrics during training and then be evaluated as a predictive model. Sched leverages this ability to predict 𝙶𝙾𝙾𝙳𝙿𝚄𝚃 to maximize 𝙵𝙸𝚃𝙽𝙴𝚂𝚂 via a search procedure, and then it applies the outputted allocations to the cluster.
  
  
  Fairness and the effect of p.
  When p=1, 𝙵𝙸𝚃𝙽𝙴𝚂𝚂_p is the average of 𝚂𝙿𝙴𝙴𝙳𝚄𝙿 values across all jobs.
  
  This causes Sched to allocate more GPUs to jobs that achieve a high 𝚂𝙿𝙴𝙴𝙳𝚄𝙿 when provided with many GPUs (i.e., jobs that scale well). However, as p → -∞, 𝙵𝙸𝚃𝙽𝙴𝚂𝚂_p smoothly approaches the minimum of 𝚂𝙿𝙴𝙴𝙳𝚄𝙿 values, in which case maximizing 𝙵𝙸𝚃𝙽𝙴𝚂𝚂_p promotes equal 𝚂𝙿𝙴𝙴𝙳𝚄𝙿 between training jobs, but ignores the overall cluster goodput and resource efficiency.
  
  
  
  Thus, p can be considered a “fairness knob”, with larger negative values being more fair. 
  A cluster operator may select a suitable value, 
  
  based on organizational priorities.
  In our experience and results in <ref>, we find that p=-1 achieves 
  
  most goodput improvements and reasonable fairness.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  Re-allocation penalty. Each time a job is re-allocated to a different set of GPUs, it incurs some delay to re-configure the training process. Using the the popular checkpoint-restart method, we measured between 15 and 120 seconds of delay depending on the size of the model being trained and other initialization tasks in the training code. To prevent an excessive number of re-allocations, when Sched evaluates the fitness function for a given allocation matrix, it applies a penalty for every job that needs to be re-allocated,
  
      𝚂𝙿𝙴𝙴𝙳𝚄𝙿_j(A_j) ⟵𝚂𝙿𝙴𝙴𝙳𝚄𝙿_j(A_j) ×𝚁𝙴𝙰𝙻𝙻𝙾𝙲_𝙵𝙰𝙲𝚃𝙾𝚁_j(δ).
  
  
  We define 𝚁𝙴𝙰𝙻𝙻𝙾𝙲_𝙵𝙰𝙲𝚃𝙾𝚁_j(δ) = (T_j - R_j δ) / (T_j + δ), where T_j is the age of the training job, R_j is the number of re-allocations incurred by the job so far, and δ is an estimate of the re-allocation delay. Intuitively, 𝚁𝙴𝙰𝙻𝙻𝙾𝙲_𝙵𝙰𝙲𝚃𝙾𝚁_j(δ) scales 𝚂𝙿𝙴𝙴𝙳𝚄𝙿_j(A_j) according to the assumption that the historical average rate of re-allocations for job j will continue indefinitely into the future. Thus, a job that has historically experienced a higher rate of re-allocations will be penalized more for future re-allocations.
  
  Interference avoidance.
  When multiple distributed DL jobs share a single node, their network usage while synchronizing gradients and model parameters may interfere with each other, causing both jobs to slow down <cit.>; Xiao et al. <cit.> report up to 50% slowdown for DL jobs which compete with each other for network resources. Sched mitigates this issue by disallowing different distributed jobs (each using GPUs across multiple nodes) from sharing the same node.
  
  Interference avoidance is implemented as a constraint in 's
  
  search algorithm, by ensuring at most one distributed job is allocated to each node. We study the effects of interference avoidance in <ref>.
  
  Supporting non-adaptive jobs. In certain cases, a user may want to run a job with a fixed batch size, i.e. M = M_0. These jobs are well-supported by Sched, which simply fixes 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t for that job to 1 and can continue to adapt its resource allocations based solely on its system throughput.
  
  
  
   §.§ Implementation
  
  
  Agent is implemented as a Python library that is imported into DL training code.
  We integrated Agent with PyTorch <cit.>, which uses all-reduce as its gradient synchronization algorithm.
  Agent inserts performance profiling code that measures the time taken for each iteration of training, as well as calculating the gradient noise scale.
  At a fixed time interval, Agent fits the system throughput model (Eqn. <ref>) to the profiled metrics collected so far, and reports the fitted system throughput parameters, along with the latest gradient statistics, to Sched.
  After reporting to Sched, Agent updates the job's per-GPU batch size and gradient accumulation steps, by optimizing its now up-to-date goodput function (Eqn. <ref>) with its currently allocated resources.
  
  
  
  
  
  
  
  Sched is implemented as a service in Kubernetes <cit.>. At a fixed time interval, Sched runs
  
  its search algorithm, and then applies the resultant allocation matrix by creating and terminating Kubernetes Pods that run the job workers.
  
  To find a good allocation matrix, Sched uses a population-based search algorithm that perturbs and combines candidate allocation matrices to produce higher-value allocation matrices, and finally modifies them to satisfy node resource constraints and interference avoidance. The allocation matrix with the highest fitness score is applied to the jobs running in the cluster.
  
  
  Both Agent and Sched require a sub-procedure that optimizes 𝙶𝙾𝙾𝙳𝙿𝚄𝚃_t(a, m, s) given a fixed a (Eqn. <ref>). We implemented this procedure by first sampling a range of candidate values for the total batch size M, then finding the smallest s such that m = ⌈ M / s ⌉ fits into GPU memory according to a user-defined upper-bound, and finally taking the configuration which results in the highest 𝙶𝙾𝙾𝙳𝙿𝚄𝚃 value.
  
  
  
  § EVALUATION
  
  
  
  
  We compare  with two state-of-the-art DL schedulers using a  testbed cluster with 64 GPUs. Although one primary advantage of  is automatically selecting the configurations for each job, we find that  still reduces average job completion times by 37–50% even when the baseline schedulers are supplied with well-tuned job configurations (a scenario that strongly favors the baseline schedulers).  is able to dynamically adapt each job by trading-off between high-throughput/low-efficiency and low-throughput/high-efficiency modes of training, depending on the current cluster state and training progress.
  
  Using a cluster simulator, we evaluate the impact of specific settings on , including the total workload intensity, prior-driven exploration, scheduling interval, and interference avoidance. With its fairness knob,  can improve finish-time fairness <cit.> by 1.5–5.4× compared to baseline DL schedulers. We also reveal a new opportunity for auto-scaling in the cloud by showing that a -based auto-scaler can potentially reduce the cost of training large models (e.g. ImageNet) by 25%.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
   §.§ Experimental Setup
  
  
  
  Testbed. We conduct experiments using a cluster consisting of 16 nodes and 64 GPUs. Each node is an AWS EC2  instance with 4 NVIDIA T4 GPUs, 48 vCPUs, 192GB memory, and a 900GB SSD. All instances are launched within the same placement group. We deployed Kubernetes 1.18.2 on this cluster, along with CephFS 14.2.8 to store checkpoints for checkpoint-restart elasticity.
  
  Synthetic Workload Construction. 
  We randomly sampled 160 jobs from the busiest 8-hour range (hours 3–10) in the deep learning cluster traces published by Microsoft <cit.>.
  
  Each job in the orginal trace has information on its submission time, number of GPUs, and duration. However, no information is provided on the model architectures being trained or dataset characteristics. Instead, our synthetic workload consists of the models and datasets described in Table <ref>.
  
  
  
  
  
  
  We categorized each job in the trace and in Table <ref> based on their total GPU-time: Small (0–1 GPU-hours), Medium (1–10 GPU-hours), Large (10–100 GPU-hours), and XLarge (100–1000 GPU-hours). 
  For each job in the trace, we picked a training job from Table <ref> that is in the same category.
  
  
  
  
  
  
  Manually-tuned jobs for baseline DL schedulers. 
  We manually tuned the number of GPUs and batch sizes for each job in our synthetic workload, as follows. We measured the time per training iteration for each model in Table <ref> using a range of GPU allocations and batch sizes, and fully trained each model using a range of different batch sizes (see <ref> for details). We considered a number of GPUs valid if using the optimal batch size for that number of GPUs achieves 50% – 80% of the ideal (i.e., perfectly linear) scalability versus using the optimal batch size on a single GPU. For each job submitted from our synthetic workload, we selected its number of GPUs and batch size randomly from its set of valid configurations.
  
  Our job configurations assume that the users are highly rational and knowledgeable about the scalability of the models they are training. Less than 50% of the ideal scalability would lead to under-utilization of resources, and more than 80% of the ideal scalability means the job can still utilize more GPUs efficiently. We emphasize that this assumption of uniformly sophisticated users is unrealistically biased in favor of the baseline schedulers and only serves for comparing  with the ideal performance of baseline systems.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  Comparison of DL schedulers. We compare  to two recent deep learning schedulers, Tiresias <cit.> and Optimus <cit.>, as described in <ref>.
  
  Whereas  dynamically co-adapts the number of GPUs and batch sizes of DL training jobs, Optimus only adapts the number of GPUs, and Tiresias adapts neither. To establish a fair baseline for comparison, for all three schedulers, we scale the learning rate using AdaScale for SGD, and the square-root scaling rule for Adam and AdamW.
  
  . We configured Sched to use a 60s scheduling interval, and compute 𝚁𝙴𝙰𝙻𝙻𝙾𝙲_𝙵𝙰𝙲𝚃𝙾𝚁(δ) using δ = 30s. Agent reports its most up-to-date system throughput parameters and gradient statistics every 30s.
  Unless otherwise specified, the default fairness knob value of p = -1 is used.
  
  Tiresias. We configured Tiresias as described in the testbed experiments of Gu et al. <cit.>, with two priority queues and the  disabled. We manually tuned the queue threshold to perform well for our synthetic workload. Whenever possible, we placed jobs onto as few different nodes as possible to promote worker locality.
  
  Optimus+Oracle. Optimus leverages a throughput prediction model that is specific to jobs using the parameter server architecture. To account for differences due to the performance model, our implementation of Optimus uses our own throughput model as described in <ref>. Furthermore, Optimus predicts the number of training iterations until convergence by fitting a simple function to the model's convergence curve. Since this method does not work consistently for all models in our synthetic workload, we run each job ahead of time and provide Optimus with the exact number of iterations until completion. We call this version of Optimus Optimus+Oracle.
  
  For each job, Tiresias uses the number of GPUs and batch size specified in our synthetic workload. Optimus+Oracle uses the batch size specified, but determines the number of GPUs dynamically. Each job uses gradient accumulation if they are allocated too few GPUs to support the specified batch size.
  
  
  
  
  
  
   §.§ Testbed Macrobenchmark Experiments
  
  
  
  
  Table <ref> summarizes the results of our testbed experiments for seven configurations:  compared with, first, baseline schedulers using well-tuned job configurations; second, baseline schedulers using more realistic job configurations; third,  using two alternate values for its fairness knob.
  
  
  
  Comparisons using well-tuned job configurations. Even when Optimus+Oracle and Tiresias are given well-tuned job configurations as described in <ref>, they are still significantly behind . In this setting,  (with p = -1) achieved 50% and 37% shorter average JCT, 27% and 27% shorter tail (99th percentile) JCT, and 20% and 33% shorter makespan, in comparison to Optimus+Oracle+TunedJobs and Tiresias+TunedJobs, respectively. As we previously noted, this setting highly favors the baseline schedulers, essentially mimicking users who possess expert knowledge about system throughput, statistical efficiency, and how their values change with respect to resource allocations and batch sizes.
  
  One key source of improvement for  is its ability to trade-off between high-throughput/low-efficiency and low-throughput/high-efficiency modes during training.
  Fig. <ref> shows the total number of allocated GPUs and average 𝙴𝙵𝙵𝙸𝙲𝙸𝙴𝙽𝙲𝚈_t during the execution of our synthetic workload.
  During periods of low cluster contention,  can allocate more GPUs (indicated by (A)) and use larger batch sizes to boost training throughput, even at the cost of lower statistical efficiency, because doing so results in an overall higher goodput. On the other hand, during periods of high cluster contention,  may instead use smaller batch sizes to increase statistical efficiency (indicated by (B)).
  
  
  
  
  Comparisons using realistic job configurations. Without assistance from a system like , users are likely to try various numbers of GPUs and batch sizes, before finding a configuration that is efficient. Other users may not invest time into configuring their jobs well in the first place.
  
  To set a more realistically configured baseline, we ran Optimus+Oracle and Tiresias on a version of our synthetic workload with the number of GPUs exactly as specified in the Microsoft cluster trace. The batch size was chosen to be the baseline batch size M_0 times the number of GPUs, which is how we expect most users to initially configure their distributed training jobs. We find that these jobs typically use fewer GPUs and smaller batch sizes than their well-configured counterparts.
  
  Using this workload, we find that  has 72% and 73% shorter average JCT, 50% and 56% shorter tail JCT, and 43% and 48% shorter makespan, in comparison to Optimus+Oracle and Tiresias, respectively. Even though Optimus+Oracle can dynamically increase the GPU allocation of each job, it still only slightly outperforms Tiresias because it does not also increase the batch size to better utilize those additional GPUs.
  
  A closer look at co-adapted job configurations. Fig. <ref> (LEFT) shows the configurations chosen by  for one ImageNet training job as the synthetic workload progresses. (A) during the initial period of low cluster contention, more GPUs are allocated to ImageNet, causing a larger batch size to be used and lowering statistical efficiency. (B) during the subsequent period of high cluster contention, fewer GPUs are allocated to ImageNet, causing a smaller batch size to be used and raising statistical efficiency. (C) when the cluster contention comes back down, ImageNet continues to be allocated more GPUs and uses a larger batch size. However, we note that the batch size per GPU is much higher than in the first low-contention period, since the job is now in its final, high-statistical-efficiency phase of training. We see similar trade-offs being made over time for two YOLOv3 jobs (RIGHT).
  
  
  
  
  Effect of the fairness knob. We ran  using three values of the fairness knob, p = 1, -1, -10. Compared with no fairness (p = 1), introducing a moderate degree of fairness (p=-1) improved the average job completion time (JCT) but degraded the tail JCT. This is because[We note that p = -1 (harmonic mean over speedups) may be more suitable than p = 1 (arithmetic mean) when optimizing for the average JCT.], in our synthetic workload, the tail JCT comprises of long but scalable jobs (i.e. ImageNet), which take a large number of GPUs away from other jobs in the absence of fairness (p=1). However, further increasing fairness (p = -10) degraded performance in average JCT, tail JCT, and makespan. In <ref>, we present a more detailed analysis of the impact of p on scheduling fairness.
  
  System overheads. During each 60s scheduling interval, Sched spent an average of 1 second on 1 vCPU computing the cluster allocations by optimizing the 𝙵𝙸𝚃𝙽𝙴𝚂𝚂_p function. On average, each job was re-allocated resources once every 7 minutes, resulting in an average 8% run-time overhead due to checkpoint-restarts. Each Agent fits its throughput model parameters on its latest observed metrics every 30 seconds, taking an average of 0.2 seconds each time. Finding the optimal per-GPU batch size and gradient accumulation steps by optimizing 𝙶𝙾𝙾𝙳𝙿𝚄𝚃_t takes an average of 0.4 milliseconds.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
   §.§ Simulator Experiments
  
  
  
  
  
  We built a discrete-time cluster simulator in order to evaluate a broader set of workloads and settings.
  
  Our simulator is constructed by measuring the performance and gradient statistics of each model in Table <ref>, under many different resource and batch size configurations, and re-playing them for each simulated job. This way, we are able to simulate both the system throughput and statistical efficiency of the jobs in our workload.
  
  Unless stated otherwise, each experiment in this section is repeated on 8 different workload traces generated using the same duration, number of jobs, and job size distributions as in <ref>, and we report the average results across all 8 traces.
  
  
  
  Simulator construction. For each job in Table <ref>, we measured the time per training iteration for 146 different GPU allocations+placements in our testbed cluster of 16 nodes and 64 total GPUs. For each allocation, we measured a range of batch sizes up to the GPU memory limit. To simulate the throughput for a job, we queried a multi-dimensional linear interpolation on the configurations we measured. For each model, we also measured the (pre-conditioned) gradient noise scale during training using a range of batch sizes, and across every epoch. To simulate the statistical efficiency for a job using a certain batch size, we linearly interpolated its value of the PGNS between the two nearest batch sizes we measured.
  
  
  Simulator fidelity. The data we collected about each job enables our simulator to reproduce several system effects, including the performance impact of different GPU placements. We also simulate the overhead of checkpoint-restarts by injecting a 30-second delay for each job that has its resources re-allocated. Unless stated otherwise, we do not simulate any network interference between different jobs. We study the effects of interference in more detail in <ref>. 
  
  Compared with our testbed experiments in <ref>, we find that our simulator obtains similar factors of improvement, showing that  reduces the average JCT by 48% and 32% over Optimus+Oracle+TunedJobs and Tiresias+TunedJobs.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
    §.§.§ Scheduling Fairness
  
  
  
  We evaluate the scheduling fairness of  using finish-time fairness<cit.> (denoted by ρ), which is defined to be the ratio of a job's JCT running on shared resources to that of the job running in an isolated and equally-partitioned cluster. Under this metric, jobs with ρ < 1 have been treated better-than-fair by the cluster scheduler, while jobs with ρ > 1 have been treated worse-than-fair.
  
  In Fig. <ref>, we compare the finish-time fairness of  with Optimus+Oracle+TunedJobs and Tiresias+TunedJobs.  with p = 1 results in poor fairness, similar to Tiresias+TunedJobs, which is apparent as a long tail of jobs with ρ > 4. Optimus+Oracle+TunedJobs obtains better fairness due to its allocation algorithm which attempts to equalize the JCT improvement for each job. 
  
   with p = -1 provides the best fairness, 
  
  with 99% of jobs achieving ρ < 2, and does so while still providing significant performance increases (Table <ref>). 
  For p = -10, we observe slightly worse fairness overall,
  
  caused by
  Sched incurring a larger number of re-allocations due to ignoring the cost in favor of equalizing speedups at all times.
  
  
  
  
  To provide context, we note that the curves for Tiresias and Optimus are consistent with those reported (for different workloads) by Mahajan et al. <cit.>.
  Although their Themis system is not available for direct comparison, the ρ range for  with p = -1 is similar to the range reported for Themis.  
  The max-ρ improvements (1.5× and 5.4×) over Tiresias and Optimus are also similar.
  
  
  
  
  
  
  
    §.§.§ Other Effects on Scheduling
  
  
  
  Sensitivity to job load. We compare the performance of , Optimus+Oracle+TunedJobs, and Tiresias+TunedJobs for increasing workload intensity in terms of rate of job submissions. Fig. <ref> shows the results. As expected, all three scheduling policies suffer longer average JCT and makespan as the load is increased. Across all job loads,  maintains similar relative improvements over the baseline schedulers.
  
  
  Impact of prior-driven exploration. 
   explores GPU allocations for each DL job from scratch during training (Sec. <ref>). We evaluated the potential improvement from more efficient exploration by seeding each job's throughput models using historical data collected offline. We observed minor (2–5%) reduction in JCT for short jobs like CIFAR10, but no significant change for longer running jobs, indicating low overhead from 's prior-driven exploration.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  Impact of scheduling interval. We ran  using a range of values for its scheduling interval, as shown in Fig. <ref>. We find that  performs similarly well in terms of average JCT for intervals up to 2 minutes, while longer intervals result in performance degradation. Since newly-submitted jobs can only start during the next scheduling interval, we would expect an increase in the average queuing time due to longer scheduling intervals. However, we find that queuing contributed to roughly half of the performance degradation observed, indicating that  still benefits from a relatively frequent adjustment of resource allocations.
  
  Impact of interference avoidance. 
  To evaluate the impact of Sched's interference avoidance constraint, we artificially inject various degrees of slowdown for distributed jobs sharing the same node. Fig. <ref> shows the results. With interference avoidance enabled, the average JCT is unaffected by even severe slowdowns, because network contention is completely mitigated. However, without interference avoidance, the average JCT is 1.4× longer when the interference slowdown is 50%.
  On the other hand, in the ideal scenario when there is zero slowdown due to interference, Sched performs similarly whether or not interference avoidance is enabled. This indicates that Sched is still able to find efficient cluster allocations while obeying the interference avoidance constraint.
  
  
  
  
   §.§ More Applications of 
  
  
  
  
  
    §.§.§ Cloud Auto-scaling
  
  
  
  
  
  In cloud environments, computing resources can be obtained and released as required, and users pay for the duration they hold onto those resources. 
  
  
  Goodput-driven scheduling presents a unique opportunity: when a DL model's statistical efficiency increases during training, it may be more cost-effective to provision more cloud resources and use larger batch sizes during the later epochs of a large training job, rather than earlier on. We present some preliminary evidence using our cluster simulator, and note that a full design of an auto-scaling system based on goodput may be the subject of future work.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  Auto-scaling ImageNet training.
  We implemented a simple auto-scaling policy using 's goodput function. During training, we scaled up the number of nodes whenever max_m,s𝙶𝙾𝙾𝙳𝙿𝚄𝚃_t(a, m, s)/𝚂𝚄𝙼(a) > U·max_m,s𝙶𝙾𝙾𝙳𝙿𝚄𝚃_t(1, m, s), i.e. the goodput exceeds some fraction U of the predicted ideal goodput assuming perfect scalability. We set U=2/3, and increased to a number of nodes such that the predicted goodput is approximately L=1/2 of the predicted ideal goodput.
  
  Fig. <ref> compares our -based auto-scaler with the auto-scaler proposed by Or et al. <cit.>, which allows the batch size to be increased during training, but models job performance using the system throughput rather than the goodput. Since the system throughput does not change with training progress, throughput-based autoscaling (Or et al.) quickly scales out to more nodes and a larger batch size (Fig. <ref>), which remains constant thereafter. On the other hand,  starts with a small number of nodes, and gradually increases the number of nodes as the effectiveness of larger batch sizes improves over time. Fig. <ref> shows that  maintains a high statistical efficiency throughout training.
  Overall, compared to Or et al.'s throughput-based auto-scaling,  trains ImageNet with 25% cheaper cost, with only a 6% longer completion time.
  
  
  
  
  
  
  
  
  
    §.§.§ Hyper-parameter Optimization (HPO)
  
  
  
  Hyper-parameter optimization (HPO) 
  is an
  important DL workload.
  
  In HPO, the user defines a search space over relevant model hyper-parameters. 
  A HPO algorithm (aka a trial scheduler) submits many training jobs (trials) to evaluate the effectiveness of particular hyper-parameters, in terms of objectives such as model accuracy or energy efficiency.
  
  
  Different HPO algorithm types manage trials differently.
  For example, Bayesian optimization algorithms <cit.> may submit a few training jobs at a time, and determine future trials based on the fully-trained results of previous trials. 
  
  Bandit-based algorithms <cit.> may launch a large number of trials at once and early-stop ones that appear unpromising.
  
  A full evaluation on how  affects different HPO algorithm types is
  future work.
  
  
  Table <ref> shows results from 
  
  
  
  
  
  tuning
  a ResNet18 model trained on the CIFAR10 dataset, using a popular Bayesian optimization-based HPO algorithm known as the Tree-structured Parzen Estimator (TPE) <cit.>.
  The search space covers the learning rate and annealing, momentum, weight decay, and network width hyper-parameters. We configured TPE so that 4 trials run concurrently with each other, and 100 trials are run in total. The testbed consists of two NVIDIA DGX A100 nodes, each with 8 A100 GPUs. The baseline scheduler assigns a static allocation of 4 GPUs (all on the same node) to each trial and uses a fixed per-GPU batch size for every trial.
  As expected, similar accuracy values are achieved, but 
   completes HPO 30% faster due to adaptive (re-)allocation of resources as trials progress and adaptive batch sizes.
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
  
   §.§ Artifact
  
  
  We provide an artifact containing the full implementation of , benchmark model implementations (Table <ref>), testbed experiment scripts (Sec. <ref>), cluster simulator implementation and results (Sec. <ref>), available at <https://github.com/petuum/adaptdl/tree/osdi21-artifact>.
  The raw testbed experiment (Sec. <ref>) logs and analysis scripts are provided at <https://github.com/petuum/pollux-results>.
  
  
  
  § ADDITIONAL RELATED WORK
  
  
  
  
  Prior DL schedulers are discussed in <ref>.
  
  
  
  
  
  
  
  
  Other resource schedulers for DL.
  AntMan <cit.> uses dynamic scaling and fine-grained GPU sharing in order to improve cluster utilization, resource fairness, and job completion times.
  Gavel <cit.> uses a new abstraction called the effective throughput and extends existing scheduling policies to be aware of heterogeneous devices in DL clusters. Gavel's effective throughput extends system throughput to be comparable between DL training jobs running on different accelerator types. This is in contrast with 's goodput, which incorporates both system throughput and statistical efficiency.
  
  Even though heterogeneity is not the primary focus of our paper, we believe that  can be extended to be heterogeneity-aware in future work.
  
  
  Adaptive batch size training. Recent work on DL training algorithms have explored dynamically adapting batch sizes for better efficiency and parallelization.
  AdaBatch <cit.> increases the batch size at pre-determined iterations during training, while linearly scaling the learning rate.
  Smith et al. <cit.> suggest that instead of decaying the learning rate during training, the batch size should be increased instead.
  CABS <cit.> adaptively tunes the batch size and learning rate during training using similar gradient statistics as .
  
  These works have a common assumption that extra computing resources are available to parallelize larger batch sizes whenever desired, which is rarely true inside shared-resource environments.  complements existing adaptive batch size strategies by adapting the batch size and learning rate in conjunction with the amount of resources currently available. Alternatively, anytime minibatch <cit.> adapts the batch size to mitigate stragglers in distributed training.
  
  KungFu <cit.> supports adaptive training algorithms, including adaptive batch sizes, by allowing applications to define custom adaptation policies and enabling efficient adaptation and monitoring during training. Although KungFu is directed at single-job training and  at cluster scheduling, we believe KungFu offers useful tools which can be used to implement the adaptive policies used by the Agent.
  
  Hyper-parameter tuning. A large body of work focuses on tuning the hyper-parameters for ML and DL models <cit.>, which typically involves many training jobs <cit.> as discussed earlier. 
  Although batch size and learning rate are within the space of hyper-parameters often optimized by these systems, 's goal is fundamentally different. Whereas HPO algorithms search for the highest model quality,  adapts the batch size and learning rate for the most efficient execution for each job, while not degrading model quality.
  
  
  
  
  § CONCLUSION
  
  
   is a DL cluster scheduler that co-adaptively allocates resources, while at the same time tuning each training job to best utilize those resources. We present a formulation of goodput that combines system throughput and statistical efficiency for distributed DL training. Based on the principle of goodput maximization,  automatically and jointly tunes the resource allocations, batch sizes, and learning rates for DL jobs, which can be particularly difficult for users to configure manually. 
   outperforms and is more fair than
  recent
  
  Is “SOTA" too strong a claim now that new papers are out? Would it be better to (honestly) call out that we simply beat Optimus and Tiersas? I replaced SOTA with recent here... but, do we need to do the same thing in abstract, eval and intro?  Also shortened, so take a look ;).  And, wondering if we should we drop the "fair" aspect from the conclusion. 
  DL schedulers, even if users can configure their jobs well, and provides even bigger benefits with more realistic user knowledge. 
  
  
  
  
  
  
  
  
  
  § ACKNOWLEDGEMENTS
  
  
  We thank our shepherd, Michael Isard, and the anonymous OSDI reviewers for their insightful comments and suggestions that improved our work.
  We also thank our colleagues from Petuum — Omkar Pangarkar, Richard Fan, Peng Wu, Jayesh Gada, and Vishnu Vardhan — for their invaluable contributions toward the open source implementation of .
  
  plain
  
  
  
  
  
  
  
  
  