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
- Deep learning workloads are now common in production clusters because of the proliferation of DL-driven AI services (speech recognition, machine translation).
- A DL training job is both resource-intensive and time-consuming; efficient resource scheduling is the key to maximal DL-cluster performance.
- Existing cluster schedulers are largely not tailored to DL jobs and typically specify a fixed amount of resources per job, prohibiting high resource efficiency and job performance.
- Optimus is a customized scheduler that minimizes job training time based on online resource-performance models.
- Two models are built per job: online fitting to predict model convergence during training, and a performance model that estimates training speed as a function of allocated resources.
- On top of those models, "a simple yet effective method" dynamically allocates resources and places DL tasks to minimize job completion time.
- Implemented on Kubernetes; evaluated on a DL cluster of 7 CPU servers and 6 GPU servers running 9 training jobs on the MXNet framework.
- Headline: Optimus outperforms representative cluster schedulers by about 139% in job completion time and 63% in makespan.
"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.
- Deep learning has expanded across AI applications: speech recognition, natural language processing, computer vision.
- Scaling both model size and data volume increases learning accuracy but significantly lengthens training duration.
- Major IT companies deploy distributed ML/DL clusters of hundreds-to-thousands of CPU/GPU servers running frameworks such as TensorFlow and MXNet.
Cost of a training job.
- DL training jobs are highly resource-intensive and time-consuming; the paper's running example is DeepSpeech2 on LibriSpeech, which takes 3–5 days across 16 GPUs.
- Because clusters are shared, efficient resource scheduling is crucial both to maximize hardware utilization and to speed up individual job completion.
Limitation 1 — static resource allocation.
- Existing schedulers (Borg, YARN) require the job submitter to fix the resource allocation at submission time.
- A running job therefore cannot exploit transient idle cluster resources without manual reconfiguration or full resubmission, producing resource inefficiency.
Limitation 2 — no DL-specific optimizations.
- General-purpose schedulers (Mesos, Borg, YARN, Corral, TetriSched) contain no tailor-made optimizations for DL framework structure or DL job dynamics.
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.
- Optimus is a customized scheduler for data-parallel DL jobs on the parameter server (PS) framework that continuously models job performance and dynamically adjusts resource allocations.
- Objective: minimize average job completion time (JCT) and cluster makespan under dynamic cluster load.
Stated contributions.
- Performance modeling — online convergence-curve fitting plus a hardware-agnostic resource-to-speed model, requiring no knowledge of hardware internals or job internals.
- Scheduling algorithms — a marginal-gain-driven dynamic resource allocation algorithm and a communication-aware task placement scheme.
- Framework fix — identification and resolution of parameter-server load imbalance in MXNet.
- 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
- DL model training is defined as optimizing model parameters (e.g., DNN weights) over a large training dataset so as to minimize a predefined loss function.
2.1 DL Model Training
Iterativeness.
- Because DNNs are non-linear, there is no closed-form solution, and datasets are far too large to process at once; training therefore proceeds by iterative mini-batch updates.
- The update rule is
new_parameter = old_parameter − learning_rate × gradient. - An epoch is one complete pass over all mini-batches of the training dataset.
Convergence.
- Models train over many epochs until parameter changes plateau.
- The paper assumes mature production models with well-tuned hyper-parameters, so the non-convergence and severe-overfitting problems typical of experimental models are out of scope.
- Job completion criterion: the decrease in training loss across consecutive epochs falls below a threshold \(\delta\).
Why training loss (not accuracy / validation loss).
- Training loss is computed at every step, which enables fine-grained online fitting.
- It directly reflects convergence without requiring separate validation-evaluation loops that would cost extra compute.
2.2 The Parameter Server Architecture
- The PS architecture is used by MXNet, TensorFlow, PaddlePaddle, Angel and Petuum. Parameters are partitioned across parameter servers; training data is partitioned across workers.
- Workers compute gradients locally on their data shard and push them to the PS; each PS applies the update algorithm (e.g., SGD) and pushes the updated weights back (workers pull).
- Asynchronous training: workers run un-synchronized; a PS updates its parameters immediately upon receiving gradients from any worker.
- Synchronous training: workers synchronize per step; a PS updates only after gathering gradients from all workers.
2.3 Existing Cluster Schedulers
Static allocation in practice.
- PS and workers run in containers managed by general cluster schedulers, and are bound to a static task count for the job's entire lifetime.
Motivating measurement — resource configuration matters non-trivially.
- Training-speed sensitivity study on ResNet-50 / ImageNet with fixed per-container limits of 5 CPU cores and 10 GB RAM.
- With a fixed total of 20 containers, peak speed occurs at a specific split — 8 workers + 12 PS — not at a naive ratio.
- Arbitrary 1:1 PS:worker scaling, or simply adding more tasks, degrades speed due to network congestion.
Dynamics demand adaptation.
- Runtime conditions (e.g., available link bandwidth) change, so PS/worker counts must be adapted continuously rather than fixed at job entry.
Job-size unawareness.
- Default cluster policies (FIFO, DRF) ignore job size, so short jobs queue behind long-running jobs (head-of-line blocking).
- Training time spans minutes (CNN-rand) to weeks (ResNet-50 on TITAN X Pascal) — a spread Optimus exploits by estimating completion and prioritizing allocations dynamically.
3. Performance Modeling of DL Jobs
- Two quantities must be predicted per job: Progress — remaining epochs/steps to convergence — and Speed — the mapping from a resource allocation to epoch execution speed.
3.1 Learning the Convergence Curve
Data preprocessing.
- Outlier removal: a loss data point is an outlier if it does not fall within the range defined by its neighbours — between the minimum loss in the subsequent 5 epochs and the maximum loss in the previous 5 epochs — and is replaced by the neighbours' average.
- Normalization: each raw loss is divided by the maximum loss collected so far (typically the first value), mapping all losses into \([0,1]\).
- Training progress is defined as epochs trained ÷ total epochs to convergence.
- The loss curves of Fig. 5 were collected with MXNet on a server with 1 × E5-1650 v4 CPU and 2 × NVIDIA TITAN X GPUs, with a fixed learning rate.
Online fitting.
- Most DL jobs use SGD, which converges at rate \(O(1/k)\) in the number of steps \(k\); Eqn. (1) is chosen to match that shape.
- After each training step a loss point \((k, l)\) is collected, preprocessed, and fed to a non-negative least squares (NNLS) solver, which refits the coefficients over all points so far.
- When hundreds of thousands of steps are required, loss data can be sampled every few steps or averaged per epoch into a single point to bound solver input size.
- The fitted model improves continuously as more data arrives. Fig. 7 shows the fit for Sequence-to-Sequence.
Predicting remaining work.
- At each step, the fitted loss model plus the convergence threshold \(\delta\) yield the total number of steps/epochs to convergence.
- Fig. 6: prediction error decreases rapidly as training progresses.
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.
- Forward propagation: \(m \cdot T_{forward}\).
- Backward propagation: \(T_{back}\) (independent of mini-batch size).
- Parameter/gradient transfer (both directions): \(2\frac{S/p}{B/w'_\rho}\).
- Parameter updates at the PS: \(T_{update}\cdot\frac{w'_\rho}{p}\).
- Communication overhead (TCP connections, control messages), growing linearly: \(\delta\cdot w + \delta'\cdot p\).
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 |
- Implication drawn by the authors: workers should have similar processing speeds and PS should be load-balanced to minimize per-step time. These two implications directly motivate §5.2 (straggler handling) and §5.3 (PS load balancing).
Defining training speed.
- Overall job training speed \(f(p,w)\) is expressed in steps per unit time, and is derived separately for asynchronous and synchronous modes.
Asynchronous mode.
- Throughput is \(w \cdot T^{-1}\). Assuming concurrent PS connections scale linearly with workers (\(w'_\rho \propto w\)), Eqn. (3) follows with positive coefficients \(\theta\) corresponding to the respective terms of Eqn. (2) — e.g., \(\theta_0\) corresponds to \(m\cdot T_{forward}+T_{back}\).
- Crucially, the coefficients are fitted, not measured term-by-term, which is what makes the model hardware-agnostic.
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.
- The step rate is \(T^{-1}\) with \(w'_\rho = w\). The global batch size \(M\) is held fixed, so per-worker mini-batch is \(m = M/w\), giving Eqn. (4).
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.
- To learn the \(\theta\)'s for Eqns. (3) and (4), the scheduler pre-runs a new job across a set of \((p,w)\) combinations on small data samples for tens of seconds, then fits with NNLS.
Empirical validation.
- Fig. 9 (ResNet-50) validates speed-model accuracy, exposes diminishing returns caused by network overhead, and shows synchronous-mode speed degradation when too many workers shrink \(M/w\) below the efficient per-GPU compute threshold.
- Table 2's fitted coefficients indicate forward/backward compute and network transfer are the dominant bottleneck terms.
- Fig. 8: as few as ~10 sample configurations keep speed-estimation error below 10%.
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
- DL jobs arrive online.
"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.
- Over the set of active jobs \(\mathcal{J}\), minimize total remaining running time, where job \(j\)'s remaining time is "estimated as \(Q_j/f(p_j,w_j)\)", subject to cluster capacity.
\[\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\}\]
- First term: JCT reduction from adding one parameter server, divided by that PS's dominant-resource consumption.
- Second term: JCT reduction from adding one worker, divided by that worker's dominant-resource consumption.
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).
- Allocate 1 worker + 1 PS to every active job first, to avoid starvation.
- Sort jobs by marginal gain from Eqn. (9).
- Iteratively pick the job with the largest marginal gain and add either one worker or one PS, whichever of the two terms in (9) is larger.
- Recompute marginal gains for any job whose allocation changed.
- Stop when cluster resources are exhausted or all marginal gains are non-positive.
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.
- Step processing time depends heavily on inter-node gradient/parameter transmission, so Optimus maximizes speed by minimizing cross-node communication through placement.
Worked example (Fig. 10).
- Setting: 3 physical servers, each hosting 3 containers; a synchronous job with 2 PS and 4 workers; unit bandwidth per task and unit gradient volume.
- Placement (a) produces per-task cross-server transfer counts \((3,3,1,1,2,2)\), giving a step transfer time of 3.
- Placement (b) also gives 3.
- Placement (c) — the symmetric, colocated and evenly-spread layout — gives 2, the minimum. Caption: "An example of worker/parameter server placement: (c) is the best."
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."
- The two underlying principles are (a) colocating workers and parameter servers and (b) even spreading. The detailed proof is in the Appendix.
Appendix — the placement optimization.
- Cross-server data transmission time per synchronous step:
\[\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\}\]
- The placement problem minimizes that maximum subject to \(\sum_k p_{jk}=p_j\), \(\sum_k w_{jk}=w_j\), \(p_{jk}, w_{jk}\in\mathbb{Z}^+\).
- It decomposes into two lexicographical min-max subproblems:
\[\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\) |
- Each subproblem's optimum is to place tasks evenly; combining them yields even PS placement and even worker placement over the \(K\) nodes.
- The remaining claim — that a smaller \(K\) yields smaller transmission time — is proved by mathematical induction: fewer nodes means more PS and workers per node, so less data crosses the inter-server network.
- Node capacities are assumed sufficient to place the job, and servers are assumed homogeneous.
The placement algorithm (prose).
- Sort servers in descending order of available capacity.
- Sort jobs in ascending order of total resource requirement (smallest first, to avoid starvation).
- For each job, find the smallest \(k\) such that the top \(k\) servers can host all of its tasks, then distribute PS and workers as evenly as possible across those \(k\) servers.
- Deduct the consumed capacities and re-sort. Jobs that cannot fit are deferred to the next scheduling interval.
5. System Implementation
- Optimus is implemented on top of Kubernetes with MXNet as the training framework.
5.1 Data Serving
- Datasets are stored in HDFS with default chunk size 128 MB and replication factor 2.
- At job submission, chunks are partitioned round-robin across assigned workers.
- On any dynamic worker-count change, chunks are re-partitioned across the updated worker set.
5.2 Straggler Handling
- In synchronous training a single straggler delays every step because all workers must synchronize.
"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."
- Async detection rule: a worker is a straggler if its processing speed is below 50% of the median worker speed.
- Sync detection rule: worker step times are measured via gradient arrival timestamps at the PS.
- Mitigation: stragglers are evicted and replaced with fresh worker containers.
5.3 Load Balancing on Parameter Servers
The bug in MXNet (and TensorFlow).
- MXNet assigns any parameter block smaller than \(10^6\) parameters randomly to a single PS, and slices blocks larger than \(10^6\) equally across all PS.
- This fixed threshold produces severe compute and communication skew across parameter servers.
Three optimization targets.
- Minimize the maximum difference in total parameter size assigned to any two PS.
- Minimize the total number of parameter-update network requests per step.
- Minimize the maximum difference in the number of update requests across PS.
Parameter Assignment Algorithm (PAA).
- Compute \(\text{avg\_size} = (\text{total parameter bytes}) / p\) and sort blocks in descending order of size.
- Very small blocks (\(< 1\%\) of avg_size) → assigned to the PS with the fewest update requests.
- Medium blocks (\(1\%\)–\(100\%\) of avg_size) → best-fit into the PS with the smallest remaining headroom.
- Large blocks (\(>\) avg_size) → sliced into partitions of size \(\le\) avg_size, each assigned to the PS with the smallest total assigned size.
5.4 Elastic Training on MXNet
- When an allocation changes, Optimus checkpoints model parameters to HDFS, terminates the current containers, and restarts the job with the new worker/PS topology, resuming from the checkpoint.
5.5 Scheduler on Kubernetes
- Optimus runs as a master pod (a Kubernetes deployment coupling one or more containers) on Kubernetes 1.7, polling the Kubernetes master for cluster information and job states.
- For fault tolerance it stores job states in etcd, a distributed reliable key-value store; Kubernetes automatically restarts the scheduler on failure.
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.
- A discrete-time simulator extends the study to larger scale and more parameter choices; it is driven by testbed traces: per-job-type training losses, training speeds under different resource configurations, per-server resource capacities, job configurations (worker/PS resource requirements), and DL model details such as parameter size.
Workload.
- Job arrivals uniformly random in \([0, 12000]\) seconds; job type chosen randomly from Table 1; training mode (sync/async) chosen randomly.
- Convergence threshold varied between 1% and 5%.
- Datasets for the large models (ResNet-50, DeepSpeech2) were downscaled so an experiment finishes in reasonable time; the authors verified the jobs still converge.
- Each run takes about 6 hours; each experiment is repeated 3 times and averaged.
Baselines.
- DRF fairness scheduler — dominant resource fairness as used in Hadoop / YARN / Mesos, dynamically rescheduling jobs each interval, with Kubernetes-default load-balancing placement.
- 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.
- Both baselines use a PS:worker ratio of 1:1.
Metrics.
- Average JCT — indicator of system performance.
- Makespan — indicator of resource efficiency, defined as "the total time elapsed from the arrival of the first job to the completion of all jobs"; minimizing makespan is equivalent to maximizing resource efficiency.
Default parameters.
- Speed function initialized by pre-running each job on a small dataset with 5 different \((p,w)\) combinations.
- Scheduling interval 10 minutes; priority factor set to 1 by default; the very-small-parameter-block threshold of §5.3 set to 1% of avg_size.
6.2 Performance
End-to-end comparison.
- Fig. 11 (normalized, Optimus = 1.00):
| Metric | Optimus | DRF | Tetris |
|---|---|---|---|
| Avg. JCT | 1.00 | 2.39 | 1.74 |
| Makespan | 1.00 | 1.63 | 1.22 |
- Fig. 13 (absolute averages, with standard deviations across the 3 repeated runs):
| 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) |
- Fig. 14 tracks the number of running tasks and normalized CPU utilization (CPU utilization divided by the overall allocated CPU capacity on a PS or worker) over a ~\(2.4\times10^4\) s experiment.
- Optimus deliberately runs fewer tasks than DRF: DRF is work-conserving and allocates as many resources as possible, but more resources do not imply higher training speed (as established in §3.2).
- Optimus's normalized CPU utilization on both workers and PS exceeds DRF's and Tetris's — it uses the resources it holds more efficiently.
Resource adjustment overhead.
- Overhead of changing from one \((p,w)\) configuration to another, measured as the fraction of time spent adjusting resources, is 2.54% of the makespan — acceptable relative to the performance gain.
Scalability.
- Emulated submission and scheduling of many jobs across thousands of nodes (Fig. 12 sweeps 1000 / 2000 / 4000 / 8000 jobs over \(10^3\)–\(10^4\) nodes).
- Optimus schedules 4,000 jobs (~100,000 tasks) within 5 seconds on a cluster of 16,000 nodes, measured on one core of an Intel E5-1620 v4 CPU.
- Reference point: Kubernetes' default scheduler handles 150,000 tasks in 5,000 nodes within 5 seconds.
6.3 Sensitivity Analysis
Error-injection setup.
- Errors are injected into convergence-epoch and training-speed estimates via the perturbation \(v\cdot(1\pm e)\), with the injected error decaying as training progresses.
- Results averaged over 100 simulation runs.
Prediction-error sensitivity (Fig. 15).
- Both JCT and makespan increase with error, but with diminishing slope.
- At 20% convergence-estimation error and 10% speed-estimation error there is about a 15% performance gap versus zero error.
- Speed-estimation error hurts more than convergence-estimation error — fortunately speed is estimable far more accurately (10% error) than convergence (20% error).
Priority factor.
- With the priority factor set to 0.95, average JCT and makespan are 2.66% and 1.88% smaller respectively.
Training modes (Fig. 16).
- Replacing random mode selection with all-async or all-sync:
| 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 |
- Optimus's gain is larger under all-synchronous training: under sync all workers hold the most up-to-date parameters, so convergence is more stable and convergence-estimation error is smaller; also all workers run at the same speed, so speed-estimation error is smaller.
Arrival processes (Fig. 17).
- Two additional arrival processes: a Poisson process with 3 arrivals per scheduling interval, and one extracted from Google cluster workload traces over a 7-hour period.
| 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 |
- The gain is larger under the Google traces because they contain many arrival spikes, which Optimus absorbs better through efficient resource allocation.
6.4 Inspecting Detailed Designs in Optimus
Resource allocation ablation (Fig. 18).
- Optimus's allocation is swapped for the fairness scheduler's or Tetris's, keeping Optimus's placement.
| Metric | Optimus | DRF | Tetris |
|---|---|---|---|
| Avg. JCT | 1.00 | 1.62 | 1.33 |
| Makespan | 1.00 | 1.31 | 1.14 |
- Text result: average completion time and makespan are reduced by 62% and 31% vs. the fairness scheduler — the allocation algorithm is the critical component.
Task placement ablation (Fig. 19).
- Optimus's placement is swapped for the fairness scheduler's (load-balancing) or Tetris's (fragmentation-minimizing), keeping Optimus's allocation.
| Metric | Optimus | DRF | Tetris |
|---|---|---|---|
| Avg. JCT | 1.00 | 1.17 | 1.12 |
| Makespan | 1.00 | 1.13 | 1.09 |
- Text result: Optimus's placement reduces average completion time and makespan by about 10% vs. Tetris and 15% vs. DRF.
PS load balancing — parameter distribution (Table 3).
- Three factors capture PS load imbalance and overhead: difference of parameter sizes among PS, difference of number of update requests among PS, and total number of update requests between PS and workers.
- Measured on ResNet-50 with 25 million parameters formed into 157 parameter blocks.
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 |
- PAA splits no parameter block further — 157 requests is the minimum achievable for 157 blocks — while simultaneously keeping the parameter-size difference (0.1M) and request-count difference (1) minimal.
PS load balancing — speedup (Figs. 20–21).
- Fig. 20: ResNet-50 on ILSVRC2012-ImageNet, workers fixed at 10, PS count varied over 4/8/12/16/20, synchronous training. PAA's advantage grows with the number of PS.
- Fig. 21: across ResNet-50, ResNext-101, Inception-BN and VGG under synchronous training with 10 workers and 10 PS, PAA achieves up to 29% speedup over MXNet's algorithm.
- Similar results were observed with asynchronous training.
Summary of results (as stated by the authors).
- 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%.
- Further improvement of estimation accuracy would not increase Optimus's performance much (~15%), and Optimus beats DRF and Tetris under various workloads.
- 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.
- Optimus targets DL jobs but can be deployed in clusters with mixed workloads (data analytics, online services): Kubernetes supports plugging in multiple schedulers, each responsible for one kind of workload, with Optimus requesting resources from a central cluster resource manager and scheduling DL jobs on a varying portion of the cluster.
Convergence estimation.
- For some models (e.g., ResNet) the learning rate is reduced sharply (e.g., ×0.1) once training reaches a predefined condition, in order to push loss lower. Optimus can reset the online loss-curve fitting, treating the post-adjustment phase as a new training job.
- For algorithms whose loss curves do not follow the \(O(1/k)\) SGD model of Eqn. (1) — the paper's example is A3C in deep reinforcement learning — users can supply their own parametric convergence functions.
Scaling overhead.
- Adjusting worker/PS counts relies on checkpointing to HDFS and restarting tasks; for very large models or many instances, frequent scaling introduces non-trivial overhead.
- Mitigation: impose a threshold on the number/frequency of re-allocations, with a smaller threshold for large jobs.
8. Related Work
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
- The parameter server architecture was first introduced by Smola and Narayanamurthy, and improved with update primitives, fault tolerance and communication optimization in later work.
- Most distributed ML frameworks — MXNet, Petuum, TensorFlow, Angel — are implicitly or explicitly built on this architecture.
- Optimus targets scheduling jobs on these frameworks, and additionally finds that the PS load-imbalance problem is common across them, proposing and implementing PAA inside MXNet.
9. Conclusion
- Optimus is a customized cluster scheduler targeting high training performance and resource efficiency in DL clusters.
- Its core is an accurate performance model for DL workloads, built by exploiting DL training characteristics (convergence property, iterativeness) and the communication patterns of the parameter server architecture.
- On that model rest a marginal-gain-based resource allocation algorithm and a training-speed-maximizing task placement scheme.
- Experiments on a Kubernetes cluster show Optimus significantly outperforms representative cluster schedulers.
Assumptions Made by Optimus
- Workload is data-parallel DL training on the parameter server architecture (§1, §2.2).
- Production models are mature with tuned hyper-parameters, so convergence is stable and overfitting/divergence is out of scope (§2.1).
- Job completion is defined by the training-loss decrease over consecutive epochs falling below \(\delta\) (§2.1).
- SGD's \(O(1/k)\) convergence rate justifies the functional form of Eqn. (1) (§3.1).
- The network bandwidth bottleneck lies at the parameter servers, not the workers (§3.2).
- Theorem 1's optimality assumes homogeneous servers and node capacity sufficient to place the job (§4.2, Appendix).
- In synchronous training the global mini-batch \(M\) is fixed, so \(m = M/w\) (§3.2).
- Per-container resource footprint is user-specified and fixed; only task counts are scheduled (§2.3).
Limitations
- Checkpoint/restart elasticity introduces scaling overhead — measured at 2.54%, but larger for very big models or frequent scaling (§5.4, §6.2, §7).
- 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).
- Step-decay learning-rate schedules invalidate the current fit and force a reset of the fitting process (§7).
- Prediction error costs performance — about a 15% gap at 20% convergence / 10% speed error (§6.3).
- 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).
- Data chunks must be re-partitioned across workers on every scaling event (§5.1).
- 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)
- 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.
- Broader convergence estimation — reset fitting for step-decay learning rates, and accept user-supplied parametric convergence functions for non-SGD-like algorithms.
- 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.