Optimus: An Efficient Dynamic Resource Scheduler for Deep Learning Clusters — Detailed Summary

Yanghua Peng, Yixin Bao, Yangrui Chen, Chuan Wu, Chuanxiong Guo | The University of Hong Kong / Bytedance Inc. | EuroSys '18: Thirteenth EuroSys Conference, April 23–26, 2018, Porto, Portugal | ACM ISBN 978-1-4503-5584-1/18/04 | DOI: 10.1145/3190508.3190517 | 14 pages

Per-section summary organized by the paper's heading structure. Each section includes paragraph-level bullet points and exact quantitative results where the paper provides them. CCS concepts: Computing methodologies → Machine learning; Computer systems organization → Cloud computing. Keywords: resource management; deep learning. Shepherd: Paolo Romano.


Abstract

"Results show that Optimus outperforms representative cluster schedulers by about 139% and 63% in terms of job completion time and makespan, respectively."


1. Introduction

Growth of DL workloads.

Cost of a training job.

Limitation 1 — static resource allocation.

Limitation 2 — no DL-specific optimizations.

DL clusters need schedulers that leverage "structures of deep learning frameworks (e.g., the parameter server architecture) and characteristics of deep learning jobs (e.g., iterativeness, convergence properties) for maximal training efficiency."

The proposal.

Stated contributions.

  1. Performance modeling — online convergence-curve fitting plus a hardware-agnostic resource-to-speed model, requiring no knowledge of hardware internals or job internals.
  2. Scheduling algorithms — a marginal-gain-driven dynamic resource allocation algorithm and a communication-aware task placement scheme.
  3. Framework fix — identification and resolution of parameter-server load imbalance in MXNet.
  4. System and evaluation — implemented on Kubernetes, tested across 13 servers (7 CPU, 6 GPU), showing 139% JCT and 63% makespan improvement over the fairness baseline.

2. Background and Motivation

2.1 DL Model Training

Iterativeness.

Convergence.

Why training loss (not accuracy / validation loss).

2.2 The Parameter Server Architecture

2.3 Existing Cluster Schedulers

Static allocation in practice.

Motivating measurement — resource configuration matters non-trivially.

Dynamics demand adaptation.

Job-size unawareness.


3. Performance Modeling of DL Jobs

3.1 Learning the Convergence Curve

Data preprocessing.

Online fitting.

Predicting remaining work.

Equation (1) — training-loss convergence model:

\[l = \frac{1}{\beta_0 \cdot k + \beta_1} + \beta_2\]

Symbol Meaning
\(l\) normalized training loss, \(0 \le l \le 1\)
\(k\) cumulative training-step index
\(\beta_0, \beta_1, \beta_2\) non-negative coefficients fitted online via NNLS

3.2 Resource-Speed Modeling

System model — the per-step time breakdown.

Equation (2) — duration of one training step:

\[T = \max_{\rho}\Big[\, m\cdot T_{forward} + T_{back} + \frac{2\,S/p}{B/w'_\rho} + \frac{T_{update}\cdot w'_\rho}{p} + \delta\cdot w + \delta'\cdot p \,\Big]\]

Symbol Meaning
\(T\) duration of one training step on a worker
\(\rho\) parameter-server index (the max is over PS)
\(m\) per-worker mini-batch size
\(T_{forward}\) average forward-propagation time per example
\(T_{back}\) backward-propagation time
\(S\) total model parameter size in bytes
\(p\) number of parameter servers
\(w\) number of workers
\(B\) network bandwidth of each parameter server
\(w'_\rho\) number of workers concurrently transferring data to PS \(\rho\)
\(T_{update}\) time to update parameters of size \(S\)
\(\delta, \delta'\) linear overhead coefficients for worker / PS connection and control handling

Defining training speed.

Asynchronous mode.

Equation (3) — asynchronous training speed:

\[f(p,w) = w\cdot\Big(\theta_0 + \theta_1\cdot\frac{w}{p} + \theta_2\cdot w + \theta_3\cdot p\Big)^{-1}\]

Synchronous mode.

Equation (4) — synchronous training speed:

\[f(p,w) = \Big(\theta_0\cdot\frac{M}{w} + \theta_1 + \theta_2\cdot\frac{w}{p} + \theta_3\cdot w + \theta_4\cdot p\Big)^{-1}\]

Symbol Meaning
\(f(p,w)\) training speed in steps per unit time
\(M\) fixed global mini-batch size (\(m = M/w\))
\(\theta_\bullet\) positive coefficients learned per job via NNLS

Model fitting procedure.

Empirical validation.

Table 2: Coefficients in speed functions

Mode \(\theta_1\) \(\theta_2\) \(\theta_3\) \(\theta_4\) \(\theta_5\) Residual sum of squares for fitting
Async 2.83 3.92 0.00 0.11 0.10
Sync 1.02 2.78 4.92 0.00 0.02 0.00

(The table prints \(\theta_1\ldots\theta_5\) while the equations use \(\theta_0\ldots\theta_4\); the correspondence is positional across the equation terms.)

Table 1: Deep learning jobs used for tests and experiments

Model # of parameters (Million) Network type Application domain Dataset Dataset size (# of examples)
ResNext-110 1.7 CNN image classification CIFAR10 60,000
ResNet-50 25 CNN image classification ILSVRC2012-ImageNet 1,313,788
Inception-BN 11.3 CNN image classification Caltech 30,607
KAGGLE 1.4 CNN image classification Kaggle-NDSB1 37,920
CNN-rand 6 CNN sentence classification MR 10,662
DSSM 1.5 RNN word representation text8 214,288
RNN-LSTM-Dropout 4.7 RNN language modeling PTB 1,002,000
Sequence-to-Sequence 9.1 RNN machine translation WMT17 1,000,000
DeepSpeech2 38 RNN speech recognition LibriSpeech 45,000

Figures in §2–§3

Fig. Caption / content
1 Training curves (ResNext-110 / CIFAR10): loss 0.6–1.8 on the left axis, accuracy on the right, epochs 0–100; smooth loss decay and accuracy saturation
2 Training time of the Table 1 models on a TITAN X Pascal GPU; log-scale bars spanning \(10^0\)\(10^2\) hours
3 Parameter server architecture: partitioned parameters across PS, bidirectional push/pull with data-parallel workers
4 Varying training speeds with different resource configurations. (a) \(w+p=20\): peak at 8 workers + 12 PS. (b) PS:workers = 1:1: non-linear scaling with degradation past ~12 workers
5 Training loss curves for the 9 DL jobs, normalized loss 0.00–1.00 vs. progress 0–100%; shared \(O(1/k)\)-like decay
6 Prediction errors in different DL jobs — convergence-step error (%) vs. progress (%); shrinks rapidly
7 Online model fitting example for Seq2Seq: sampled loss points overlaid with the Eqn. (1) curve
8 Estimation errors of training speeds vs. number of pre-run sample configurations (async and sync); ~10 samples ⇒ <10% error
9 Data points and fitted curves of the speed functions: (a)(b) async speed vs. workers / vs. PS (4–20); (c)(d) sync counterparts (4–16)

4. Dynamic Scheduling

"Optimus periodically allocates resources to the active jobs (new jobs submitted in the previous scheduling interval and unfinished jobs submitted earlier), by adjusting the numbers and placement of parameter servers/workers in each job in the shared DL cluster."

4.1 Resource Allocation

Problem formulation.

\[\text{minimize}\ \sum_{j\in\mathcal{J}} t_j \tag{5}\] \[\text{s.t.}\quad t_j = \frac{Q_j}{f(p_j,w_j)}\quad \forall j\in\mathcal{J} \tag{6}\] \[\sum_{j\in\mathcal{J}}\big(w_j\cdot O_j^r + p_j\cdot N_j^r\big) \le C_r \quad \forall r\in\mathcal{R} \tag{7}\] \[p_j\in\mathbb{Z}^+,\ w_j\in\mathbb{Z}^+ \quad \forall j\in\mathcal{J} \tag{8}\]

Symbol Meaning
\(\mathcal{J}\) set of active DL jobs
\(t_j\) estimated remaining running time of job \(j\)
\(Q_j\) estimated remaining steps/epochs to convergence for \(j\)
\(f(p_j,w_j)\) fitted training-speed function of job \(j\)
\(p_j, w_j\) PS count and worker count assigned to \(j\)
\(\mathcal{R}\) set of resource types
\(O_j^r\) type-\(r\) resource consumed per worker of \(j\)
\(N_j^r\) type-\(r\) resource consumed per PS of \(j\)
\(C_r\) cluster capacity of resource type \(r\)

Hardness.

"The problem is a non-linear (and even non-convex) integer programming problem since Eqn. 6 is not a linear/convex constraint. It can not be solved using LP/convex solvers and is NP-hard in general, so we design an efficient heuristic to solve it."

Equation (9) — marginal gain:

\[\max\left\{ \Big(\frac{Q_j}{f(p_j,w_j)} - \frac{Q_j}{f(p_j+1,w_j)}\Big)\Big/ N_j^{D},\ \ \Big(\frac{Q_j}{f(p_j,w_j)} - \frac{Q_j}{f(p_j,w_j+1)}\Big)\Big/ O_j^{D'} \right\}\]

A dominant resource is "the type of resource that has the maximal share in the overall capacity of the cluster, among all resources used by a worker (parameter server)."

The greedy allocation procedure (described in prose in the paper — there is no numbered algorithm block).

Priority damping for early-stage jobs.

"To mitigate its performance degration due to prediction errors, we can downgrade the priority of a job a bit when it is at the beginning state (i.e., larger prediction errors) by multiplying its marginal gain (i.e., the computed value in (9)) by a factor (e.g., 0.95)."

4.2 Task Placement

Why placement matters.

Worked example (Fig. 10).

Theorem 1.

"Given the numbers of workers and parameter servers in a synchronous training job, the optimal worker/parameter server placement principle to achieve the maximal training speed for the job, in a cluster of homogeneous servers, is to use the smallest number of servers to host the job, such that the same number of parameter servers and the same number of workers are deployed on each of these servers."

Appendix — the placement optimization.

\[\max_k\left\{\ \frac{\frac{S_j}{p_j}(w_j - w_{jk})}{B_j},\ \ \frac{\frac{S_j}{p_j}(p_j - p_{jk})}{b_j}\ \right\}\]

\[\textbf{Subproblem 1: } \min\ \max_k \frac{\frac{S_j}{p_j}(w_j-w_{jk})}{B_j}\ \ \text{s.t.}\ \sum_k w_{jk}=w_j\] \[\textbf{Subproblem 2: } \min\ \max_k \frac{\frac{S_j}{p_j}(p_j-p_{jk})}{b_j}\ \ \text{s.t.}\ \sum_k p_{jk}=p_j\]

Symbol Meaning
\(K\) number of physical nodes
\(p_{jk}, w_{jk}\) PS / workers of job \(j\) placed on node \(k\)
\(S_j\) model size of job \(j\)
\(B_j\) bandwidth requirement of each PS in job \(j\)
\(b_j\) bandwidth requirement of each worker in job \(j\)

The placement algorithm (prose).


5. System Implementation

5.1 Data Serving

5.2 Straggler Handling

"For asynchronous training, it is also important to ensure the workers have similar training speeds so that the parameters on any worker are not too stale; parameter staleness may lead to unstable training progress and hence additional training steps to achieve convergence."

5.3 Load Balancing on Parameter Servers

The bug in MXNet (and TensorFlow).

Three optimization targets.

  1. Minimize the maximum difference in total parameter size assigned to any two PS.
  2. Minimize the total number of parameter-update network requests per step.
  3. Minimize the maximum difference in the number of update requests across PS.

Parameter Assignment Algorithm (PAA).

5.4 Elastic Training on MXNet

5.5 Scheduler on Kubernetes


6. Evaluation

6.1 Methodology

Testbed.

Component Specification
Total servers 13 (7 CPU servers + 6 GPU servers)
CPU server 2 × 8-core Intel E5-2650 (16 cores), 80 GB memory, 2 × 300 GB HDD
GPU server 1 × 8-core Intel E5-1660, 2 × GeForce 1080Ti, 48 GB memory, 1 × 500 GB SSD + 1 × 4 TB HDD
Total GPUs 12
Network 48-port Dell N1548 1GbE switch
Cluster manager Kubernetes 1.7
Storage HDFS 2.8 (128 MB chunks, replication 2)
Framework MXNet (modified for elastic scaling and PAA)
Job-state store etcd
Fitting NNLS solver
Scalability host one core of an Intel E5-1620 v4 CPU
Loss-curve collection host (§3.1) 1 × E5-1650 v4 CPU + 2 × NVIDIA TITAN X GPUs
Fig. 2 timing host TITAN X Pascal (donated by NVIDIA)

Simulator.

Workload.

Baselines.

  1. DRF fairness scheduler — dominant resource fairness as used in Hadoop / YARN / Mesos, dynamically rescheduling jobs each interval, with Kubernetes-default load-balancing placement.
  2. Tetris — preferentially allocates to jobs with low duration or small resource consumption and packs jobs to minimize fragmentation. Because Tetris has no mechanism to estimate DL job remaining time, Optimus's speed function and convergence estimation are supplied to it.

Metrics.

Default parameters.

6.2 Performance

End-to-end comparison.

Metric Optimus DRF Tetris
Avg. JCT 1.00 2.39 1.74
Makespan 1.00 1.63 1.22
Metric Optimus DRF Tetris
Avg. JCT 1161 s 2780 s 2016 s
Makespan 14835 s (4.1 h) 24255 s (6.7 h) 18127 s (5.0 h)

Resource adjustment overhead.

Scalability.

6.3 Sensitivity Analysis

Error-injection setup.

Prediction-error sensitivity (Fig. 15).

Priority factor.

Training modes (Fig. 16).

Mode Metric Optimus DRF Tetris
Async JCT 1.00 1.97 1.64
Async Makespan 1.00 1.36 1.11
Sync JCT 1.00 2.29 1.91
Sync Makespan 1.00 1.45 1.21

Arrival processes (Fig. 17).

Arrival process Metric Optimus DRF Tetris
Poisson (3/interval) JCT 1.00 2.15 1.82
Poisson (3/interval) Makespan 1.00 1.40 1.15
Google trace (7 h) JCT 1.00 2.21 1.78
Google trace (7 h) Makespan 1.00 1.46 1.24

6.4 Inspecting Detailed Designs in Optimus

Resource allocation ablation (Fig. 18).

Metric Optimus DRF Tetris
Avg. JCT 1.00 1.62 1.33
Makespan 1.00 1.31 1.14

Task placement ablation (Fig. 19).

Metric Optimus DRF Tetris
Avg. JCT 1.00 1.17 1.12
Makespan 1.00 1.13 1.09

PS load balancing — parameter distribution (Table 3).

Table 3: Comparison of parameter distribution

Algorithm Difference of parameter sizes Difference of # of requests Total # of requests
MXNet 3.6M 43 247
PAA 0.1M 1 157

PS load balancing — speedup (Figs. 20–21).

Summary of results (as stated by the authors).

  1. Testbed experiments show Optimus improves average JCT and makespan by 139% and 63% compared to the fairness scheduler. Optimus scales to schedule 100,000 tasks on 16,000 nodes in 5 seconds, with resource adjustment overhead of only 2.54%.
  2. Further improvement of estimation accuracy would not increase Optimus's performance much (~15%), and Optimus beats DRF and Tetris under various workloads.
  3. The resource allocation algorithm, task placement scheme and PS load balancing algorithm contribute about 62%, 17%, 20% of the improvement respectively.

7. Discussions

Various workloads.

Convergence estimation.

Scaling overhead.


Performance modeling

System / work Approach How Optimus differs
Jockey, Morpheus Historical traces of periodic jobs; dynamically adjust allocation to meet deadlines Optimus does not depend on previous runs of the same job, since production training data often change (e.g., daily)
PerfOrator Resource-to-performance model of big-data queries via query-size estimation and hardware profiling Optimus uses high-level system modeling with no knowledge of hardware or job internals
Job execution-time estimation; SQL query data-size estimation Fit a parametric model from sample runs Same family of approach that Optimus adopts
Ernest Performance prediction for data analytics with an experimental design to minimize sampling overhead Optimus's configuration space (number of tasks) is small, so 5–10 sample runs suffice
PREDIcT Sample runs to capture convergence trend of a graph algorithm Infeasible for DL training, since dataset size affects convergence
Yan et al. Fine-grained DL modeling (per-operator compute time on a specific CPU, NN structure) Optimus captures high-level computation and communication patterns instead
FABOLAS, BOAT, CherryPick Bayesian-optimization parameter-free search for best hyperparameters / configuration Not applicable: Optimus needs a parametric speed model so the scheduler can optimize globally across all concurrent jobs

Job scheduling

System / work Focus How Optimus differs
Corral, Morpheus Periodic or predictable workloads Optimus handles online, non-periodic DL arrivals
Borg, Fuxi, Firmament Heterogeneous workloads at scale; policy-based scheduling (fairness, locality, priority) Optimus focuses on deep learning workloads
Mesos, YARN (DRF) Allocation by dominant resource fairness Optimus targets resource efficiency and job performance rather than fairness
TetriSched, Morpheus Global dynamic allocation for reservation-based/periodic jobs with deadlines Optimus has no deadlines or reservations
Eagle Hybrid scheduler for head-of-line blocking; partitions cluster into short-job and long-job pools Optimus adjusts dynamic resource configurations rather than partitioning
Huang et al., SLAQ, Dorm (Spark MLlib) Spark memory optimizer; quality-driven scheduling for experimental ML with similar online loss fitting for convex algorithms; utilization-fairness optimizer Optimus targets DL jobs on the PS architecture, using job characteristics in both allocation and placement
STRADS Programming approach that schedules parameter updates for model-parallel ML Optimus does not modify the underlying ML frameworks (beyond the PAA fix)
Azalia et al. Model-free deep RL for model parallelism on a single machine That approach is "yet to be general and efficient" for cluster-level resource allocation
Proteus Exploits transient EC2 VMs for cheap ML training Uses a simpler performance model and focuses on expected cost under dynamic bidding

Distributed ML frameworks


9. Conclusion


Assumptions Made by Optimus

  1. Workload is data-parallel DL training on the parameter server architecture (§1, §2.2).
  2. Production models are mature with tuned hyper-parameters, so convergence is stable and overfitting/divergence is out of scope (§2.1).
  3. Job completion is defined by the training-loss decrease over consecutive epochs falling below \(\delta\) (§2.1).
  4. SGD's \(O(1/k)\) convergence rate justifies the functional form of Eqn. (1) (§3.1).
  5. The network bandwidth bottleneck lies at the parameter servers, not the workers (§3.2).
  6. Theorem 1's optimality assumes homogeneous servers and node capacity sufficient to place the job (§4.2, Appendix).
  7. In synchronous training the global mini-batch \(M\) is fixed, so \(m = M/w\) (§3.2).
  8. Per-container resource footprint is user-specified and fixed; only task counts are scheduled (§2.3).

Limitations

  1. Checkpoint/restart elasticity introduces scaling overhead — measured at 2.54%, but larger for very big models or frequent scaling (§5.4, §6.2, §7).
  2. The Eqn. (1) loss-fitting form may not apply to algorithms whose loss curves deviate from \(O(1/k)\), e.g., A3C in deep RL (§7).
  3. Step-decay learning-rate schedules invalidate the current fit and force a reset of the fitting process (§7).
  4. Prediction error costs performance — about a 15% gap at 20% convergence / 10% speed error (§6.3).
  5. Requires an initial profiling phase: pre-running each job on a small dataset with 5 (in evaluation) to 5–10 (claimed sufficient) \((p,w)\) combinations for tens of seconds (§3.2, §6.1, §8).
  6. Data chunks must be re-partitioned across workers on every scaling event (§5.1).
  7. PAA is implemented in one framework only (MXNet), though the imbalance problem is stated to be common across frameworks (§5.3, §8).

Future Work / Extensions (all stated in §7)

  1. Mixed-workload clusters — plug multiple schedulers into Kubernetes, with Optimus requesting resources from a central manager and scheduling DL jobs over a varying share of the cluster.
  2. Broader convergence estimation — reset fitting for step-decay learning rates, and accept user-supplied parametric convergence functions for non-SGD-like algorithms.
  3. Bounded scaling frequency — a re-allocation threshold, smaller for large jobs, to cap checkpoint/restart overhead.

Glossary of Optimus-Specific Terms (the paper's own definitions)

Term Definition as given in the paper
Straggler (§5.2) A slow worker; in sync training it delays every step, in async training it produces stale gradients, causing unstable progress and additional steps to converge
Marginal gain (§4.1, Eqn. 9) The reduction in JCT from adding one worker (or one PS), divided by the amount of dominant resource that worker (or PS) occupies
Dominant resource (§4.1) "the type of resource that has the maximal share in the overall capacity of the cluster, among all resources used by a worker (parameter server)"
DRF (§2.3, §6.1) Dominant Resource Fairness — the multi-resource fair allocation policy used by Hadoop/YARN/Mesos; size-unaware and work-conserving; the paper's primary baseline
JCT (§6.1) Job completion time — per-job time to convergence; the indicator of system performance
Makespan (§6.1) "the total time elapsed from the arrival of the first job to the completion of all jobs"; minimizing it is equivalent to maximizing resource efficiency
Online fitting (§3.1) Continuously refitting the loss model's coefficients as new loss data points arrive, so the model improves as the job runs
NNLS (§3.1, §3.2) Non-negative least squares solver used to fit both the \(\beta\) coefficients of Eqn. (1) and the \(\theta\) coefficients of Eqns. (3)–(4)
Elastic / dynamic scaling (§5.4) Changing a running job's worker and PS counts via checkpoint-to-HDFS, container termination, and restart from checkpoint
Task placement (§4.2) Mapping a job's PS and worker containers onto physical servers to maximize training speed by minimizing cross-server data transfer
Resource-speed model (§3.2) The fitted function \(f(p,w)\) giving training speed in steps per unit time as a function of PS count and worker count
PAA (§5.3) Parameter Assignment Algorithm — balances parameter sizes and update-request counts across PS while avoiding unnecessary block splitting
Convergence threshold \(\delta\) (§2.1) The training-loss decrease bound over consecutive epochs used as the completion criterion (varied 1%–5% in evaluation)
Priority factor (§4.1, §6.3) The discount (e.g., 0.95) multiplied into a job's marginal gain while it is early in training and prediction errors are large

Structural Note

The paper contains no numbered algorithm/pseudocode blocks — the greedy allocation (§4.1), the placement algorithm (§4.2) and PAA (§5.3) are all described in prose. The paper also makes no asymptotic complexity claims; scalability is argued purely empirically via Fig. 12. The NP-hardness statement for Eqns. (5)–(8) is explicit in the text. Acknowledgements name shepherd Paolo Romano, Hong Kong RGC grants HKU 17204715, 17225516 and C7036-15G (CRF), and the NVIDIA-donated Titan X Pascal.