Optimus: An Efficient Dynamic Resource Scheduler for Deep Learning Clusters

Yanghua Peng, Yixin Bao, Yangrui Chen, Chuan Wu, Chuanxiong Guo | The University of Hong Kong / Bytedance Inc. | EuroSys '18, April 23–26, 2018, Porto, Portugal | DOI: 10.1145/3190508.3190517

Problem

Deep learning training jobs are resource-intensive and long-running — the paper's running example, DeepSpeech2 on LibriSpeech, takes 3–5 days across 16 GPUs — and they are increasingly submitted to shared production clusters. Existing cluster schedulers fail these workloads in two ways. First, general schedulers such as Borg and YARN require the submitter to fix the resource allocation at submission time, so a running job cannot absorb transient idle cluster capacity without manual reconfiguration or resubmission. Second, general-purpose schedulers (Mesos, Borg, YARN, Corral, TetriSched) contain no optimizations for DL framework structure or DL job dynamics, and default policies (FIFO, DRF) are job-size unaware, so short jobs queue behind long ones even though DL training times span minutes (CNN-rand) to weeks (ResNet-50 on a TITAN X Pascal). A motivating measurement makes the cost concrete: with 20 containers of 5 CPU cores and 10 GB RAM each running ResNet-50 on ImageNet, peak training speed occurs at a specific 8-worker / 12-parameter-server split, while naive 1:1 PS-to-worker scaling or simply adding tasks degrades speed through network congestion.

Core Insight

Because DL jobs are iterative and converge in a predictable \(O(1/k)\) SGD-like fashion, a scheduler can fit two online models per job — a loss-convergence curve giving remaining work, and a resource-to-speed function giving throughput as a function of the parameter-server and worker counts — and then use their ratio as an estimated remaining completion time to drive greedy, marginal-gain-per-dominant-resource reallocation every scheduling interval.

Method

Optimus builds two models per job and schedules on top of them.

Convergence model (Progress). Loss points are outlier-filtered against the minimum loss in the subsequent 5 epochs and the maximum loss in the previous 5 epochs, then normalized by the maximum loss seen so far into \([0,1]\). A non-negative least squares (NNLS) solver continuously refits

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

where \(l\) is normalized training loss and \(k\) is the cumulative step index. The fit plus a convergence threshold \(\delta\) yields the remaining steps to convergence. Loss data can be sampled every few steps or averaged per epoch to bound solver input size.

Speed model (Speed). A per-step time model (Eqn. 2) decomposes a step into forward propagation \(m \cdot T_{forward}\), backward propagation \(T_{back}\), parameter/gradient transfer \(2\frac{S/p}{B/w'_\rho}\), PS-side updates \(T_{update}\cdot\frac{w'_\rho}{p}\), and linearly growing connection overhead \(\delta\cdot w + \delta'\cdot p\). From it come two fitted speed functions:

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

\[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} \tag{4, sync}\]

The \(\theta\) coefficients are fitted rather than measured term-by-term, which makes the model hardware-agnostic. They are learned by pre-running the job across several \((p,w)\) combinations on small data samples for tens of seconds.

Resource allocation. Minimizing \(\sum_j Q_j/f(p_j,w_j)\) under cluster capacity (Eqns. 5–8) is a non-linear, non-convex integer program that the paper states is NP-hard in general. Optimus instead uses a greedy heuristic on the marginal gain of Eqn. (9) — the JCT reduction from adding one PS or one worker, divided by that task's dominant-resource consumption. Every active job first gets 1 worker + 1 PS to avoid starvation; the job with the largest marginal gain then receives one more worker or one more PS (whichever term of Eqn. 9 is larger); gains are recomputed on change; the loop stops when resources run out or all marginal gains are non-positive. Early-stage jobs can have their marginal gain multiplied by a damping factor (e.g., 0.95) because their predictions are least reliable.

Task placement. Theorem 1 states that, for a synchronous job on homogeneous servers, the optimal placement uses the smallest number of servers and puts the same number of PS and the same number of workers on each — colocate workers with parameter servers, then spread evenly. The Appendix decomposes the min-max cross-server transmission problem into two lexicographical min-max subproblems (workers and PS), each optimized by even placement, and proves by induction that fewer servers means less inter-server traffic. The runtime algorithm sorts servers by descending available capacity and jobs by ascending requirement, finds the smallest \(k\) servers that fit each job, and spreads its tasks evenly.

Implementation and PAA. Built on Kubernetes 1.7 with MXNet; datasets in HDFS 2.8 are round-robin partitioned across workers and re-partitioned on every scaling event; stragglers (below 50% of median worker speed in async mode, or detected via gradient arrival timestamps at the PS in sync mode) are evicted and replaced; elastic scaling checkpoints parameters to HDFS, kills containers, and restarts with the new topology; the scheduler itself runs as a Kubernetes pod with job state in etcd. MXNet assigns any parameter block under \(10^6\) parameters randomly to a single PS and slices larger blocks equally across all PS, producing severe skew; the Parameter Assignment Algorithm (PAA) instead sends very small blocks (<1% of avg_size) to the PS with the fewest update requests, best-fits medium blocks (1%–100% of avg_size), and slices only blocks larger than avg_size.

Experimental Setup

Headline Quantitative Results

End-to-end (Fig. 11 normalized, Fig. 13 absolute):

Stated as improvements: 139% in average JCT and 63% in makespan over the fairness scheduler.

Overhead and scalability:

• Resource adjustment overhead: 2.54% of the makespan.
• Schedules 4,000 jobs (~100,000 tasks) on 16,000 nodes in under 5 seconds on one core of an Intel E5-1620 v4; Kubernetes' default scheduler reference point is 150,000 tasks / 5,000 nodes / 5 s.

Model accuracy:

• ~10 pre-run sample configurations keep speed-estimation error below 10%.
• Practical error levels: 10% for speed, 20% for convergence; at those levels the gap versus perfect prediction is about 15%.
• Priority factor 0.95 reduces JCT by 2.66% and makespan by 1.88%.

Robustness across workloads (normalized, Optimus = 1.00):

Gains are larger under all-synchronous training (more stable convergence and uniform worker speed reduce both estimation errors) and under the Google trace (many arrival spikes).

Ablations (normalized, Optimus = 1.00):

Allocation alone accounts for 62% JCT and 31% makespan reduction versus fairness; placement contributes ~10% versus Tetris and ~15% versus DRF. Overall contribution split: allocation 62%, placement 17%, PS load balancing 20%.

PS load balancing (Table 3, ResNet-50, 25 M parameters in 157 blocks):

PAA splits no block further (157 requests is the minimum for 157 blocks) and delivers up to 29% training speedup over MXNet's algorithm across ResNet-50, ResNext-101, Inception-BN and VGG at 10 workers and 10 PS, with the advantage growing as PS count rises.

Limitations

• Checkpoint/restart-based elasticity carries a scaling cost — 2.54% measured, larger for very big models or frequent rescaling.
• The Eqn. (1) fitting form assumes \(O(1/k)\) SGD-like loss decay and does not apply to algorithms such as A3C in deep reinforcement learning.
• Step-decay learning-rate schedules (e.g., ResNet's ×0.1 drop) invalidate the current fit and force a reset of the fitting process.
• Prediction error costs performance: ~15% gap at 20% convergence / 10% speed error.
• An initial profiling phase is required — pre-running each job with 5 (evaluation) to 5–10 (claimed sufficient) \((p,w)\) combinations.
• Data chunks must be re-partitioned across workers on every scaling event.
• PAA is implemented in MXNet only, although the imbalance problem is stated to be common across PS-based frameworks.
• Scope assumptions: data-parallel training on the parameter server architecture; mature production models with tuned hyper-parameters; the bandwidth bottleneck at parameter servers; homogeneous servers for Theorem 1; fixed global mini-batch \(M\) in sync mode; fixed per-container footprint, with only task counts scheduled.

Open Problems

1. Mixed-workload clusters. Running Optimus alongside other schedulers in Kubernetes, requesting resources from a central manager and scheduling DL jobs over a varying share of the cluster.
2. Broader convergence estimation. Handling step-decay learning-rate schedules by resetting the fit, and accepting user-supplied parametric convergence functions for non-SGD-like algorithms.
3. Bounded scaling frequency. A re-allocation threshold — smaller for large jobs — to cap the checkpoint/restart overhead of elasticity.