# Toward Efficient Online Scheduling for Distributed Machine Learning Systems

Menglu Yu, Jia Liu , Chuan Wu , Bo Ji Elizabeth S. Bentley  Menglu Yu is with
the Department of Computer Science, Iowa State University, Ames, IA 50011, USA
(e-mail: mengluy@iastate.edu). Jia Liu is with the Department of Electrical
and Computer Engineering, The Ohio State University, Columbus, OH 43210, USA
(e-mail: liu@ece.osu.edu). Bo Ji is with the Department of Computer Science,
Virginia Tech, Blacksburg, VA 24061, USA (e-mail: boji@vt.edu). Chuan Wu is
with the Department of Computer Science, The University of Hong Kong, Hong
Kong (e-mail: cwu@cs.hku.hk). Elizabeth S. Bentley is with the Air Force
Research Laboratory, Information Directorate, Rome, NY 13441, USA (e-mail:
elizabeth.bentley.3@us.af.mil). Digital Object Identifier
xx.xxxx/XXXX.xxxx.xxxxx.

###### Abstract

Recent years have witnessed a rapid growth of distributed machine learning
(ML) frameworks, which exploit the massive parallelism of computing clusters
to expedite ML training. However, the proliferation of distributed ML
frameworks also introduces many unique technical challenges in computing
system design and optimization. In a networked computing cluster that supports
a large number of training jobs, a key question is how to design efficient
scheduling algorithms to allocate workers and parameter servers across
different machines to minimize the overall training time. Toward this end, in
this paper, we develop an online scheduling algorithm that jointly optimizes
resource allocation and locality decisions. Our main contributions are three-
fold: i) We develop a new analytical model that considers both resource
allocation and locality; ii) Based on an equivalent reformulation and
observations on the worker-parameter server locality configurations, we
transform the problem into a mixed packing and covering integer program, which
enables approximation algorithm design; iii) We propose a meticulously
designed approximation algorithm based on randomized rounding and rigorously
analyze its performance. Collectively, our results contribute to the state of
the art of distributed ML system optimization and algorithm design.

###### Index Terms:

Online resource scheduling, distributed machine learning, approximation
algorithm

##  1 Introduction

Fueled by the rapid growth of data analytics and machine learning (ML)
applications, recent years have witnessed an ever-increasing hunger for
computing power. However, with hardware capability no longer advancing at the
pace of the Moore’s law, it has been widely recognized that a viable solution
to sustain such computing power needs is to exploit parallelism at both
machine and chip scales. Indeed, the recent success of deep neural networks
(DNN) is enabled by the use of distributed ML frameworks, which exploit the
massive parallelism over computing clusters with a large number of GPUs. These
distributed ML frameworks have significantly accelerated the training of DNN
for many applications (e.g., image and voice recognition, natural language
processing, etc.). To date, prevailing distributed ML frameworks include
TensorFlow [1], MXNet [2], PyTorch [3], Caffe [4], to name just a few.

However, the proliferation of distributed ML frameworks also introduces many
unique technical challenges on large-scale computing system design and network
resource optimization. Particularly, due to the decentralized nature, at the
heart of distributed learning system optimization lies the problem of
scheduling ML jobs and resource provisioning across different machines to
minimize the total training time. Such scheduling problems involve dynamic and
combinatorial worker and parameter server allocations, which are inherently
NP-hard. Also, the allocations of workers and parameter servers should take
locality into careful consideration, since co-located workers and parameter
servers can avoid costly network communication overhead. However, locality
optimization adds yet another layer of difficulty in scheduling algorithm
design. Exacerbating the problem is the fact that the future arrival times of
training jobs at an ML computing cluster are hard to predict, which
necessitates online algorithm design without the knowledge of future job
arrivals. So far in the literature, there remains a lack of holistic
theoretical studies that address all the aforementioned challenges. Most of
the existing scheduling schemes are based on simple heuristics without
performance guarantee (see Section 2 for detailed discussions). This motivates
us to fill this gap and pursue efficient online scheduling designs for
distributed ML resource optimization, which offer provable performance
guarantee.

The main contribution of this paper is that we develop an online scheduling
algorithmic framework that jointly yields resource scheduling and locality
optimization decisions with strong competitive ratio performance. Further, we
reveal interesting insights on how distributed ML frameworks affect online
resource scheduling optimization. Our main technical results are summarized as
follows:

  * •

By abstracting the architectures of prevailing distributed ML frameworks, we
formulate an online resource scheduling optimization problem that: i) models
the training of ML jobs based on the parameter server (PS) architecture and
stochastic gradient descent (SGD) method; and ii) explicitly takes locality
optimization into consideration. We show that, due to the heterogeneous
internal (between virtual machines or containers) and external (between
physical machines) communications, the locality-aware scheduling problem
contains non-deterministic constraints and is far more complex compared to the
existing works that are locality-oblivious (see, e.g., [5, 6]).

  * •

To solve the locality-aware scheduling problem, we develop an equivalent
problem reformulation to enable subsequent developments of online
approximation algorithms. Specifically, upon carefully examining the locality
configurations of worker-server relationships, we are able to transform the
original problem to a special-structured integer nonlinear program with mixed
cover/packing-type constraints, and the low-complexity approximation algorithm
design with provable performance can be further entailed.

  * •

To tackle the integer nonlinear problem with mixed cover/packing-type
constraints, we propose an approximation algorithm based on a meticulously
designed randomized rounding scheme and then rigorously prove its performance.
We note that the results of our randomized rounding scheme are general and
could be of independent theoretical interest. Finally, by putting all
algorithmic designs together, we construct a primal-dual online resource
scheduling (PD-ORS) scheme, which has an overall competitive ratio that only
logarithmically depends on ML job characteristics (e.g., required epochs,
training samples).

Collectively, our results contribute to a comprehensive and fundamental
understanding of distributed machine learning system optimization. The
remainder of this paper is organized as follows. In Section 2, we review the
literature to put our work in comparative perspectives. Section 3 introduces
the system model and problem formulation. Section 4 presents our algorithms
and their performance analysis. Section 5 presents numerical results and
Section 6 concludes this paper.

##  2 Related Work

As mentioned in Section 1, due to the high computational workload of ML
applications, many distributed ML frameworks (e.g., TensorFlow [1], MXNet [2],
PyTorch [3], Caffe [4]) have been proposed to leverage modern large-scale
computing clusters. A common distributed training architecture implemented in
these distributed ML frameworks is the PS architecture [7, 8], which employs
multiple workers and PSs (implemented as virtual machines or containers) to
collectively train a global ML model. Coupled with the iterative ML training
based on stochastic gradient descent (SGD), the interactions between machines
in the distributed ML cluster are significantly different from those in
traditional cloud computing platforms (e.g., MapReduce [9] and Dryad [10] and
references therein). For example, a MapReduce job usually partitions the input
data into independent chunks, which are then processed by the map step in a
parallel fashion. The output of the maps are then fed to the reduce step to be
aggregated to yield the final result. Clearly, the data flows in MapReduce are
a “one-way traffic”, which is unlike those iterative data flows in ML training
jobs whose completions highly depend on the ML job’s convergence property. As
a result, existing job scheduling algorithms for cloud systems are not
suitable for distributed ML frameworks.

Among distributed ML system studies, most of the early attempts (see, e.g.,
[7, 8] and references therein) only considered static allocation of workers
and parameter servers. To our knowledge, the first work on understanding the
performance of distributed ML frameworks is [11], where Yan et al. developed
analytical models to quantify the impacts of model-data partitioning and
system provisioning for DNN. Subsequently, Chun et al. [5] developed heuristic
dynamic system reconfiguration algorithms to allocate workers and parameter
servers to minimize the runtime, but without providing optimality guarantee.
The first dynamic distributed scheduling algorithm with optimality guarantee
was reported in [12], where Sun et al. used standard mixed integer linear
program (MILP) solver to dynamically compute the worker-parameter server
partition solutions. Due to NP-hardness of the MILP, the scalability of this
approach is limited. The most recent work [13] proposed an online scheduling
algorithm to schedule synchronous training jobs in ML clusters with the goal
to minimize the weighted completion time. However, the consecutive time slots
were allocated for each training job, and the numbers of workers and parameter
servers could not be adjusted.

Another line of the research is to leverage the learning-based approach to do
the resource scheduling. There are a number of recent works using deep
reinforcement learning (DRL) for resource allocation, device placement, and
video streaming. For example, Mao et al. [14] and Chen et al. [15] designed a
multi-resource cluster scheduler using DRL with the goal to minimize average
job slowdown. The proposed scheduler picks one or more of the waiting jobs in
the queue and allocate to machines at each time slot, and the resource demand
of each job is unknown until after its arrival. Later, Mao et al. [16, 17]
used DRL to heuristically train scheduling policies for graph-based parallel
jobs by setting both parallelism level and execution order. Meanwhile,
Mirhoseini et al. [18, 19] utilized DRL to design a model for efficient
placement of computational graphs onto hardware devices, aiming at minimize
the running time of each individual TensorFlow job. Although various
performance gains have been empirically reported, these DRL-based studies do
not offer optimality performance guarantee due to the lack of theoretical
foundation of DRL as of today.

Figure 1: Illustration of distributed training with the PS architecture.

Figure 2: The workflow of iterative training.

Figure 3: Colocated parameter servers and workers on physical machines.

The most relevant work to ours is [6], where Bao et al. developed an online
primal-dual approximation algorithm, OASiS, to solve the scheduling problem
for distributed ML systems. Our work differs from [6] in the following key
aspects: 1) In [6], the workers and parameter servers are allocated on two
strictly separated sets of physical machines, i.e., no worker and parameter
server can share the same physical machine, which significantly simplified the
underlying optimization problem. In this work, we consider the cases that
workers and parameter servers can be co-located on the same physical machine,
which is the common practice in existing ML systems (see, e.g., [20, 21]).
Such co-location can significantly reduce inter-server communication, expedite
training, and improve resource utilization efficiency between workers and
parameter servers. However, as will be shown later, the co-location setting
leads to an integer non-convex optimization problem with non-deterministic
constraints, which is much harder and necessitates new algorithm design. 2)
Ref. [6] advocates dynamic worker number adjustment, but does not guarantee
the same global batch size across the training iterations. According to recent
literature [22], maintaining a consistent global batch size is important for
ensuring convergence of DNN training, when the worker number varies. We ensure
a consistent global batch size in our model. We note that the co-location
setting was considered in [23]. However, the scheduling algorithm therein is a
heuristic and does not provide performance guarantee. This motivates us to
develop new algorithms with provable performance to fill this gap.

##  3 System Model and Problem Formulation

In this section, we first provide a quick overview on the architecture of
distributed ML frameworks to familiarize readers with the necessary
background. Then, we will introduce our analytical models for ML jobs and
resource constraints, as well as the overall problem formulation.

1) Distributed Machine Learning: A Primer. As illustrated in Fig. 3, the key
components of a PS-based distributed ML system include parameter servers,
workers, and the training dataset, which are usually implemented over a
connected computing cluster that contains multiple physical machines. The
training dataset of an ML job is stored in distributed data storage (e.g.,
HDFS [24]) and usually divided into equal-sized data chunks. Each data chunk
further contains multiple equal-sized mini-batches.

To date, one of the most widely adopted training algorithms in distributed ML
frameworks is the stochastic gradient descent method (SGD) [25]. With SGD, the
interactions between workers and parameter servers are illustrated in Fig 3. A
worker is loaded with the DNN model (we focus on data parallel training) with
current values of the model parameters (e.g., the weights of a DNN) and
retrieves a new data chunk from the data storage. In each training iteration,
a worker processes one mini-batch from its data chunk to compute gradients
(i.e., directions and magnitudes of parameter changes).111 As an example, in a
DNN model, gradients can be computed by the well-known “back-propagation”
approach.  Upon finishing a mini-batch, the worker sends the gradients to the
parameter servers, receives updated global parameters, and then continues with
the next training iteration to work on the next mini-batch. On the parameter
server side, parameters are updated as:
$\mathbf{w}[k]=\mathbf{w}[k-1]+\alpha_{k}\mathbf{g}[k]$, where
$\mathbf{w}[k]$, $\alpha_{k}$, and $\mathbf{g}[k]$ denote the parameter
values, step-size, and stochastic gradient in the $k$-th update, respectively.

2) Modeling of Learning Jobs: We consider a time-slotted system. The
scheduling time-horizon is denoted as $\mathcal{T}$ with $|\mathcal{T}|=T$. We
use $\mathcal{I}$ to represent the set of training jobs and let $a_{i}$ denote
the arrival time-slot of job $i\in\mathcal{I}$. As shown in Fig. 3, parameter
servers and workers could spread over multiple physical machines. We let
$\mathcal{H}$ represent the set of physical machines. For each job $i$, we use
$w_{ih}[t],s_{ih}[t]\geq 0$ to represent the allocated numbers of workers and
parameter servers on machine $h\in\mathcal{H}$ in each time-slot $t\geq
a_{i}$, respectively. Further, we let
$\mathcal{P}_{i}[t]\\!\triangleq\\!\\{h\\!\in\\!\mathcal{H}|s_{ih}[t]\\!>\\!0\\}$
and
$\mathcal{W}_{i}[t]\\!\triangleq\\!\\{h\\!\in\\!\mathcal{H}|w_{ih}[t]\\!>\\!0\\}$
denote the sets of physical machines that contain parameter servers and
workers for job $i$ in time-slot $t$, respectively.

We use a binary variable $x_{i}\in\\{0,1\\}$ to indicate whether job $i$ is
admitted ($x_{i}=1$) or not ($x_{i}=0$). We use $\tau_{i}$ to denote the
training for each sample of job $i$. We let $b_{i}(h,p)$ denote the data rate
of the link between a worker for job $i$ (on machine $h$) and a parameter
server (on machine $p$). Each worker or parameter server is exclusively
assigned to some job $i$, and $b_{i}(h,p)$ is reserved and decided by the user
upon job submission, which is common to ensure the data transfer performance
[6]. Note that the value of $b_{i}(h,p)$ is locality-dependent where the
slowest worker will become the bottleneck since we focus on Bulk-Synchronous-
Parallel (BSP) scheme [26]. Specifically, we have:

$\displaystyle b_{i}(h,p)=\begin{cases}b_{i}^{(i)},&\text{if $h=p$},\\\
b_{i}^{(e)},&\text{otherwise},\end{cases}$

where $b_{i}^{(i)}$ and $b_{i}^{(e)}$ denote the internal and external
communication link rates, respectively. For example, as shown in Fig. 3, since
Job 1’s worker $\mathrm{W}_{4}$ and parameter server $\mathrm{PS}_{2}$ are
both on the same machine, they communicate at the internal link rate
$b_{1}^{(i)}$. On the other hand, since Job 1’s worker $\mathrm{W}_{3}$ and
parameter server $\mathrm{PS}_{2}$ are on different physical machines, they
communicate at the external link rate $b_{1}^{(e)}$. In practice, it usually
holds that $b_{i}^{(e)}\ll b_{i}^{(i)}$.

Next, we calculate the amount of time for a worker on machine $h$ to process a
sample. We use $F_{i}$ to denote the global batch size of job $i$, which is a
fixed constant across all time-slots.222We note that this fixed global batch
size requirement is compliant with the standard SGD implementation [27] and
important for ensuring convergence [22]. In contrast, the global batch size in
some existing works on dynamic ML resource allocation (e.g., [6]) could be
time-varying, which necessitates time-varying dynamic learning rate
adjustments to offset correspondingly and further complicates the SGD
implementation.  We assume $F_{i}$ is equally divided among workers, i.e., the
local batch size at each worker is:
$F_{i}/\sum_{h^{\prime}\in\mathcal{H}}w_{ih^{\prime}}[t]$.333 Most distributed
ML frameworks (e.g., Tensorflow [28]) set the same local batch size to each
worker for the distributed training.

We assume symmetric link speed in both directions between a worker and a PS.
Let $g_{i}$ denote the size of gradients/parameters of job $i$. Then, from a
worker’s perspective, to push gradients to and pull updated parameters from
the PSs for job $i$, the combined uplink/downlink communication time can be
computed as:
$(2{g_{i}/\sum_{h^{\prime}\in\mathcal{H}}s_{ih^{\prime}}[t])/(\min_{p\in\mathcal{P}_{i}[t]}b_{i}(h,p)})$,
where the numerator term
$g_{i}/\sum_{h^{\prime}\in\mathcal{H}}s_{ih^{\prime}}[t]$ follows from the
assumption of even parameter distribution among the PSs, and the denominator
is due to the fact that push/pull time is decided by the slowest link among
all connections from the worker to all PSs (i.e., $\min_{p\in
P_{i}[t]}b_{i}(h,p)$). Hence, the average computation and communication time
to process a sample on machine $h\in\mathcal{W}_{i}[t]$ for job $i$ in time
slot $t$ can be computed as:

$\displaystyle\underbrace{\tau_{i}}_{\begin{subarray}{c}\mathrm{Training\,\,time}\\\
{\mathrm{per\,\,sample}}\end{subarray}}\\!\\!\\!+\underbrace{\bigg{(}\frac{2g_{i}/\sum_{h^{\prime}\in\mathcal{H}}s_{ih^{\prime}}[t]}{\min_{p\in\mathcal{P}_{i}[t]}b_{i}(h,p)}\bigg{)}\bigg{/}\bigg{(}\frac{F_{i}}{\sum_{h^{\prime}\in\mathcal{H}}w_{ih^{\prime}}[t]}\bigg{)}}_{\mathrm{Communication\,\,time\,\,per\,\,sample}}.$

Recall that we focus on the BSP scheme, where all workers are synchronized
before they proceed to the next iteration. In other words, the total number of
samples trained on machine $h\in\mathcal{W}_{i}[t]$ for job $i$ in time slot
$t$ is determined by the slowest link among all connections from all workers
to all PS (i.e.,
$\min_{p\in\mathcal{P}_{i}[t],h^{\prime}\in\mathcal{W}_{i}[t]}b_{i}(h^{\prime},p)$).
It then follows that the number of samples trained on machine
$h\in\mathcal{W}_{i}[t]$ for job $i$ in time-slot $t$ can be computed as:

$\displaystyle\frac{w_{ih}[t]}{\tau_{i}+\Big{(}\frac{2g_{i}/\sum_{h^{\prime}\in\mathcal{H}}s_{ih^{\prime}}[t]}{\min_{p\in\mathcal{P}_{i}[t],h^{\prime}\in\mathcal{W}_{i}[t]}b_{i}(h^{\prime},p)}\Big{)}\Big{/}\Big{(}\frac{F_{i}}{\sum_{h^{\prime}\in\mathcal{H}}w_{ih^{\prime}}[t]}\Big{)}}.$
(1)

Note that in practice, ML users usually specify a fixed ratio between worker
number and PS number (e.g., often 1:1 444In practice, the ratio between
numbers and parameter servers are specified by the user upon the job’s
submission (e.g., 1:1 in Kubernetes [29]).) when launching their training jobs
to ensure appropriate coordinations between workers and PSs in terms of
channel bandwidth, memory allocation, etc. To model this practice, we define
the ratio of worker number to PS number for each job $i$ as:

$\displaystyle\gamma_{i}\triangleq\frac{\sum_{h^{\prime}\in\mathcal{H}}w_{ih^{\prime}}[t]}{\sum_{h^{\prime}\in\mathcal{H}}s_{ih^{\prime}}[t]},\quad\forall
i,t.$ (2)

With $\gamma_{i}$, we can rewrite Eq. (1) as:

$\displaystyle\frac{w_{ih}[t]}{\tau_{i}+\frac{\gamma_{i}}{F_{i}}\frac{2g_{i}}{\min_{p\in\mathcal{P}_{i}[t],h^{\prime}\in\mathcal{W}_{i}[t]}b_{i}(h^{\prime},p)}}.$

Suppose that, for job $i$, there are $K_{i}$ data samples in its training
dataset. In practice, $K_{i}\gg F_{i}$. In ML systems, an epoch is defined as
a round of training that exhausts all data samples. We let $E_{i}$ denote the
number of epochs needed by job $i$. In this paper, we assume that the epoch of
each job is predetermined. This is because it is often difficult to estimate
the required number of epochs for SGD-type methods’ convergence. Therefore,
most SGD-type algorithms in practice stop after a fixed number of iterations
(i.e., fixed number of epochs, see, e.g., [30] and references therein) to
avoid excessive training delay.

Then, the total number of samples to be processed for job $i$ over the entire
training process is $E_{i}K_{i}$. To make sure that there are sufficient
workers allocated for job $i$ over the entire training horizon, we have:

$\displaystyle\\!\\!\\!\\!\\!\\!\sum_{t\in\mathcal{T}}\sum_{h\in\mathcal{H}}\frac{w_{ih}[t]}{\tau_{i}+\frac{\gamma_{i}}{F_{i}}\frac{2g_{i}}{\min_{p\in\mathcal{P}_{i}[t],h^{\prime}\in\mathcal{W}_{i}[t]}b_{i}(h^{\prime},p)}}\\!\geq\\!x_{i}E_{i}K_{i},\forall
i\in\mathcal{I}.\\!\\!\\!\\!$ (3)

We note that, with co-located workers and parameter servers on each machine,
Eq. (3) is non-deterministic due to the existence of the $\min\\{\cdot\\}$
operator. As will be shown later, this non-determistic constraint makes the
scheduling design far more complicated than related works [7, 8, 5, 6].

To model the fact that the largest number of assigned concurrent workers is no
more than the global batch size (otherwise, some workers will be idle), we
have:

$\displaystyle\sum_{h\in\mathcal{H}}w_{ih}[t]\leq x_{i}F_{i},\quad\forall
i\in\mathcal{I},a_{i}\leq t\leq T.$ (4)

3) Resource Constraint Modeling: We let $\mathcal{R}$ denote the set of
resources (e.g., CPU/GPU, memory, storage, etc.). Let $\alpha_{i}^{r}$ and
$\beta_{i}^{r}$ be the amount of type-$r$ resource required by a worker and a
parameter server for job $i$, respectively. Let $C_{h}^{r}$ be the capacity of
type-$r$ resource on machine $h$. To ensure the resources do not exceed
type-$r$’s limit, we have:

$\displaystyle\sum_{i\in\mathcal{I}}(\alpha_{i}^{r}w_{ih}[t]+\beta_{i}^{r}s_{ih}[t])\leq
C_{h}^{r},\forall t\in\mathcal{T},r\in\mathcal{R},h\in\mathcal{H}.$ (5)

Note that for job $i$, its completion time $\tilde{t}_{i}$ corresponds to the
latest time-slot where there remain some active workers allocated for it.
Therefore, we have:

$\displaystyle\tilde{t}_{i}=\arg\max_{t\in\mathcal{T}}\bigg{\\{}\sum_{h\in\mathcal{H}}w_{ih}[t]>0\bigg{\\}},\quad\forall
i\in\mathcal{I}.$ (6)

To ensure that no workers and parameter servers are allocated before job $i$’s
arrival, we have:

$\displaystyle w_{ih}[t]=s_{ih}[t]=0,\quad\forall
i\in\mathcal{I},h\in\mathcal{H},t<a_{i}.$ (7)

4) Objective Function and Problem Statement: Let $u_{i}(\tilde{t}_{i}-a_{i})$
be the utility function for job $i$, which is non-increasing with respect to
the training time $\tilde{t}_{i}-a_{i}$. The utility functions could play the
role of various performance metrics based on job completion times (e.g.,
fairness). In this paper, our goal is to maximize the overall utility for all
jobs. Putting all constraints and the objective function together, the offline
(with knowledge of $a_{i}$, $\forall i$) distributed ML resource scheduling
problem (DMLRS) can be formulated as:

DMLRS:
$\displaystyle\underset{\mathbf{x},\mathbf{w},\mathbf{s}}{\text{Maximize
}}\sum_{i\in\mathcal{I}}x_{i}u_{i}(\tilde{t}_{i}-a_{i})$
$\displaystyle\text{subject to }\text{Constraints
(\ref{eqn_min_workers_minibatch}) -- (\ref{yz})}.$

Problem DMLRS is an integer nonlinear program, which is NP-hard in general
[31]. Also, Problem DMLRS involves two non-deterministic constraints in (3)
and (6), which are not amenable for conventional optimization techniques.
Moreover, the arrivals $\\{a_{i},\forall i\\}$ are often unknown in practice,
which necessitates online optimization. Overcoming these challenges
constitutes the rest of this paper. To conclude this section, we summarize the
key notation used in this paper in Table I for easy reference.

TABLE I: Notation. $\mathcal{I/T}$ | The set of jobs$/$System timespan
---|---
$\tilde{t}_{i}/a_{i}$ | Completion time of job $i$ $/$Arrival time of job $i$
$u_{i}(\cdot)/K_{i}$ | Job $i$’$s$ utility function$/$Number of samples in $i$
$\mathcal{R/H}$ | The set of resource types$/$The set of machines
$E_{i}/F_{i}$ | # of training epochs/Global batch size for job $i$
$x_{i}$ | Admission decision variable to accept job $i$ or not
$C_{h}^{r}$ | Capacity of type-$r$ resource on server $h$
$\alpha_{i}^{r}$ | Type-$r$ resource required by a worker in job $i$
$\beta_{i}^{r}$ |  | Type-$r$ resource required by a PS in job $i$
---
$w_{ih}[t]$ | Number of workers of job $i$ on server $h$ in $t$
$s_{ih}[t]$ | Number of PSs of job $i$ on server $h$ in $t$
$b_{i}(h,p)$ |  | Bandwidth consumed by a worker of job $i$, where
---
$b_{i}(h,p)=b_{i}^{(e)}$, if $h\neq p$ or $b_{i}^{(i)}$, otherwise.
$\tau_{i}$ | Time to train a sample for job $i$
$g_{i}$ |  | Size of gradients and parameters for job $i$
---
$\mathcal{W}_{i}[t]$ |  | Set of physical machines containing workers
---
for job $i$ in $t$
$\mathcal{P}_{i}[t]$ |  | Machines containing parameter servers
---
for job $i$ in $t$
$x_{\pi_{i}}$ |  | Binary decision variable to select schedule $\pi$
---
for job $i$ or not
$\tilde{t}_{\pi_{i}}$ | The completion time slot of job $i$ with schedule $\pi$
$w_{ht}^{\pi_{i}}$ | # of workers on server $h$ in $t$ for job $i$ in schedule $\pi$
$s_{ht}^{\pi_{i}}$ | # of PSs on server $h$ for schedule $\pi$ in $t$
$\Pi_{i}$ | Set of all feasible schedules for job $i$
$\gamma_{i}$ | The ratio of worker number to PS number for job $i$
$\rho_{h}^{r}[t]$ | Allocated type-$r$ resource on machine $h$ in time $t$
$Q_{h}^{r}(\cdot)$ | Price function for type-$r$ resource on machine $h$

##  4 Online Scheduling Algorithm Design

In this section, we structure the key components of our online scheduling
algorithm design for solving Problem DMLRS into three steps from Sections 4.1
to 4.3. We state our main theoretical performance results in Section 4.4.

###  4.1 Reformulation for Non-Deterministic Constraint (6)

The first challenge in solving Problem DMLRS stems from the non-deterministic
“argmax” structure in constraint (6). To address this challenge, we let
$\Pi_{i}$ be the set of all feasible schedules for job $i\in\mathcal{I}$ that
satisfy constraints (3), (4). Each schedule $\pi_{i}\in\Pi_{i}$ is defined by
the numbers of workers $w_{ht}^{\pi_{i}}$ and parameter servers
$s_{ht}^{\pi_{i}}$ allocated for job $i$ on machine $h$ in each time-slot $t$,
i.e., $\pi_{i}\\!\triangleq\\!\\{w_{ht}^{\pi_{i}},s_{ht}^{\pi_{i}},\forall
t\\!\in\\!\mathcal{T},h\\!\in\\!\mathcal{H}\\}$. Note that $w_{ht}^{\pi_{i}}$
and $s_{ht}^{\pi_{i}}$ are constants, not to be confused with decision
variables $w_{ih}[t]$ and $s_{ih}[t]$. We define a binary variable
$x_{\pi_{i}}\\!\in\\!\\{0,1\\}$ that is equal to 1 if job $i$ is admitted and
scheduled under schedule $\pi_{i}$, or 0, otherwise. Clearly, due to the
combinatorial nature, $|\Pi_{i}|$ is exponential. We let $\tilde{t}_{\pi_{i}}$
denote job $i$’s completion time under schedule $\pi_{i}$. Then, one can
equivalently reformulate Problem DMLRS as:

R-DMLRS: $\displaystyle\underset{\mathbf{x}}{\text{Maximize
}}\sum_{i\in\mathcal{I}}\sum_{\pi_{i}\in\Pi_{i}}x_{\pi_{i}}u_{i}(\tilde{t}_{\pi_{i}}-a_{i})$
$\displaystyle\text{subject to
}\sum_{i\in\mathcal{I}}\sum_{\pi_{i}\in\Gamma(t,h)}(\alpha_{i}^{r}w_{ht}^{\pi_{i}}+\beta_{i}^{r}s_{ht}^{\pi_{i}})x_{\pi_{i}}\leq
C_{h}^{r},$ $\displaystyle\hskip 72.26999pt\forall
t\in\mathcal{T},r\in\mathcal{R},h\in\mathcal{H},$ (8) $\displaystyle\hskip
43.36243pt\sum_{\pi_{i}\in\Pi_{i}}x_{\pi_{i}}\leq 1,\quad\forall
i\in\mathcal{I},$ (9) $\displaystyle\hskip
43.36243ptx_{\pi_{i}}\in\\{0,1\\},\quad\forall
i\in\mathcal{I},\pi_{i}\in\Pi_{i},$

where we use $\Gamma(t,h)$ to represent the set of feasible schedules that use
machine $h$ to deploy workers or parameter servers in time-slot $t$.
Constraint (4.1) guarantees that, in any time-slot $t$ and on any machine $h$,
the total amount of consumed type-$r$ resources will not exceed the capacity
limit $C_{h}^{r}$. Constraint (9) ensures that, for each job $i$, at most one
feasible schedule from $\Pi_{i}$ will be selected. Note that Problem R-DMLRS
is an integer linear program (ILP) and a feasible solution to Problem R-DMLRS
has a corresponding feasible solution to the original Problem DMLRS, and vice
versa. Notice that the non-deterministic constraint (6) no longer exists in
Problem R-DMLRS. Further, if relaxing binary $x_{\pi_{i}}$-variables to real-
valued, Problem R-DMLRS is a linear program (LP). However, it remains
difficult to solve Problem R-DMLRS since it has an exponential number of
$x_{\pi_{i}}$-variables due to the combinatorial nature of feasible schedules.
We will address this challenge in the next subsection.

###  4.2 An Online Primal-Dual Framework for R-DMLRS

In what follows, we adopt a primal-dual online algorithmic framework to reduce
the number of binary variables, which is an effective approach to address this
kind of challenge in the literature (see, e.g., [32, 6]). Note that, in the
dual of Problem R-DMLRS, the number of dual variables is polynomial.
Meanwhile, although there are an exponential number of constraints in the dual
problem, one only needs to be concerned with the set of active (binding)
constraints, which are easier to deal with. To see this, we associate dual
variables $p_{h}^{r}[t]\geq 0$, $\forall t\in\mathcal{T}$, $h\in\mathcal{H}$,
$r\in\mathcal{R}$ and $\lambda_{i}>0$, $i\in\mathcal{I}$, with (4.1) and (9),
respectively. Then, following the standard procedure of dualization (relaxing
the integrality constraints), we obtain the following dual problem:

D-R-DMLRS:
$\displaystyle\underset{\boldsymbol{\lambda},\mathbf{p}}{\text{Minimize }}$
$\displaystyle\sum_{i\in\mathcal{I}}\lambda_{i}+\sum_{t\in\mathcal{T}}\sum_{h\in\mathcal{H}}\sum_{r\in\mathcal{R}}p_{h}^{r}[t]C_{h}^{r}$
(10) subject to $\displaystyle\lambda_{i}\geq
u_{i}(\tilde{t}_{\pi_{i}}-a_{i})-\sum_{t\in\mathcal{T}(\pi_{i})}\sum_{h\in\mathcal{H}(\pi_{i}[t])}\sum_{r\in\mathcal{R}}(\alpha_{i}^{r}w_{ht}^{\pi_{i}}$
$\displaystyle+\beta_{i}^{r}s_{ht}^{\pi_{i}})p_{h}^{r}[t],\quad\forall
i\in\mathcal{I},\pi_{i}\in\Pi_{i},$ (11) $\displaystyle p_{h}^{r}[t]\geq
0,\quad\forall t\in\mathcal{T},h\in\mathcal{H},r\in\mathcal{R},$
$\displaystyle\lambda_{i}\geq 0,\quad\forall i\in\mathcal{I},$

where $\mathcal{T}(\pi_{i})$ denotes the time-slots utilized by schedule
$\pi_{i}$ and $\mathcal{H}(\pi_{i}[t])$ denotes the set of machines containing
workers and/or parameter servers under $\pi_{i}$ in time-slot $t$. Here,
$p_{h}^{r}[t]$ can be viewed as the price for type-$r$ resource in time $t$.
Then, the right-hand-side (RHS) of (11) can be interpreted as job utility
minus overall resource cost of job $i$ using schedule $\pi_{i}$. Thus
$\lambda_{i}\geq 0$ can be viewed as the payoff of admitting job $i$ with
$\pi_{i}$. Next, we examine the properties of Problem D-R-DMLRS. To minimize
(4.2), we tend to reduce $\lambda_{i}$ and $p_{h}^{r}[t]$ until they hit zero.
However, as $\lambda_{i}$ and $p_{h}^{r}[t]$ decrease, the left-hand-side
(LHS) and RHS of (11) decreases and increases, respectively (the term
$u_{i}(\tilde{t}_{\pi_{i}}-a_{i})$ in the RHS of (11) is a constant given
$\pi_{i}$). Therefore, $\lambda_{i}$ will drop to a value $\lambda_{i}^{*}$,
which is equal to maximum of the RHS of (11) achieved by some schedule
$\pi_{i}^{*}$ and dual price $p_{h}^{r*}[t]$, i.e.,

$\displaystyle\lambda_{i}^{*}\\!=\\!u_{i}(\tilde{t}_{\pi_{i}^{*}}\\!-\\!a_{i})\\!-\\!\\!\sum_{t\in\mathcal{T}(\pi_{i}^{*})}\\!\sum_{h\in\mathcal{H}(\pi_{i}^{*}[t])}\\!\sum_{r\in\mathcal{R}}(\alpha_{i}^{r}w_{ht}^{\pi_{i}^{*}}\\!+\\!\beta_{i}^{r}s_{ht}^{\pi_{i}^{*}})p_{h}^{r*}[t].$

This optimality structural insight implies that Problem D-R-DMLRS is
equivalent to finding an optimal schedule $\pi_{i}^{*}$ and dual price
$p_{h}^{r*}[t]$ to maximize the RHS of (11). The above insights motivate the
following primal-dual-based algorithm as shown in Algorithm 1.

Algorithm 1: Primal-Dual Online Resource Scheduling (PD-ORS).  Initialization:

  1. 1.

Let $w_{ih}[t]=0$, $s_{ih}[t]=0$, $\forall i,t,h$. Let $\rho_{h}^{r}[t]=0$,
$\forall h,r,t$. Choose some appropriate initial values for $p_{h}^{r}[0]$.

Main Loop:

  1. 2.

Upon the arrival of job $i$, determine a schedule $\pi_{i}^{*}$ to maximize
the RHS of (11) and its corresponding payoff $\lambda_{i}$ using Algorithm 2
(to be specified).

  2. 3.

If $\lambda_{i}>0$, set $x_{i}=1$. Set $w_{ih}[t]$ and $s_{ih}[t]$ according
to schedule $\pi_{i}^{*}$, $\forall t\in\mathcal{T}(\pi_{i}^{*})$,
$h\in\mathcal{H}(\pi_{i}^{*}[t])$.

     * •

Update
$\rho_{h}^{r}[t]\leftarrow\rho_{h}^{r}[t]+\alpha_{i}^{r}w_{ih}[t]+\beta_{i}^{r}s_{ih}[t]$,
$\forall t\in\mathcal{T}(\pi_{i}^{*})$, $h\in\mathcal{H}(\pi_{i}^{*}[t])$,
$r\in\mathcal{R}$.

     * •

Update $p_{h}^{r}[t]=Q_{h}^{r}(\rho_{h}^{r}[t])$, $\forall
t\in\mathcal{T}(\pi_{i}^{*})$, $h\in\mathcal{H}(\pi_{i}^{*}[t])$,
$r\in\mathcal{R}$. Schedule job $i$ based on $\pi_{i}^{*}$ and go to Step 2.

  3. 4.

If $\lambda_{i}\leq 0$, set $x_{i}=0$ and reject job $i$ and go to Step 2.

The intuition of Algorithm 1 is as follows: By the complementary slackness
condition of the Karush-Kuhn-Tucker (KKT) conditions [27], the primal
constraint (9) must be tight when dual variable $\lambda_{i}>0$, which implies
that $x_{i}=1$ (Step 3) in Problem DMLRS. Otherwise, if $\lambda_{i}=0$, then
the RHS of (11) is non-positive, meaning the utility is low compared to the
cost of resource consumption under schedule $\pi_{i}^{*}$. Therefore, we
should reject job $i$ ($x_{i}=0$ in Step 4). However, in order for the PD-ORS
algorithm to work, two challenging components need to be specified: the
schedule $\pi_{i}^{*}$ and how to update the cost function
$Q_{h}^{r}(\cdot)$555 Here, we note that the “cost function”
$Q_{h}^{r}(\cdot)$ is interpreted from servers’ perspective rather than jobs’
perspective. Specifically, higher cost means servers allocated more resources
to jobs, which implies higher utility for the jobs since jobs receive more
resources from servers. In what follows, we will first focus on designing
$Q_{h}^{r}(\cdot)$ and defer the challenging problem of finding $\pi_{i}^{*}$
to Section 4.3. For the design of $Q_{h}^{r}(\cdot)$, consider the following
choice of $Q_{h}^{r}(\cdot)$:

$\displaystyle
Q_{h}^{r}(\rho_{h}^{r}[t])=L(U^{r}/L)^{\frac{\rho_{h}^{r}[t]}{C_{h}^{r}}},$
(12)

where constants $U^{r}$, $\forall r$, and $L$ are defined as follows:

$\displaystyle\\!\\!\\!\\!\\!\\!U^{r}\\!\triangleq\\!\max_{i\in\mathcal{I}}\frac{u_{i}(\lceil\frac{E_{i}K_{i}}{F_{i}}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(i)}F_{i}))\rceil-
a_{i})}{\alpha_{i}^{r}+\beta_{i}^{r}},\forall
r\in\mathcal{R},\\!\\!\\!\\!\\!\\!$ (13)
$\displaystyle\\!\\!\\!\\!\\!\\!L\triangleq\min_{i\in\mathcal{I}}\frac{1/(2\mu)u_{i}(T-a_{i})}{\sum_{r\in\mathcal{R}}\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i})\rceil(\alpha_{i}^{r}+\beta_{i}^{r})}.\\!\\!\\!\\!\\!\\!$
(14)

The scaling factor $\mu$ in the definition of $L$ satisfies as follows:

$\displaystyle\frac{1}{\mu}\leq\frac{\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i}))\rceil\sum_{r\in\mathcal{R}}(\alpha_{i}^{r}+\beta_{i}^{r})}{T\sum_{h\in\mathcal{H}}\sum_{r\in\mathcal{R}}C_{h}^{r}},\forall
i\in\mathcal{I}.$

Here, $U^{r}$ is the maximum unit-resource job utility to deploy workers and
parameter servers with type-$r$ resource . Here,
$u_{i}(\lceil\frac{E_{i}K_{i}}{F_{i}}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(i)}F_{i}))\rceil-
a_{i})$ is the largest utility job $i$ can achieve by using the maximum number
of co-located workers and parameter servers (hence communicating rate is
$b_{i}^{(i)}$) at all times during all $E_{i}$ epochs, so that
$\lceil\frac{E_{i}K_{i}}{F_{i}}(\tau_{i}\\!+\\!2g_{i}\gamma_{i}/(b_{i}^{(i)}F_{i}))\rceil\\!-\\!a_{i}$
is the earliest possible job completion time. Similarly, $L$ represents the
minimum unit-time unit-resource job utility among all jobs, with
$u_{i}(T-a_{i})$ being the smallest utility for job $i$, and workers and
parameter servers communicate at small external rate $b_{i}^{(e)}$. We use
$\rho_{h}^{r}[t]$ to denote the allocated amount of type-$r$ resource to
machine $h$ for (future) time slot $t$. The intuition behind the
$Q_{h}^{r}(\cdot)$ function is as follows: i) At $t\\!=\\!0$,
$\rho_{h}^{r}[0]\\!=\\!0,\forall h\in\mathcal{H},r\in\mathcal{R}$. Hence, the
price $p_{h}^{r}[0]=L$ is the lowest, $\forall h,r$, and any job can be
admitted; ii) As allocated resources increases, the price increases
exponentially fast to reject early coming jobs with low utility and to reserve
resources for later arrived jobs with higher utility; iii) When some type-$r$
resource is exhausted, i.e., $\rho_{h}^{r}[t]=C_{h}^{r},\exists
r\in\mathcal{R}$, $Q_{h}^{r}[C_{h}^{r}]=U^{r}$ and no job that requires
type-$r$ resources will be admitted since the $U^{r}$ is the highest price. As
will be shown later, this price function leads to a logarithmically scaling
competitive ratio in online algorithm design. Note that computing the price
function in Algorithm 1 requires the information of constants $U^{r}$, $L$,
which can usually be estimated empirically based historical data.

Here, we point out a few interesting insights on the design choices of the
cost function in Eq. (12). Note that $U^{r}$ and $L$ are defined in Eqns (13)
and (14), respectively. Here, we intentionally choose $U^{r}$ to be dependent
on $r$ and $L$ to be independent on $L$ due to the following reasons:

First, the rationality of choosing an upper bound $U^{r}$ that varies with
different resource types is to ensure that when some type-$r$ resource is
exhausted, no more jobs that require type-$r$ resource should be allocated. In
other words, when the allocated amount of type-$r$ resource reaches the
capacity of physical machine $h$, i.e., $\rho_{h}^{r}[t]=C_{n}^{r},\exists
r\in\mathcal{R}$, the price $p_{h}^{r}=Q_{h}^{r}(C_{h}^{r})$ should reach the
upper bound $U^{r}$, indicating that any job that requires type-$r$ resource
will not be allocated to $h$. However, jobs that do not require type-$r$
resource should still be able to be scheduled on machine $h$. For example,
jobs that do not require GPU can still be placed on a machine with no
available GPUs.

Second, the reason that we choose the lower bound $L$ to be independent of any
resource type $r$ is to yield a larger ratio of $\frac{U^{r}}{L}$. The larger
the ratio of $\frac{U^{r}}{L}$ is, the greater the price will be. Intuitively,
the ratio $\frac{U^{r}}{L}$ can be interpreted as the scheduling
“uncertainty,” which increases as the ratio gets larger, implying the price
function reacts to the accumulative resource consumption more aggressively.
Thus, choosing $L$ to be independent of resource type $r$ allows the price
function $Q_{h}^{r}(\cdot)$ to react more aggressively to the accumulative
allocated resource amount. We note that one can also choose the lower bound to
be dependent on resource type $r$ by replacing $L$ with $L^{r}$. By doing so,
the log-scaling theoretical competitive ratio in Theorem 5 still holds and the
proof in Appendix C only needs to be slightly updated with the new notation
$L^{r}$. However, the empirical performance of using $L^{r}$ as lower bound is
worse since the price function reacts less aggressively to the accumulative
allocated resources.

(a) $|\mathcal{P}_{i}[t]|\neq 1$.

(b) $|\mathcal{P}_{i}[t]|=1,|\mathcal{W}_{i}[t]|\neq 1$.

(c)
$|\mathcal{P}_{i}[t]|\\!\\!=\\!\\!|\mathcal{W}_{i}[t]|\\!\\!=\\!\\!1,\mathcal{P}_{i}[t]\\!\\!\neq\\!\\!\mathcal{W}_{i}[t]$.

(d)
$|\mathcal{P}_{i}[t]|\\!\\!=\\!\\!|\mathcal{W}_{i}[t]|\\!\\!=\\!\\!1,\mathcal{P}_{i}[t]\\!\\!=\\!\\!\mathcal{W}_{i}[t]$.

Figure 4: Values of
$\frac{2g_{i}}{\min_{p\in\mathcal{P}_{i}[t],h\in\mathcal{W}_{i}[t]}b_{i}(h,p)}$
under various settings of $\mathcal{P}_{i}[t]$ and $\mathcal{W}_{i}[t]$. Here,
$\frac{2g_{i}}{\min_{p\in\mathcal{P}_{i}[t],h\in\mathcal{W}_{i}[t]}b_{i}(h,p)}=2\frac{g_{i}}{b_{i}^{(i)}}$
if and only if in (d).

###  4.3 Determining Schedule $\pi_{i}^{*}$ in Step 2 of Algorithm 1

Now, consider the problem of finding a schedule $\pi_{i}^{*}$ in Step 2 of
Algorithm 1 to maximize the RHS of (11). First, we note that any schedule for
job $i$ has a unique completion time $\tilde{t}_{i}$, including the maximizer
$\pi_{i}^{*}$ for (11). Hence, the problem of finding the maximum RHS of (11)
can be decomposed as:

$\displaystyle\\!\\!\\!\underset{{\tilde{t}_{i}}}{\text{Max}}\left\\{\begin{array}[]{l}\underset{{\mathbf{w},\mathbf{s}}}{\text{Max}}\,\,u_{i}(\tilde{t}_{i}\\!-\\!a_{i})\\!-\\!\sum\limits_{t\in\mathcal{T}}\sum\limits_{h\in\mathcal{H}}\sum\limits_{r\in\mathcal{R}}p_{h}^{r}[t]\\\
\hskip 79.49744pt\times(\alpha_{i}^{r}w_{ih}[t]+\beta_{i}^{r}s_{ih}[t])\\\
\text{s.t. }\
\alpha_{i}^{r}w_{ih}[t]+\beta_{i}^{r}s_{ih}[t]\leq\hat{C}_{h}^{r}[t],\\\
\hskip 79.49744pt\forall t\in\mathcal{T},r\in\mathcal{R},h\in\mathcal{H},\\\
\hskip 21.68121pt\text{Constraints
(\ref{eqn_min_workers_minibatch})(\ref{eqn_max_workers_datachunk})(\ref{yz})
for }x_{i}=1,\end{array}\right\\},\\!\\!\\!\\!\\!$ (20)

where $\hat{C}_{h}^{r}[t]\triangleq C_{h}^{r}-\rho_{h}^{r}[t]$. Note that in
the inner problem, $u_{i}(\tilde{t}_{i}\\!-\\!a_{i})$ is a constant for any
given $\tilde{t}_{i}$. Thus, the inner problem can be simplified as:

$\displaystyle\underset{\mathbf{w},\mathbf{s}}{\text{Minimize }}$
$\displaystyle\sum_{t\in[a_{i},\tilde{t}_{i}]}\sum_{h\in\mathcal{H}}\sum_{r\in\mathcal{R}}p_{h}^{r}[t](\alpha_{i}^{r}w_{ih}[t]+\beta_{i}^{r}s_{ih}[t])$
(21) subject to
$\displaystyle\sum_{t\in[a_{i},\tilde{t}_{i}]}\sum_{h\in\mathcal{H}}\frac{w_{ih}[t]}{\tau_{i}+\frac{\gamma_{i}}{F_{i}}\frac{2g_{i}}{\min_{p\in\mathcal{P}_{i}[t],h^{\prime}\in\mathcal{W}_{i}[t]}b_{i}(h^{\prime},p)}}\geq
V_{i},\\!\\!\\!$ (22)
$\displaystyle\alpha_{i}^{r}w_{ih}[t]\\!+\\!\beta_{i}^{r}s_{ih}[t]\\!\leq\\!\hat{C}_{h}^{r}[t],\forall
r,h,\forall t\\!\in\\![a_{i},\tilde{t}_{i}],\\!\\!\\!$ (23)
$\displaystyle\text{Constraint (\ref{eqn_max_workers_datachunk}) for all
}t\in[a_{i},\tilde{t}_{i}],$

where $V_{i}\triangleq E_{i}K_{i}$ represents the total training workload
(total count of samples trained, where a sample is counted $E_{i}$ times if
trained for $E_{i}$ times). Note that in Problem (21), the only coupling
constraint is (22). This observation inspires a dynamic programming approach
to solve Problem (21). Consider the problem below if training workload at time
$t$ is known (denoted as $V_{i}[t]$):

$\displaystyle\underset{w_{ih}[t],s_{ih}[t],\forall h}{\text{Minimize }}$
$\displaystyle\hskip
7.22743pt\sum_{h\in\mathcal{H}}\sum_{r\in\mathcal{R}}p_{h}^{r}[t](\alpha_{i}^{r}w_{ih}[t]+\beta_{i}^{r}s_{ih}[t])$
(24) subject to $\displaystyle\hskip
7.22743pt\sum_{h\in\mathcal{H}}\frac{w_{ih}[t]}{\tau_{i}+\frac{\gamma_{i}}{F_{i}}\frac{2g_{i}}{\min_{p\in\mathcal{P}_{i}[t],h^{\prime}\in\mathcal{W}_{i}[t]}b_{i}(h^{\prime},p)}}\geq
V_{i}[t],$ (25) $\displaystyle\hskip 7.22743pt\text{Constraints
(\ref{eqn_max_workers_datachunk})(\ref{eqn_rsrc_limit}) for the given }t.$

Let $\Theta(\tilde{t}_{i},V_{i})$ and $\theta(t,V_{i}[t])$ denote the optimal
values of Problems (21) and (24), respectively. Then, Problem (21) is
equivalent to the following dynamic program:

$\displaystyle\Theta(\tilde{t}_{i},V_{i})=\min_{v\in[0,V_{i}]}\left\\{\theta(\tilde{t}_{i},v)+\Theta(\tilde{t}_{i}-1,V_{i}-v)\right\\}.$
(26)

We find the optimal workload $v$ to be completed in time slot $\tilde{t}_{i}$
by enumerating it from 0 to $E_{i}K_{i}$, and the remaining workload
$E_{i}K_{i}-v$ will be carried out in time span $[a_{i},\tilde{t}_{i}-1]$. The
optimal workload would be the schedule with minimum costs, i.e., the objective
function of Problem (24) is minimum. Then we proceed to the next time slot
$\tilde{t}_{i}-1$ with the workload $E_{i}K_{i}-v$, which is the same as
finding the optimal schedule and cost as the last time slot $\tilde{t}_{i}$
except in a smaller scale. Then, by enumerating all
$\tilde{t}_{i}\in[a_{i},T]$ and solving the dynamic program
$\Theta(\tilde{t}_{i},V_{i})$ in (26) for every choice of $\tilde{t}_{i}$, we
can solve Problem (20) and determine the optimal schedule $\pi_{i}^{*}$. We
summarize this procedure in Algorithms 2 and 3:  Algorithm 2: Determine
$\pi_{i}^{*}$ in Step 2 of Algorithm 1.  Initialization:

  1. 1.

Let $\tilde{t}_{i}\\!=\\!a_{i}$. Let $\lambda_{i}\\!=\\!0$,
$\pi_{i}^{*}\\!=\\!\varnothing$, $w_{ih}[t]\\!=\\!s_{ih}[t]\\!=\\!0$, $\forall
i,t,h$.

Main Loop:

  1. 2.

Compute $\Theta(\tilde{t}_{i},V_{i})$ in (26) using Algorithm 3. Denote the
resulted schedule as $\pi_{i}$. Let
$\lambda^{\prime}_{i}=u_{i}(\tilde{t}_{i}-a_{i})-\Theta(\tilde{t}_{i},V_{i})$.
If $\lambda^{\prime}_{i}>\lambda_{i}$, let
$\lambda_{i}\leftarrow\lambda^{\prime}_{i}$ and
$\pi_{i}^{*}\leftarrow\pi_{i}$.

  2. 3.

Let $\tilde{t}_{i}\leftarrow\tilde{t}_{i}+1$. If $\tilde{t}_{i}>T$, stop;
otherwise, go to Step 2.

Algorithm 3: Solving $\Theta(\tilde{t}_{i},V_{i})$ by Dynamic Programming.
Initialization:

  1. 1.

Let cost-min = $\infty$, $\pi_{i}=\varnothing$, and $v=0$.

Main Loop:

  1. 2.

Compute $\theta(\tilde{t}_{i},v)$ using Algorithm 4 (to be specified). Denote
the resulted cost and schedule as cost-v and $\hat{\pi}_{i}$.

  2. 3.

Compute $\Theta(\tilde{t}_{i}-1,V_{i}-v)$ by calling Algorithm 3 itself.
Denote the resulted cost and schedule as cost-rest and $\tilde{\pi}_{i}$.

  3. 4.

If $\text{{\em cost-min}}>\text{{\em cost-v}}+\text{{\em cost-rest}}$ then
$\text{{\em cost-min}}=\text{{\em cost-v}}+\text{{\em cost-rest}}$ and let
$\pi_{i}\leftarrow\hat{\pi}_{i}\cup\tilde{\pi}_{i}$.

  4. 5.

Let $v\leftarrow v+1$. If $v>V_{i}$ stop; otherwise go to Step 2.

In Algorithm 3, however, how to compute $\theta(t,v)$ in Step 2 (i.e., Problem
(24)) is yet to be specified. A challenge in solving (24) is the non-
deterministic constraint in (25), where $b_{i}(h,p)$ can be either
$b_{i}^{(i)}$ or $b_{i}^{(e)}$. Therefore, we need to handle both cases. To
this end, we observe the following basic fact about
$\frac{2g_{i}}{\min_{p\in\mathcal{P}_{i}[t],h\in\mathcal{W}_{i}[t]}b_{i}(h,p)}$
(also illustrated in Fig. 4) as stated in Fact 1. We omit the proof of this
fact due to its simplicity, which is illustrated in Fig. 4.

######  Fact 1.

The function
$(2g_{i}/\min_{p\in\mathcal{P}_{i}[t],h\in\mathcal{W}_{i}[t]}b_{i}(h,p))=2g_{i}/b_{i}^{(i)}$
if and only if $|\mathcal{P}_{i}[t]|=|\mathcal{W}_{i}[t]|=1$ and
$\mathcal{P}_{i}[t]=\mathcal{W}_{i}[t]$; otherwise,
$(2g_{i}/\min_{p\in\mathcal{P}_{i}[t],h\in\mathcal{W}_{i}[t]}b_{i}(h,p))=2g_{i}/b_{i}^{(e)}$.

With Fact 1, we now consider the following two cases:

Case 1): $b_{i}^{(i)}$ (Internal Communication): In Case 1), Fact 1 implies
that Problem (24) reduces to a single-machine problem (i.e., discarding
$\sum_{h\in\mathcal{H}}\\{\cdot\\}$ in (24) and (25)). Note further that if we
temporarily ignore the workload-coupling constraint (25) and use the worker-PS
ratio in (2), Problem (24) can be decoupled across resources and simplified
as:

$\sum_{r\in\mathcal{R}}\begin{Bmatrix}[l]&\text{Min}&p_{h}^{r}[t]s_{ih}[t](\alpha_{i}^{r}\gamma_{i}+\beta_{i}^{r})\\\
&\text{s.t.}&s_{ih}[t](\alpha_{i}^{r}\gamma_{i}+\beta_{i}^{r})\leq\hat{C}_{h}^{r}[t],\\\
&&\text{Constraint }(\ref{eqn_max_workers_datachunk})\text{ for given
}r,h\end{Bmatrix},\\!\\!\\!\\!$ (27)

where each summand in (27) is an integer linear program (ILP) having a trivial
solution $w_{ih}[t]=s_{ih}[t]=0$, $\forall h\in\mathcal{H}$. However,
$w_{ih}[t]=0$, $\forall h\in\mathcal{H}$, clearly violates the workload
constraint (25). Thus, when (27) is optimal, there should be exactly one
machine $h^{\prime}\in\mathcal{H}$ with $w_{ih^{\prime}}[t]\geq 1$ and exactly
one machine $h^{\prime\prime}\in\mathcal{H}$ with
$s_{ih^{\prime\prime}}[t]\geq 1$. This observation shows that the optimal
solution of (24) tends to favor $|\mathcal{P}_{i}[t]|=|\mathcal{W}_{i}[t]|=1$
if the workload constraint (25) is not binding.

Notice that the workload constraint (25) in Case 1) becomes
$\gamma_{i}s_{ih}[t]\geq
V_{i}[t](\tau_{i}+\frac{2g_{i}\gamma_{i}}{b_{i}^{(i)}F_{i}})$. This implies
the following simple solution: We can first sort each physical machine $h$
according to
$\sum_{r\in\mathcal{R}}p_{h}^{r}[t](\alpha_{i}^{r}\gamma_{i}+\beta_{i}^{r})$
and calculate the minimum number of
$s_{ih}[t]=V_{i}[t](\tau_{i}+\frac{2g_{i}\gamma_{i}}{b_{i}^{(i)}F_{i}})/\gamma_{i}$
from the workload constraint. The last step is to check if the machine satisfy
the resource capacity constraint (23) and constraint (4). If so, we return the
schedule $(w_{ih}[t],s_{ih}[t])$ and the corresponding cost value.

Case 2): $b_{i}^{(e)}$ (External Communication): For those settings that do
not satisfy $|\mathcal{P}_{i}[t]|=|\mathcal{W}_{i}[t]|=1$ and
$\mathcal{P}_{i}[t]=\mathcal{W}_{i}[t]$, Fact 1 indicates that parameter
servers and workers are communicating at external rate $b_{i}^{(e)}$. In this
case, the workload constraint (25) simply becomes:
$\sum_{h\in\mathcal{H}}w_{ih}[t]\geq
V_{i}[t](\tau_{i}+\frac{2g_{i}\gamma_{i}}{b_{i}^{(e)}F_{i}})$. Then, we can
rewrite Problem (24) as:

$\displaystyle\underset{w_{ih}[t],s_{ih}[t],\forall h}{\text{Minimize }}$
$\displaystyle\hskip
7.22743pt\sum_{h\in\mathcal{H}}p_{h}^{w}[t]w_{ih}[t]+p_{h}^{s}[t]s_{ih}[t]$
(28) subject to $\displaystyle\hskip
7.22743pt\alpha_{i}^{r}w_{ih}[t]+\beta_{i}^{r}s_{ih}[t]\leq\hat{C}_{h}^{r}[t],\,\,\forall
h,r,$ (29) $\displaystyle\hskip
7.22743pt\sum\nolimits_{h\in\mathcal{H}}w_{ih}[t]\leq F_{i},$ (30)
$\displaystyle\hskip 7.22743pt\sum\nolimits_{h\in\mathcal{H}}w_{ih}[t]\geq
V_{i}[t]\bigg{(}\tau_{i}+\frac{2g_{i}\gamma_{i}}{b_{i}^{(e)}F_{i}}\bigg{)},$
(31)

where $p_{h}^{w}[t]\triangleq\sum_{r\in\mathcal{R}}p_{h}^{r}[t]\alpha_{i}^{r}$
and $p_{h}^{s}[t]\triangleq\sum_{r\in\mathcal{R}}p_{h}^{r}[t]\beta_{i}^{r}$
denote the aggregated prices of all resources of allocating workers and PSs on
machine $h$ in time $t$, respectively.

Unfortunately, Problem (28) is a highly challenging integer programming
problem with generalized packing and cover type constraints (i.e., integer
variables rather than 0-1 variables) in (29)–(31), respectively, which is
clearly NP-Hard. Also, it is well-known that there are no polynomial time
approximation schemes (PTAS) even for the basic set-cover and bin-packing
problems unless P = NP [31]. In what follows, we will pursue an instance-
dependent constant ratio approximation scheme to solve Problem (28) in this
paper. To this end, we propose a randomized rounding scheme: First, we solve
the linear programming relaxation of Problem (28). Let
$\\{\bar{w}_{ih}[t],\bar{s}_{ih}[t],\forall h,t\\}$ be the fractional optimal
solution. We let $\delta\in(0,1]$ be a parameter. Let $G_{\delta}$ be a
constant (the notation $G_{\delta}$ signifies that $G_{\delta}$ is dependent
on $\delta$) to be defined later, and let
$w^{\prime}_{ih}[t]\\!=\\!G_{\delta}\bar{w}_{ih}[t],s^{\prime}_{ih}[t]\\!=\\!G_{\delta}\bar{s}_{ih}[t]$,
$\forall h,t$. Then, we randomly round
$\\{w^{\prime}_{ih}[t],s^{\prime}_{ih}[t],\forall h,t\\}$ to obtain an integer
solution as follows:

$\displaystyle w_{ih}[t]$ $\displaystyle=\begin{cases}\lceil
w^{\prime}_{ih}[t]\rceil,&\text{with probability }w^{\prime}_{ih}[t]-\lfloor
w^{\prime}_{ih}[t]\rfloor,\\\ \lfloor w^{\prime}_{ih}[t]\rfloor,&\text{with
probability }\lceil
w^{\prime}_{ih}[t]\rceil-w^{\prime}_{ih}[t],\end{cases}\\!\\!\\!$ (32)
$\displaystyle s_{ih}[t]$ $\displaystyle=\begin{cases}\lceil
s^{\prime}_{ih}[t]\rceil,&\text{with probability }s^{\prime}_{ih}[t]-\lfloor
s^{\prime}_{ih}[t]\rfloor,\\\ \lfloor s^{\prime}_{ih}[t]\rfloor,&\text{with
probability }\lceil s^{\prime}_{ih}[t]\rceil-s^{\prime}_{ih}[t].\end{cases}$
(33)

We will later prove in Theorem 3 (when $0<G_{\delta}\leq 1)$ and Theorem 4
(when $G_{\delta}>1$) that the approximation ratio of this randomized rounding
scheme in (32)-(33) enjoys a ratio that is independent on the problem size.

Lastly, summarizing results in Cases 1) – 2) yields the following
approximation algorithm for solving Problem (24):

Algorithm 4: Solving $\theta(t,v)$ (i.e., Problem (24)).  Initialization:

  1. 1.

Let $w_{ih}[t]\\!=\\!s_{ih}[t]\\!=\\!0$, $\forall h$. Let $h\\!=\\!1$. Pick
some $\delta\in(0,1]$. Let $G_{\delta}$ be defined as in Eq. (3) or Eqn (35).
Let $D\\!=\\!\lceil v(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(i)}F_{i}))\rceil$.
Let $h^{*}=\varnothing$. Let cost-min$=\infty$. Choose some integer $S\geq 1$.
Let $iter\leftarrow 1$.

Handling Internal Communication:

  1. 2.

Sort machines in $\mathcal{H}$ according to
$\sum_{r\in\mathcal{R}}p_{h}^{r}[t](\alpha_{i}^{r}\gamma_{i}+\beta_{i}^{r})$
in non-decreasing order into $h_{1},h_{2},...,h_{H}$.

  2. 3.

Calculate the minimum number of
$s_{ih}[t]=V_{i}[t]\Big{(}\tau_{i}+\frac{2g_{i}\gamma_{i}}{b_{i}^{(i)}F_{i}}\Big{)}/\gamma_{i}$.

  3. 4.

If Constraint (4) is not satisfied, go to Step 7.

  4. 5.

If Constraint (29) is not satisfied, go to Step 7.

  5. 6.

Return cost-
min$\sum_{r\in\mathcal{R}}p_{h}^{r}[t]s_{ih}[t](\alpha_{i}^{r}\gamma_{i}+\beta_{i}^{r})$
and $h^{*}=h$.

  6. 7.

Let $h\leftarrow h+1$. If $h>H$, stop; otherwise, go to Step 2.

Handling External Communication:

  1. 8.

Solve the linear programming relaxation of Problem (28). Let
$\\{\bar{w}_{ih}[t],\bar{s}_{ih}[t],\forall h,t\\}$ be the fractional optimal
solution.

  2. 9.

Let
$w^{\prime}_{ih}[t]=G_{\delta}\bar{w}_{ih}[t],s^{\prime}_{ih}[t]=G_{\delta}\bar{s}_{ih}[t]$,
$\forall h,t$.

  3. 10.

Generate an integer solution $\\{w_{ih}[t],s_{ih}[t],\forall h,t\\}$ following
the randomized rounding scheme in (32)–(33).

  4. 11.

If $\\{w_{ih}[t],s_{ih}[t],\forall h,t\\}$ is infeasible or $iter<S$, then
$iter\leftarrow iter+1$, go to Step 10.

Final Step:

  1. 12.

Compare the solutions between internal and external cases. Pick the one with
the lowest cost among them and return the cost and the corresponding schedule
$\\{w_{ih}[t],s_{ih}[t],\forall h,t\\}$.

In the internal communication part of Algorithm 4, we first sort the machines
and then check each machine one by one (Step 2). We calculate the minimum
number of $s_{ih}[t]$ needed to satisfy the learning workload demand $D$ (Step
3). If Constraint (4) is satisfied (Step 4), we further check the resource
capacity constraint (23) (Step 5). If we detect a machine with all above
constraints satisfied, we return the cost and schedule accordingly (Step 6).
After exploring one machine, we move on to the next one as long as it is not
the last machine (Step 7). The external communication part is based on LP
relaxation (Step 8) and randomized rounding (Step 9-12). Note that the
randomized rounding will find at most $S$ integer feasible solutions (Step
12). Finally, we choose the lowest cost among the solutions from the internal
and external communication parts (Step 13).

###  4.4 Theoretical Performance Analysis

We now examine the competitive ratio of our PD-ORS algorithm. Note that the
key component in PD-ORS is our proposed randomized rounding scheme in
(32)–(33), which is in turn the foundation of Algorithm 1. Thus, we first
prove the following results regarding the randomized rounding algorithm.
Consider an integer program with generalized cover/packing constraints:
$\min\\{\mathbf{c}^{\top}\mathbf{x}:\mathbf{A}\mathbf{x}\geq\mathbf{a},\mathbf{B}\mathbf{x}\leq\mathbf{b},\mathbf{x}\in\mathbb{Z}_{+}^{n}\\}$,
where $\mathbf{A}\in\mathbb{R}_{+}^{m\times n}$,
$\mathbf{B}\in\mathbb{R}_{+}^{r\times n}$, $\mathbf{a}\in\mathbb{R}_{+}^{m}$,
$\mathbf{b}\in\mathbb{R}_{+}^{r}$, and $\mathbf{c}\in\mathbb{R}_{+}^{n}$. Let
$\bar{\mathbf{x}}$ be a fractional optimal solution. Consider the randomized
rounding scheme: Let $\mathbf{x}^{\prime}=G_{\delta}\bar{\mathbf{x}}$ for some
$G_{\delta}$ (to be specified). Randomly round $\mathbf{x}^{\prime}$ to
$\hat{\mathbf{x}}\in\mathbb{Z}_{+}^{n}$ as: $\hat{x}_{j}=\lceil
x^{\prime}_{j}\rceil$ w.p. $x^{\prime}_{j}-\lfloor x^{\prime}_{j}\rfloor$ and
$\hat{x}_{j}=\lfloor x^{\prime}_{j}\rfloor$ o.w. Note that in the rounding
process, if $G_{\delta}>1$ ($G_{\delta}\in(0,1]$), the packing (cover)
constraint is prone to be violated and the cover (packing) constraint is
easier to be satisfied. Hence, depending on which constraint is more preferred
to be feasible, we consider two cases with respect to $G_{\delta}$.

1) $0<G_{\delta}\leq 1$: We have the following approximation result (see proof
in Appendix A):

######  Lemma 1 (Rounding).

Let
$W_{a}\\!\\!\triangleq\\!\\!\min\\{a_{i}/[\mathbf{A}]_{ij}\\!:[\mathbf{A}]_{ij}\\!\\!>\\!\\!0\\}$
and
$W_{b}\\!\\!\triangleq\\!\\!\min\\{b_{i}/[\mathbf{B}]_{ij}\\!:[\mathbf{B}]_{ij}\\!\\!>\\!\\!0\\}$.
Let $\delta\in(0,1]$ be a given constant and define $G_{\delta}$ as:

$\displaystyle G_{\delta}\triangleq
1+\frac{3\ln(3r/\delta)}{2W_{b}}-\sqrt{\bigg{(}\frac{3\ln(3r/\delta)}{2W_{b}}\bigg{)}^{2}+\frac{3\ln(3r/\delta)}{W_{b}}}.$

Then, with probability greater than $1-\delta$, $\hat{\mathbf{x}}$ achieves a
cost at most $\frac{3G_{\delta}}{\delta}$ times the cost of
$\bar{\mathbf{x}}$. Meanwhile, $\hat{\mathbf{x}}$ satisfies
$\mathrm{Pr}\Big{\\{}(\mathbf{A}\hat{\mathbf{x}})_{i}\leq
a_{i}(1-(\frac{2}{G_{\delta}W_{a}}\ln(\frac{3m}{\delta}))^{\frac{1}{2}})G_{\delta},\exists
i\Big{\\}}\leq\frac{\delta}{3m}$.

######  Remark 1 (Discussions on Feasibility).

An insightful remark of Lemma 1 is in order. Note that, theoretically, the
expression $1-(\frac{2}{G_{\delta}W_{a}}\ln(\frac{3m}{\delta}))^{\frac{1}{2}}$
in Lemma 1 could become negative. In this case, the last probabilistic
inequality in Lemma 1 trivially holds and is not meaningful in characterizing
the feasibility, even though the inequality remains valid. In order for the
probabilistic statement to be meaningful in characterizing the feasibility of
the integer linear program, we solve for $\delta$ by enforcing
$1-(\frac{2}{G_{\delta}W_{a}}\ln(\frac{3m}{\delta}))^{\frac{1}{2}}>0$, which
yields $\delta\geq 3m/e^{\frac{G_{\delta}W_{a}}{2}}$. On the other hand, we
prefer $\delta$ to be small since it bounds the feasibility violation
probability and approximation ratio achievability in Lemma 1. Hence, the
smaller the value of $3m/e^{\frac{G_{\delta}W_{a}}{2}}$, the less restrictive
the condition $\delta\geq 3m/e^{\frac{G_{\delta}W_{a}}{2}}$ is.

Figure 5: The feasibility study.

To gain a deeper understanding on how restrictive the condition $\delta\geq
3m/e^{\frac{G_{\delta}W_{a}}{2}}$ is, we conduct a case study and the results
are illustrated in Fig. 5. Here, we let RHS $\triangleq
3m/e^{\frac{G_{\delta}W_{a}}{2}}$ for convenience. We vary $\delta$ from
$0.02$ to $0.1$. Clearly, the left-hand-side (LHS) of the condition is the
$45^{\circ}$ straight line. In order for the condition to hold, the curve of
RHS should fall under this straight line. Based on typical computing cluster
parameters, we set $W_{b}$ to $15$, and set $r\triangleq RH+1$ to $401$
($R=4,H=100$). We can see from Fig. 5 that as $W_{a}$ increases, the curve of
RHS crosses the dashed line of LHS at a smaller $\delta$-value. This means
that, the larger the value of $W_{a}$, the easier for
$1-(\frac{2}{G_{\delta}W_{a}}\ln(\frac{3m}{\delta}))^{\frac{1}{2}}$ to become
positive. Hence, the probabilistic feasibility characterization in Lemma 1 is
useful for typical system parameters in practice.

2) $G_{\delta}>1$: We have the following approximation result (see proof in
Appendix B):

######  Lemma 2 (Rounding).

Let $W_{a}\triangleq\min\\{a_{i}/[\mathbf{A}]_{ij}:[\mathbf{A}]_{ij}>0\\}$ and
$W_{b}\triangleq\min\\{b_{i}/[\mathbf{B}]_{ij}:[\mathbf{B}]_{ij}>0\\}$. Let
$\delta\in(0,1]$ be a given constant and define $G_{\delta}$ as:

$\displaystyle G_{\delta}\triangleq
1+\frac{\ln(3m/\delta)}{W_{a}}+\sqrt{\bigg{(}\frac{\ln(3m/\delta)}{W_{a}}\bigg{)}^{2}+\frac{2\ln(3m/\delta)}{W_{a}}}.$

Then, with probability greater than $1-\delta$, $\hat{\mathbf{x}}$ achieves a
cost at most $\frac{3G_{\delta}}{\delta}$ times the cost of
$\bar{\mathbf{x}}$. Meanwhile, $\hat{\mathbf{x}}$ satisfies
$\mathrm{Pr}\Big{\\{}(\mathbf{B}\hat{\mathbf{x}})_{i}>b_{i}(1+(\frac{3}{G_{\delta}W_{b}}\ln(\frac{3r}{\delta}))^{\frac{1}{2}})G_{\delta},\exists
i\Big{\\}}\leq\frac{\delta}{3r}$.

Several important remarks for Lemmas 1 and 2 are summarized as follows:

  * i)

Note that Alg. 4 is a randomized algorithm. Therefore, its performance is also
characterized probabilistically, and $\delta$ is used for such probabilistic
characterization. Here, it means that with probability $1-\delta$, one
achieves an approximation ratio at most $\frac{3G_{\delta}}{\delta}$ and a
probabilistic feasibility guarantee as stated in Lemma 1 when
$0<G_{\delta}\leq 1$ and Lemma 2 when $G_{\delta}>1$. In other words, the
statements mean that the probability of getting a better approximation ratio
is smaller under randomized rounding (i.e., better approximation ratio
$\Rightarrow$ smaller $\frac{3G_{\delta}}{\delta}$ $\Rightarrow$ larger
$\delta$ $\Rightarrow$ smaller probability $1-\delta$). That is, the trade-off
between the approximation ratio value and its achieving probability is
quantified by $\delta$. A larger $\delta$ implies a smaller approximation
ratio, but the probability of obtaining a feasible solution of this ratio is
also smaller (i.e., more rounds of rounding needed). Interestingly, for
$\delta=1$, Lemmas 1 and 2 indicate that there is still non-zero probability
to achieve an approximation ratio not exceeding $3G_{\delta}$.

  * ii)

Note that if we pick $G_{\delta}\in(0,1]$, the approximation ratio
$\frac{3G_{\delta}}{\delta}$ decreases (the smaller the approximation ratio,
the better) as $\delta$ increases based on Eqn. (40) in Appendix A. However,
the growth rate of $G_{\delta}$ is slower compared to that of $\delta$ due to
the $\log$ operator. On the other hand, if we pick $G_{\delta}>1$,
$\frac{3G_{\delta}}{\delta}$ decreases as $\delta$ increases according to Eqn.
(43) in Appendix B. Therefore, the approximation ratio is ultimately
controlled by parameter $\delta$. Also, the theoretical approximation ratio
$\frac{3G_{\delta}}{\delta}$ is conservative. Our numerical studies show that
the approximation ratio performance in reality is much smaller than
$\frac{3G_{\delta}}{\delta}$.

  * iii)

The probabilistic guarantee of the cover constraint
($\mathbf{A}\mathbf{x}\geq\mathbf{a}$) when $0<G_{\delta}\leq 1$ and packing
constraint ($\mathbf{B}\mathbf{x}\leq\mathbf{b}$) when $G_{\delta}>1$, is
unavoidable and due to the fundamental hardness of satisfying both cover and
packing constraints, which are of conflicting nature: Any strategy trying to
better satisfy the packing constraints (multiplying a $G_{\delta}$-factor with
$G_{\delta}\in(0,1]$) may increase the probability of violating the cover
constraints, and the probability of violating the packing constraints may be
increased otherwise. However, the probabilistic bound here is for worst case
and may be pessimistic.

  * iv)

The results in Lemmas 1 and 2 are in fact applicable for general ILP with
mixed cover/packing constraints. Hence, the results and their insights in
Lemmas 1 and 2 could be of independent theoretical interest.

By specializing Lemma 1 and Lemma 2 with parameters in Problem (28), we have
the following approximation results for Algorithm 4. The first result
corresponds to the case where the feasibility of the resource constraint
(packing) is more favored, i.e., $0<G_{\delta}\leq 1$:

######  Theorem 3 (Approximation Performance of Alg. 4 When Resource
Constraint Feasibility is Favored).

Let
$W_{1}\\!\triangleq\\!V_{i}[t]\big{(}\tau_{i}+\frac{2g_{i}\gamma_{i}}{b_{i}^{(e)}F_{i}}\big{)}$,
$W_{2}\\!\triangleq\\!\min\\{F_{i},\hat{C}_{h}^{r}[t]/\alpha_{i}^{r},\hat{C}_{h}^{r}[t]/\beta_{i}^{r},\forall
r,h\\}$, and $\delta\in(0,1]$. Define $G_{\delta}$ as:

$\displaystyle
G_{\delta}\\!\triangleq\\!1\\!+\\!\frac{3\ln(3(RH\\!+\\!1)/\delta)}{2W_{2}}$
$\displaystyle\\!-\\!\sqrt{\bigg{(}\frac{3\ln(3(RH+1)/\delta)}{2W_{2}}\bigg{)}^{2}\\!+\\!\frac{3\ln(3(RH\\!+\\!1)/\delta)}{W_{2}}}.$
(34)

Then, with probability greater than $1\\!-\\!\delta$, Algorithm 4 obtains a
schedule $\\{w_{ih}[t],s_{ih}[t],\forall t,h\\}$ that has an approximation
ratio at most $\frac{3G_{\delta}}{\delta}$ with
$\mathrm{Pr}\\{\text{LHS(\ref{eqn_cover1})}\\!\leq\\!W_{1}(1-(\frac{2}{G_{\delta}W_{1}}\ln(\frac{3}{\delta}))^{\frac{1}{2}})G_{\delta},\exists
i\\}\leq\frac{\delta}{3}$.

The next result corresponds to the case where the feasibility of the workload
constraint (cover) is more favored, i.e., $G_{\delta}>1$:

######  Theorem 4 (Approximation Performance of Alg. 4 When Workload
Constraint Feasibility is Favored).

Let $W_{1}\triangleq
V_{i}[t]\big{(}\tau_{i}+\frac{2g_{i}\gamma_{i}}{b_{i}^{(e)}F_{i}}\big{)}$,
$W_{2}\triangleq\min\\{F_{i},\hat{C}_{h}^{r}[t]/\alpha_{i}^{r},\hat{C}_{h}^{r}[t]/\beta_{i}^{r},\forall
r,h\\}$, and $\delta\in(0,1]$. Define $G_{\delta}$ as:

$\displaystyle G_{\delta}\triangleq
1+\frac{\ln(3/\delta)}{W_{1}}+\sqrt{\bigg{(}\frac{\ln(3/\delta)}{W_{1}}\bigg{)}^{2}+\frac{2\ln(3/\delta)}{W_{1}}}.$
(35)

Then, with probability greater than $1-\delta$, Algorithm 4 obtains a schedule
$\\{w_{ih}[t],s_{ih}[t],\forall t,h\\}$ that has an approximation ratio at
most $\frac{3G_{\delta}}{\delta}$ with
$\mathrm{Pr}\\{\text{LHS(\ref{eqn_packing})}>\hat{C}_{h}^{r}[t]G_{\delta}(1+(\frac{3}{G_{\delta}W_{2}}\ln(\frac{3(HR+1)}{\delta}))^{\frac{1}{2}})\\}\leq\frac{\delta}{3(HR+1)}$.

Note that Eqn. (30) is guaranteed in practice since the number of samples is
typically far more than the number of workers. The competitive ratio of our
online algorithm is the worst-case upper bound of the ratio of the overall
utility of admitted jobs devided by the offline optimal solution of Problem
DMLRS to the total utility of admitted jobs achieved by Algorithm 1 in the
overall time horizon. Theorems 3 and 4 follow directly from Lemmas 1 and 2,
respectively, and we omit the proof here for brevity. Based on these results,
we can establish the overall competitive ratio for Algorithm 1 as follows.

######  Theorem 5 (Competitive Ratio of Alg 1 when
$0\\!<\\!G_{\delta}\\!\leq\\!1$).

Let $\delta$, $G_{\delta}$ and $W_{1}$ be as defined in Theorem 3. Let $U^{r}$
and $L$ be as defined in (13) and (14), respectively. Then, with probability
greater than $(1-(\delta/3)^{S})^{TK_{i}E_{i}}$, PD-ORS in Algorithm 1 returns
a feasible solution that is
$\frac{6G_{\delta}}{\delta}\max_{r\in\mathcal{R}}(1,\ln\frac{U^{r}}{L})$–competitive.

######  Theorem 6 (Competitive Ratio of Alg 1 when $G_{\delta}>1$).

Let $\delta$, $G_{\delta}$ and $W_{2}$ be as defined in Theorem 4. Let $U^{r}$
and $L$ be as defined in (13) and (14), respectively. Then, with probability
greater than $(1-(\delta/3(HR+1))^{S})^{TK_{i}E_{i}}$, PD-ORS in Algorithm 1
returns a feasible solution that is
$\frac{6G_{\delta}}{\delta}\max_{r\in\mathcal{R}}(1,\ln\frac{U^{r}}{L})$–competitive.

It is worth pointing out that in Theorems 5 and 6, the feasibility achieving
probability values, i.e., $(1-(\delta/3)^{S})^{TK_{i}E_{i}}$ and
$(1-(\delta/3(HR+1))^{S})^{TK_{i}E_{i}}$, can controlled by choosing
appropriate values of $\delta$ and $S$ (i.e., rounds of rounding) to offset
the impact of total number of DP iterations $TK_{i}E_{i}$. The smaller
$\delta$ and the larger $S$ are, the higher the feasibility achieving
probability. Theorems 5 and 6 can be proved by weak duality and the
approximation results in Theorems 3 and 4. We provide a proof in Appendix C.

######  Theorem 7 (Polynomial Running Time).

By combining Algorithms 1–4, the time complexity of PD-ORS is
$O(\sum_{i=1}^{|\mathcal{I}|}TK_{i}^{2}E_{i}^{2}(H^{3}+S))$, which is
polynomial.

###### Proof.

When solving $\theta(t,v)$ using Algorithm 4, it takes $O(H\log H)$ iterations
to sort machines in internal communication case under each time slot $t$ and
looping all machines to calculate the minimum number $s_{ih}[t]$ takes $O(H)$.
Thus, it takes $(H\log H)$ time for the internal communication part in
Algorithm 4. For the external communication part in Algorithm 4, solving the
LP relaxation of Problem (28) can be approximately bounded $O(H^{3})$ if we
use a polynomial time LP solver (e.g., Vaidya’s algorithm [33]). According to
Algorithm 4, the rounding time is proportional to $S$. Hence, the running time
for the external communications part is upper bounded by
$O(|\mathcal{H}|^{3}+S)$. Combining the discussions above, the running time
complexity of Algorithm 4 is $O(H\log H+H^{3}+S)=O(H^{3}+S)$. Moreover, the
number of states $(t,v)$ is $O(TK_{i}E_{i})$ in the dynamic programming (DP)
for each job $i$, and the time complexity of executing DP is $O(K_{i}E_{i})$.
Thus, the time complexity is $O(TK_{i}^{2}E_{i}^{2})$ in DP. In Algorithm 1,
the number of steps in the main loop is equal to the number of jobs.
Therefore, the overall running time complexity can be computed as
$O(\sum_{i=1}^{|\mathcal{I}|}TK_{i}^{2}E_{i}^{2}(H^{3}+S))$. ∎

##  5 Numerical Results

In this section, we conduct simulation studies to evaluate the efficacy of our
proposed PD-ORS algorithm. We test an ML system with jobs parameters generated
uniformly at random from the following intervals: $E_{i}\in[50,200]$,
$K_{i}\in[20000,500000]$, $g_{i}\in[30,575]$ MB,
$\tau_{i}\in[10^{-5},10^{-4}]$ time slots, $\gamma_{i}\in[1,10]$,
$F_{i}\in[1,200]$. We consider four types of resources: GPU, CPU, memory, and
storage. For fair comparisons, following similar settings in [34] [35] [36],
we set resource demand of each worker as follows: 0–4 GPUs, 1–10 vCPUs, 2–32
GB memory, and 5–10GB storage. We set resource demand of each parameter server
as follows: 1–10 vCPUs, 2–32GB memory and 5-10GB storage. The resource
capacity of each physical machine is set roughly 18 times of the resource
demands of a worker/PS following EC2 C5n instances [37]. We set the job
arrival pattern according to the Google Cluster data [38], but with normalized
job arrival rates in alternating time-slots as follows: the arrival rates are
$1/3$ and $2/3$ in odd and even time-slots, respectively. For fair
comparisons, we adopt the Sigmoid utility function [39, 6]:
$u_{i}(t-a_{i})=\frac{\theta_{1}}{1+e^{\theta_{2}(t-a_{i}-\theta_{3})}}$,
where $\theta_{1}\in[1,100]$ indicates the priority of job $i$, $\theta_{2}$
indicates how critical the time is for the job $i$, and $\theta_{3}\in[1,15]$
is the estimated target completion time. We set $\theta_{2}=0$ for time-
insensitive jobs (10% of jobs), $\theta_{2}\in[0.01,1]$ for time-sensitive
jobs (55% of jobs) and $\theta_{2}\in[4,6]$ for time-critical jobs (35% of
jobs).

We first compare our PD-ORS algorithm with three baseline job scheduling
policies: (1) FIFO in Hadoop and Spark [40], where the jobs are processed in
the order of their arrival times. In our setting, the fixed number of workers
(parameter servers) is between 1 to 30, (2) Dominant Resource Fairness
Scheduling (DRF) in Yarn [41] and Mesos [42], where the jobs are scheduled
based on their dominant resource share in the cloud to achieve its max-min
fairness. The number of workers and parameter servers are computed and
allocated dynamically, and (3) Dorm [36], where the numbers of workers
(parameter servers) are computed and placed by an MILP resource utilization
maximization problem with fairness and adjustment overhead constraints.
Workers and parameter servers are placed in a round-robin fashion on available
machines in Baselines (1) and (2). The comparison results are shown in Figs. 8
and 8. In Fig. 8, we set $T=20$ and $I=50$, while in Fig. 8, we set $T=20$ and
$H=100$. We can see that PD-ORS significantly outperforms other policies and
the gains in total utility becomes more pronounced as the numbers of jobs and
machines increase.

Figure 6: Total utility with increasing number of machines (synthetic data).

Figure 7: Total utility with increasing number of jobs (synthetic data).

Figure 8: Utility comparison between PD-ORS and OASiS with increasing number
of jobs.

Figure 9: Median of actual training time comparison.

Figure 10: Competitive ratio.

Figure 11: Impact of pre-rounding gain factor $G_{\delta}$ on competitive
ratio.

Figure 12: Total utility with increasing number of machines (Google cluster
data trace).

Figure 13: Total utility with increasing number of jobs (Google cluster data
trace).

Figure 14: Total utility with increasing number of machines [T=80, I=100,
(10%, 55% , 35%)].

Figure 15: Total utility with increasing number of machines [T=80, I=100,
(30%, 69% , 1%)].

Figure 16: Total utility with increasing number of jobs [T=80, H=30, (10%, 55%
, 35%)].

Figure 17: Total utility with increasing number of jobs [T=80, H=30, (30%, 69%
, 1%)].

Next, we compare our PD-ORS algorithm with the OASiS algorithm in [6], which
is also a dynamic scheduling scheme. As mentioned earlier, the key difference
in OASiS is that parameter servers and workers are located on two strictly
separated sets of machines (i.e., no co-located workers and PSs). Here, we let
$H=100$ and $T=20$. For OASiS, half of the machines host parameter servers and
the other half host workers. For fair comparisons, both algorithms adopt the
same Sigmoid utility function. The comparison results are shown in Fig. 8. We
can see that PD-ORS outperforms OASiS by allowing co-located parameter servers
and workers. We can see from Fig. 8 that the performance gap between PD-ORS
and OASiS widens as the number of jobs increases, which implies that PD-ORS is
more scalable. This is due to the advantage afforded by colocation of workers
and parameter servers, which allows each physical machine to be fully
utilized. On the other hand, the strict separation of workers and parameter
servers in OASiS may lead to the inability of placing workers on server-side
machines, should there be available resources or vice verse.

Next, we investigate the actual training time (completion time - arrival time)
under different methods, where $T=80$, $H=30$ and $I=100$. The median of the
actual training time is shown in Fig. 11. Here, we simply set its training
time to $T$ (i.e., 80) if the job cannot be finished within the scheduling
time span $\mathcal{T}$. As we can see from Fig. 11, PD-ORS outperforms other
scheduling policies, i.e., it has the smallest median time. Also, due to the
co-location advantage of PD-ORS, its median time is smaller compared to OASiS,
where workers and parameter servers are placed in strictly separated sets of
machines. We expect that the difference between PD-ORS and OASiS will become
more noticeable as the number of machines or the capacity of each machine
increases since it will allow more co-location placements.

Next, we demonstrate the competitive ratio of our algorithm PD-ORS, which is
the ratio between the total job utility of the offline optimal solution and
the total job utility achieved by PD-ORS. Recall that Problem DMLRS is a non-
convex problem with constraints (e.g., Eq. (1)) that are not amenable to be
directly solved by conventional optimization techniques. To obtain its offline
optimum, all possible combinations of $w_{ih}[t],s_{ih}[t],\forall i,h,t$ need
to be considered, which is time prohibitive. Thus, we limit the number of jobs
$I$ to 10 and time span $T$ to 10, and the result is shown in Fig. 11. As we
can see from the figure, the performance ratio is between 1.0 to 1.4,
indicating that our proposed algorithm PD-ORS has a good competitive ratio
performance.

Lastly, we examine the performance of the randomized rounding scheme in
Algorithm 4, which is the key of PD-ORS. We evaluate the rounding performance
in terms of the ratio between the optimal total utility and the total utility
obtained by our algorithm. The optimal utility is computed using the Gurobi
optimization solver. We let $H=100$, $I=50$, $T=20$. We vary the pre-rounding
gain factor $G_{\delta}$ (Theorems 3 and 4) from $0.2$ to $1.2$. The results
are shown in Fig. 11. The packing constraints are easier to satisfy with a
smaller $G_{\delta}$, while the cover constraints are prone to be violated as
$G_{\delta}$ gets smaller. In our experiments, if the total rounds of
randomized rounding before we find an integer feasible solution exceeds a
preset threshold (e.g, $5000$), we will discard the corresponding job. Theorem
3 suggests that there is a trade-off: if we set $G_{\delta}$ to be close to
one to pursue a better total utility result, the rounding time could be large
to obtain a feasible solution. As $G_{\delta}$ increases, the probability of
violating the packing constraints increases, meaning we need to have more
rounding attempts to obtain an integer feasible solution. However, according
to our numerical experiences, if the machine’s resource capacity is relatively
large compared to the jobs’ resource demands per worker/PS, the number of
rounding attempts is small and not sensitive to the variation of $G_{\delta}$.
On the other hand, as $G_{\delta}$ decreases, the probability of violating the
cover constraint increases. However, in practice, the model usually converges
with fewer number of iterations than the pre-defined training epochs since the
required number of epochs is usually overestimated [43]. In other words, the
violation of the cover constraint in one iteration may be acceptable.

As we can see from Fig. 11, the best approximation ratio value is achieved
when $G_{\delta}=1$. This is because if $G_{\delta}$ approaches $0$, it
implies that $\delta$ decreases at a larger rate (cf. Eq. (3)), resulting the
increment of the performance ratio. On the other hand, if $G_{\delta}$ goes to
infinity, it implies $\delta$ decreases (cf. Eq. (35)), resulting in a much
faster increment of the performance ratio. Also, we can see from Fig. 11 that
the performance ratios for all choices of $G_{\delta}$ are much better than
the theoretical bounds in Theorems 3 and 4, which shows that the approximation
ratio is much tighter than the worse-case bound suggested in Theorems 5 and 6.

Next, we show further experimental results with real-world data traces. We
first compare our PD-ORS algorithm with baseline scheduling algorithms, where
we follow job arrivals exactly based on timestamps recorded in the Google
Cluster data [38] by scaling down the original job trace (i.e., a “snippet” of
the trace). Here, we set $T=80$, $I=100$ and $H=30$. The comparison results
are shown in Figures 14 \- 14. Similarly, as we can see from the figures, our
algorithm PD-ORS outperforms other scheduling policies. In addition, due to
the co-location advantage of PD-ORS, it achieves more total job utility
compared to OASiS.

In the previous experiments, we have set the portions for time-insensitive
jobs, time-sensitive jobs and time-critical jobs to 10%, 55% and 35%,
respectively, which follows the default setting in [6] for fair comparison.
Theoretically, the larger the portion of time-sensitive and time-critical jobs
is, the better the performance is in terms of job utility compared to other
scheduling policies. Based on the Google trace analysis [44], there are four
categories of scheduling class of a job to indicate the latency sensitivity of
the job, where we label class 0 as time-insensitive, Classes 1 and 2 as time-
sensitive, and Class 3 as time-critical. In order to follow the practical
setting in the trace, we roughly set the ratio to 30%, 69% and 1%. We set
$T=80$. The number of machines increases from $10$ to $50$ with the number of
job fixed to $100$, and the number of jobs increases from $20$ to $100$ with
the number of machines fixed at $30$. We let Figs. 14 and 17 follow the
previous ratio setting (i.e., 10%, 55% and 35%), and Figs. 17 and 17 follow
the revised ratio setting (i.e., 30%, 69% and 1%). We examine the utility gain
compared to OASiS, where it is normalized. We present our experimental results
in Figs. 14 – 17. We can see from the figures that as the portion of critical
jobs decreases by 34%, the utility gain becomes smaller. That is, the
advantage of our proposed algorithm PD-ORS becomes less prominent.

We note that although in theory we can compare our PD-ORS algorithm (a special
case with utility function $u(x)=x$) with Optimus in [23] (which also takes
co-location into consideration), it is not straightforward to do so in
practice. Optimus requires an offline stage to estimate the
$\theta$-parameters of the speed function $f(p_{j},w_{j})$ (cf.
$\theta_{0}$–$\theta_{3}$ in [[23], Eq. (3)] for asynchronous training and
$\theta_{0}$–$\theta_{4}$ for synchronous training in [[23], Eq. (4)]).
Estimating these parameters requires specific hardware and software packages
that are not available in our current experimental environment. Due to the
above computing resource limitations and time constraints, we are unable to
conduct experiments to directly compare PD-ORS and Optimus in this work. Also,
our focus in this work is on scheduling algorithmic designs for deep learning
training with main contributions being on the theoretical aspects (proving
rigorous scheduling performance guarantees). Implementing our PD-ORS algorithm
in a similar testbed environment and having a comparison with Optimus is very
interesting and will be our next step in future studies.

##  6 Conclusion

In this paper, we investigated online resource scheduling for distributed
machine learning jobs in a shared computing cluster. We considered the most
general setting that workers and parameter servers can be co-located on the
same set of physical machines. We showed that this problem can be formulated
as a challenging integer nonlinear programming problem with non-deterministic
constraints. We developed an efficient online scheduling algorithm with
competitive ratio guarantee. Our main contributions are three-fold: i) We
developed a new analytical model that jointly considers resource locality and
allocation; ii) Through careful examinations of worker-server configuration
relationships, we resolved the locality ambiguity in the model and reduce the
problem to a mixed cover/packing integer program that entails low-complexity
approximation algorithm design; iii) We proposed a meticulously designed
randomized rounding algorithm to solve the mixed cover/packing integer program
and rigorously established its approximation ratio guarantee. Collectively,
our results expand the theoretical frontier of online optimization algorithm
design for resource optimization in distributed machine learning systems.

## References

  * [1] M. Abadi, P. Barham _et al._ , “TensorFlow: A system for large-scale machine learning,” in _Proc. of USENIX OSDI_ , 2016.
  * [2] T. Chen, M. Li, Y. Li, M. Lin, N. Wang, M. Wang, T. Xiao, B. Xu, C. Zhang, and Z. Zhang, “MXNet: A flexible and efficient machine learning library for heterogeneous distributed systems,” in _Proc. of NIPS Workshop on Machine Learning Systems (LearningSys)_ , 2016.
  * [3] A. Paszke, S. Gross, F. Massa, A. Lerer, J. Bradbury, G. Chanan, T. Killeen, Z. Lin, N. Gimelshein, L. Antiga, A. Desmaison _et al._ , “Pytorch: An imperative style, highperformance deep learning library,” in _In Advances in Neural Information Processing Systems_ , 2019, pp. 8024–8035.
  * [4] Y. Jia, E. Shelhamer, J. Donahue, S. Karayev, J. Long, R. Girshick, S. Guadarrama, and T. Darrell, “Caffe: Convolutional architecture for fast feature embedding,” in _Proc. of the 22nd ACM International Conference on Multimedia_ , 2014, pp. 675–678.
  * [5] B.-G. Chun, B. Cho, B. Jeon _et al._ , “Dolphin: Runtime optimization for distributed machine learning,” in _Proc. of ICML ML Systems Workshop_ , 2016\.
  * [6] Y. Bao, Y. Peng, C. Wu, and Z. Li, “Online job scheduling in distributed machine learning clusters,” in _Proc. of IEEE INFOCOM_ , 2018.
  * [7] M. Li, D. G. Andersen _et al._ , “Scaling distributed machine learning with the parameter server,” in _Proc. of USENIX OSDI_ , 2014.
  * [8] T. M. Chilimbi, Y. Suzue, J. Apacible, and K. Kalyanaraman, “Project Adam: Building an efficient and scalable deep learning training system,” in _Proc. of USENIX OSDI_ , 2014.
  * [9] J. Dean and S. Ghemawat, “Mapreduce: Simplified data processing on large clusters,” in _Commun. ACM_ , vol. 1, no. 51, 2008, pp. 107–113.
  * [10] M. Isard, M. Budiu, Y. Yu, A. Birrell, and D. Fetterly, “Dryad: Distributed data-parallel programs from sequential building blocks,” in _In EuroSys 2007_ , 2007, pp. 59–72.
  * [11] F. Yan, O. Ruwase, Y. He, and T. Chilimbi, “Performance modeling and scalability optimization of distributed deep learning systems,” in _Proc. of ACM KDD_ , 2015.
  * [12] P. Sun, Y. Wen, N. B. D. Ta, and S. Yan, “Towards distributed machine learning in shared clusters: A dynamically-partitioned approach,” in _Proc. of IEEE Smart Computing_ , 2017.
  * [13] Q. Zhang, R. Zhou, C. Wu, L. Jiao, and Z. Li, “Online scheduling of heterogeneous distributed machine learning jobs,” in _Proceedings of the Twenty-First International Symposium on Theory, Algorithmic Foundations, and Protocol Design for Mobile Networks and Mobile Computing_ , 2020, pp. 111–120.
  * [14] H. Mao, M. Alizadeh, I. Menache, and S. Kandula, “Resource management with deep reinforcement learning,” in _Proc. of ACM HotNets_ , 2016.
  * [15] W. Chen, Y. Xu, and X. Wu, “Deep reinforcement learning for multi-resource multi-cluster job scheduling,” in _Proc. of IEEE ICNP_ , 2017.
  * [16] H. Mao, M. Schwarzkopf, S. Venkatakrishnan, and M. Alizadeh, “Learning graph-based cluster scheduling algorithms,” in _Proc. of SysML_ , 2018.
  * [17] H. Mao, M. Schwarzkopf, S. B. Venkatakrishnan, Z. Meng, and M. Alizadeh, “Learning scheduling algorithms for data processing clusters,” in _Proc. of ACM SIGCOMM_ , 2019.
  * [18] A. Mirhoseini, A. Goldie, H. Pham, B. Steiner, Q. V. Le, and J. Dean, “A hierarchical model for device placement,” in _Proc. of ICLR_ , 2018.
  * [19] A. Mirhoseini, H. Pham, Q. V. Le, B. Steiner, R. Larsen, Y. Zhou, N. Kumar, M. Norouzi, S. Bengio, and J. Dean, “Device placement optimization with reinforcement learning,” in _Proc. of ICML_ , 2017.
  * [20] V. Kovalevskyi, “MXNet distributed training explained in depth.” [Online]. Available: https://blog.kovalevskyi.com/mxnet-distributed-training-explained-in-depth-part-1-b90c84bda725
  * [21] “Distributed training in MXNet.” [Online]. Available: https://mxnet.incubator.apache.org/faq/distributed_training.html
  * [22] P. Goyal, P. Dolla, R. Girshick, P. Noordhuis, L. Wesolowski, A. Kyrola, A. Tulloch, Y. Jia, and K. He, “Accurate, large minibatch SGD: Training imagenet in 1 hour,” _arXiv preprint arXiv:1706.02677_ , 2017.
  * [23] Y. Peng, Y. Bao, Y. Chen, C. Wu, and C. Guo, “Optimus: An efficient dynamic resource scheduler for deep learning clusters,” in _Proc. of ACM EuroSys_ , 2018.
  * [24] “Apache Hadoop.” [Online]. Available: http://hadoop.apache.org/.
  * [25] J. Dean, G. S. Corrado, R. Monga, K. Chen, M. Devin, Q. V. Le, M. Z. Mao, M. Ranzato, A. Senior, P. Tucker, K. Yang, and A. Y. Ng, “Large scale distributed deep networks,” in _Proceedings of the 25th International Conference on Neural Information Processing Systems - Volume 1_ , ser. NIPS’12. USA: Curran Associates Inc., 2012, pp. 1223–1231.
  * [26] T. Cheatham, A. Fahmy, D. Siefanescu, and L. Valiani, “Bulk synchronous parallel computing-a paradigm for transportable software,” in _Hawaii International Conference on Systems Sciences,_ , 2005.
  * [27] M. S. Bazaraa, H. D. Sherali, and C. M. Shetty, _Nonlinear Programming: Theory and Algorithms_ , 3rd ed. New York, NY: John Wiley & Sons Inc., 2006.
  * [28] “Multi-worker training with keras.” [Online]. Available: https://www.tensorflow.org/tutorials/distribute/multi_worker_with_keras
  * [29] “Run deep learning with paddlepaddle on kubernetes,” 2017. [Online]. Available: https://kubernetes.io/blog/2017/02/run-deep-learning-with-paddlepaddle-on-kubernetes/
  * [30] Z. Wang, K. Ji, Y. Zhou, Y. Liang, and V. Tarokh, “Spiderboost: A class of faster variance-reduced algorithms for nonconvex optimization,” in _arXiv preprint arXiv:1810.10690_ , 2018.
  * [31] D. S. Hochbaum, _Approximation Algorithms for NP-Hard Problems_ , D. S. Hochbaum, Ed. PWS Publishing Company, 1997\.
  * [32] N. Buchbinder and J. (Seffi) Naor, _The Design of Competitive Online Algorithms via a Primal-Dual Approach_. Now Publishers Inc., feb 2009, vol. 3.
  * [33] P. M. Vaidya, “An algorithm for linear programming which requires $o(((m+n)n^{2}+(m+n)^{1.5}n){L})$ algorithmic operations,” in _28th Annual IEEE Syposium on Foundations of Computer Science_ , 1987.
  * [34] M. Li, D. G. Andersen _et al._ , “Scaling distributed machine learning with the parameter server,” in _Proc. of USENIX OSDI_ , 2014.
  * [35] T. M. Chilimbi, Y. Suzue, J. Apacible, and K. Kalyanaraman, “Project adam: Building an efficient and scalable deep learning training system,” in _Proc. of USENIX OSDI_ , 2014.
  * [36] P. Sun, Y. Wen, N. B. D. Ta, and S. Yan, “Towards distributed machine learning in shared clusters: A dynamically-partitioned approach,” in _Proc. of IEEE Smart Computing_ , 2017.
  * [37] “Amazon ec2 instances.” [Online]. Available: https://aws.amazon.com/ec2/instance-types
  * [38] C. Reiss, A. Tumanov _et al._ , “Heterogeneity and dynamicity of clouds at scale: Google trace analysis,” in _Proc. of ACM SoCC_ , 2012.
  * [39] Z. Huang, B. Balasubramanian, M. Wang, T. Lan, M. Chiang, and D. H. Tsang, “Need for speed: Cora scheduler for optimizing completiontimes in the cloud,” in _Proc. of IEEE INFOCOM_ , 2015.
  * [40] M. Zaharia, M. Chowdhury, M. J. Franklin, and et al, “Spark: Cluster computing with working sets,” in _n Proc. of USENIX HotCloud_ , 2010.
  * [41] V. K. Vavilapalli, A. C. Murthy _et al._ , “Apache hadoop yarn: Yet another resource negotiator,” in _Proc. of ACM SoCC_ , 2013.
  * [42] B. Hindman, A. Konwinski _et al._ , “Mesos: A platform for fine-grained resource sharing in the data center,” in _Proc. of USENIX NSDI_ , 2011.
  * [43] M. Jeon, S. Venkataraman, A. Phanishayee, J. Qian, W. Xiao, and F. Yang, “Multi-tenant gpu clusters for deep learning workloads: Analysis and implications,” in _MSR-TR-2018- 13_ , 2018.
  * [44] P. Minet, Éric Renault, I. Khoufi, and S. Boumerdassi, “Analyzing traces from a google data center,” in _International Wireless Communications and Mobile Computing Conference (IWCMC)_ , 2018, pp. 1167–1172.
  * [45] D. Bertsekas, _Nonlinear Programming_ , 2nd ed. Athena Scientific, 1999.

|  Menglu Yu received her B.S. degree from the Department of Electrical and
Information Engineering at Hunan University, China in 2014. She is currently
pursuing her Ph.D. degree in the Department of Computer Science at Iowa State
University. Her primary research interests include optimization for
distributed machine learning systems and data centers, as well as network
optimization.
---|---
|  Jia Liu (S’03–M’10–SM’16) is an Assistant Professor in the Department of
Electrical and Computer Engineering at The Ohio State University, where he
joined in Aug. 2020. He received his Ph.D. degree from the Department of
Electrical and Computer Engineering at Virginia Tech in 2010. From Aug. 2017
to Aug. 2020, he was an Assistant Professor in the Department of Computer
Science at Iowa State University. His research areas include theoretical
machine learning, control and optimization for stochastic networks, and
optimization for data analytics infrastructure and cyber-physical systems. Dr.
Liu is a senior member of IEEE and a member of ACM. He has received numerous
awards at top venues, including IEEE INFOCOM’19 Best Paper Award, IEEE
INFOCOM’16 Best Paper Award, IEEE INFOCOM’13 Best Paper Runner-up Award, IEEE
INFOCOM’11 Best Paper Runner-up Award, and IEEE ICC’08 Best Paper Award. Dr.
Liu is a recipient of the NSF CAREER Award in 2020. He is a recipient of the
Google Faculty Research Award in 2020. He is also a winner of the LAS Award
for Early Achievement in Research from the College of Liberal Arts and
Sciences at Iowa State University in 2020, and the Bell Labs President Gold
Award in 2001. His research is supported by NSF, AFOSR, AFRL, and ONR.
---|---
|  Chuan Wu received her B.Engr. and M.Engr. degrees in 2000 and 2002 from the
Department of Computer Science and Technology, Tsinghua University, China, and
her Ph.D. degree in 2008 from the Department of Electrical and Computer
Engineering, University of Toronto, Canada. Since September 2008, Chuan Wu has
been with the Department of Computer Science at the University of Hong Kong,
where she is currently a Professor. Her current research is in the areas of
cloud computing, distributed machine learning systems and algorithms, and
intelligent elderly care technologies. She is a senior member of IEEE, a
member of ACM, and served as the Chair of the Interest Group on Multimedia
services and applications over Emerging Networks (MEN) of the IEEE Multimedia
Communication Technical Committee (MMTC) from 2012 to 2014. She is an
associate editor of IEEE Transactions on Cloud Computing, IEEE Transactions on
Multimedia, ACM Transactions on Modeling and Performance Evaluation of
Computing Systems, and IEEE Transactions on Circuits and Systems for Video
Technology. She was the co-recipient of the best paper awards of HotPOST 2012
and ACM e-Energy 2016.
---|---
|  Bo Ji (S’11-M’12-SM’18) received his B.E. and M.E. degrees in Information
Science and Electronic Engineering from Zhejiang University, Hangzhou, China,
in 2004 and 2006, respectively, and his Ph.D. degree in Electrical and
Computer Engineering from The Ohio State University, Columbus, OH, USA, in
2012. Dr. Ji is an Associate Professor in the Department of Computer Science
at Virginia Tech, Blacksburg, VA, USA. Prior to joining Virginia Tech, he was
an Associate Professor in the Department of Computer and Information Sciences
and a faculty member of the Center for Networked Computing at Temple
University, where he was an Assistant Professor from July 2014 to June 2020.
He was also a Senior Member of the Technical Staff with AT&T Labs, San Ramon,
CA, from January 2013 to June 2014. His research interests are in the
modeling, analysis, control, optimization, and learning of computer and
network systems, such as wired and wireless networks, large-scale IoT systems,
high performance computing systems and data centers, and cyber-physical
systems. He currently serves on the editorial boards of the IEEE/ACM
Transactions on Networking, IEEE Internet of Things Journal, and IEEE Open
Journal of the Communications Society. Dr. Ji is a senior member of the IEEE
and a member of the ACM. He is a National Science Foundation (NSF) CAREER
awardee (2017) and an NSF CISE Research Initiation Initiative (CRII) awardee
(2017). He is also a recipient of the IEEE INFOCOM 2019 Best Paper Award.
---|---
|  Elizabeth S. Bentley has a B.S. degree in Electrical Engineering from
Cornell University, a M.S. degree in Electrical Engineering from Lehigh
University, and a Ph.D. degree in Electrical Engineering from University at
Buffalo. She was a National Research Council Post-Doctoral Research Associate
at the Air Force Research Laboratory in Rome, NY. Currently, she is employed
by the Air Force Research Laboratory in Rome, NY, performing in-house research
and development in the Networking Technology branch. Her research interests
are in cross-layer optimization, wireless multiple-access communications,
wireless video transmission, modeling and simulation, and directional
antennas/directional networking.
---|---

##  Appendix A Proof of Lemma 1

###### Proof.

Consider the probabilities of the following “bad” events: 1)
$\mathbf{c}^{\top}\hat{\mathbf{x}}>\frac{3G_{\delta}}{\delta}\mathbf{c}^{\top}\bar{\mathbf{x}}$;
2) $\exists i$ such that $(\mathbf{A}\hat{\mathbf{x}})_{i}<a_{i}$; and 3)
$\exists i$ such that $(\mathbf{B}\hat{\mathbf{x}})_{i}>b_{i}$. Note that
events 2) and 3) can be equivalently rewritten as: 2’) $\exists i$ such that
$\mathbb{E}\\{(\mathbf{A}\hat{\mathbf{x}})_{i}\frac{W_{a}}{a_{i}}<W_{a}\\}$
and 3’) $\exists i$ such that
$\mathbb{E}\\{(\mathbf{B}\hat{\mathbf{x}})_{i}\frac{W_{b}}{b_{i}}>W_{b}\\}$.
Since
$\mathbb{E}\\{\hat{\mathbf{x}}\\}\\!=\\!\mathbf{x}^{\prime}\\!=\\!G_{\delta}\bar{\mathbf{x}}$,
by linearity of expectation, we have:

$\displaystyle\mathbb{E}\\{\mathbf{c}^{\top}\hat{\mathbf{x}}\\}=\mathbf{c}^{\top}\mathbb{E}\\{\hat{\mathbf{x}}\\}=\mathbf{c}^{\top}G_{\delta}\bar{\mathbf{x}}=G_{\delta}\mathbf{c}^{\top}\bar{\mathbf{x}},$
(36)
$\displaystyle\mathbb{E}\left\\{(\mathbf{A}\hat{\mathbf{x}})_{i}\frac{W_{a}}{a_{i}}\right\\}=G_{\delta}\mathbb{E}\left\\{(\mathbf{A}\bar{\mathbf{x}})_{i}\frac{W_{a}}{a_{i}}\right\\}\geq
G_{\delta}W_{a},$ (37)
$\displaystyle\mathbb{E}\left\\{(\mathbf{B}\hat{\mathbf{x}})_{i}\frac{W_{b}}{b_{i}}\right\\}=G_{\delta}\mathbb{E}\left\\{(\mathbf{B}\bar{\mathbf{x}})_{i}\frac{W_{b}}{b_{i}}\right\\}\leq
G_{\delta}W_{b}.$ (38)

Then, by the Markov inequality and (36), we can obtain the probability
$\mathrm{Pr}\left\\{\mathbf{c}^{\top}\hat{\mathbf{x}}>\frac{3G_{\delta}}{\delta}\mathbf{c}^{\top}\bar{\mathbf{x}}\right\\}\leq\frac{\delta}{3}$.

Next, we note that each $\hat{x}_{j}$ can be viewed as a sum of independent
random variables in $[0,1]$ as follows: The fixed part of $\lfloor
x^{\prime}_{j}\rfloor$ is a sum of $\lfloor x^{\prime}_{j}\rfloor$ random
variables with value 1 with probability 1.

Then, we have that
$(\mathbf{B}\hat{\mathbf{x}})_{i}\frac{W_{b}}{b_{i}}=(\sum_{j}[\mathbf{B}]_{ij}\hat{x}_{j})\frac{W_{b}}{b_{i}}$
is also a sum of independent random variables in $[0,1]$. Using the Chernoff
bound, we have

$\displaystyle\mathrm{Pr}\Big{\\{}(\mathbf{B}\hat{\mathbf{x}})_{i}\frac{W_{b}}{b_{i}}>(1+\epsilon)G_{\delta}W_{b}\Big{\\}}\leq\exp\Big{(}-\epsilon^{2}\frac{G_{\delta}W_{b}}{3}\Big{)}.$

Setting $(1+\epsilon)G_{\delta}\\!=\\!1$, i.e.,
$\epsilon\\!=\\!\frac{1}{G_{\delta}}\\!-\\!1$, we have:

$\displaystyle\mathrm{Pr}\left\\{(\mathbf{B}\hat{\mathbf{x}})_{i}\frac{W_{b}}{b_{i}}>W_{b}\right\\}\leq\exp\left(-\Big{(}\frac{1}{G_{\delta}}-1\Big{)}^{2}G_{\delta}\frac{W_{b}}{3}\right).\\!\\!\\!$
(39)

Forcing
$\exp\\!\left(-(\frac{1}{G_{\delta}}\\!-\\!1)^{2}G_{\delta}\frac{W_{b}}{3}\right)\\!=\\!\frac{\delta}{3r}$
and solving $G_{\delta}$, we have:

$\displaystyle G_{\delta}\triangleq
1+\frac{3\ln(3r/\delta)}{2W_{b}}\\!-\\!\sqrt{\bigg{(}\frac{3\ln(3r/\delta)}{2W_{b}}\bigg{)}^{2}+\frac{3\ln(3r/\delta)}{W_{b}}}.\\!\\!\\!$
(40)

Using (37), the Chernoff bound, and following similar arguments, we have:

$\displaystyle\mathrm{Pr}\Big{\\{}(\mathbf{A}\hat{\mathbf{x}})_{i}\frac{W_{a}}{a_{i}}\leq(1-\epsilon)G_{\delta}W_{a}\Big{\\}}\leq\exp\Big{(}-\epsilon^{2}\frac{G_{\delta}W_{a}}{2}\Big{)}.$

Forcing
$\exp\left(-\epsilon^{2}\frac{G_{\delta}W_{a}}{2}\right)=\frac{\delta}{3m}$
and solving for $\epsilon$, we have
$\epsilon=(\frac{2}{G_{\delta}W_{a}}\ln(\frac{3m}{\delta}))^{\frac{1}{2}}$. It
follows that

$\displaystyle\mathrm{Pr}\\{(\mathbf{A}\hat{\mathbf{x}})_{i}\frac{W_{a}}{a_{i}}\leq(1\\!-\\!(\frac{2}{G_{\delta}W_{a}}\ln(\frac{3m}{\delta}))^{\frac{1}{2}})G_{\delta}W_{a}\\}\leq\frac{\delta}{3m},$

which implies that:

$\displaystyle\\!\\!\\!\mathrm{Pr}\left\\{(\mathbf{A}\hat{\mathbf{x}})_{i}\\!\leq\\!a_{i}\left(1\\!-\\!\sqrt{\frac{2}{G_{\delta}W_{a}}\ln\Big{(}\frac{3m}{\delta}\Big{)}}\right)G_{\delta},\exists
i\right\\}\leq\frac{\delta}{3m}.\\!\\!\\!\\!\\!$ (41)

By using union bound and (39) and (41), we have that events 1)–3) occur with
probability less than
$\frac{\delta}{3}+m\cdot\frac{\delta}{3m}+r\cdot\frac{\delta}{3r}=\delta$, and
the proof is complete. ∎

##  Appendix B Proof of Lemma 2

###### Proof.

Similar to the case when $0<G_{\delta}\leq 1$, we can have the expectation
equations for the bad cases as in Eqns. (36)-(38). We also can view each
$\hat{x}_{j}$ as a sum of independent random variables in $[0,1]$ in the same
way. Then, we have that
$(\mathbf{A}\hat{\mathbf{x}})_{i}\frac{W_{a}}{a_{i}}=(\sum_{j}[\mathbf{A}]_{ij}\hat{x}_{j})\frac{W_{a}}{a_{i}}$
is also a sum of independent random variables in $[0,1]$. Using the Chernoff
bound, we have that:

$\displaystyle\mathrm{Pr}\\{(\mathbf{A}\hat{\mathbf{x}})_{i}\frac{W_{a}}{a_{i}}\leq(1-\epsilon)G_{\delta}W_{a}\\}\leq\exp(-\epsilon^{2}\frac{G_{\delta}W_{a}}{2}).$

Setting $(1-\epsilon)G_{\delta}\\!=\\!1$, i.e.,
$\epsilon=1-\frac{1}{G_{\delta}}$, we have:

$\displaystyle\mathrm{Pr}\left\\{(\mathbf{A}\hat{\mathbf{x}})_{i}\frac{W_{a}}{a_{i}}\leq
W_{a}\right\\}\leq\exp\left(-\Big{(}1-\frac{1}{G_{\delta}}\Big{)}^{2}G_{\delta}\frac{W_{a}}{2}\right).\\!\\!\\!$
(42)

Forcing
$\exp\\!\left(-(1-\frac{1}{G_{\delta}})^{2}G_{\delta}\frac{W_{a}}{2}\right)\\!=\\!\frac{\delta}{3m}$
and solving $G_{\delta}$, we have:

$\displaystyle G_{\delta}\triangleq
1+\frac{\ln(3m/\delta)}{W_{a}}+\sqrt{\bigg{(}\frac{\ln(3m/\delta)}{W_{a}}\bigg{)}^{2}+\frac{2\ln(3m/\delta)}{W_{a}}}.\\!\\!\\!$
(43)

Using (37), the Chernoff bound, and following similar arguments, we have:

$\displaystyle\mathrm{Pr}\Big{\\{}(\mathbf{B}\hat{\mathbf{x}})_{i}\frac{W_{b}}{b_{i}}>(1+\epsilon)G_{\delta}W_{b}\Big{\\}}\leq\exp\Big{(}-\epsilon^{2}\frac{G_{\delta}W_{b}}{3}\Big{)}.$

Forcing
$\exp\left(-\epsilon^{2}\frac{G_{\delta}W_{b}}{3}\right)=\frac{\delta}{3r}$
and solving for $\epsilon$, we have
$\epsilon=(\frac{3}{G_{\delta}W_{b}}\ln(\frac{3r}{\delta}))^{\frac{1}{2}}$. It
follows that

$\displaystyle\mathrm{Pr}\Big{\\{}(\mathbf{B}\hat{\mathbf{x}})_{i}\frac{W_{b}}{b_{i}}>(1+(\frac{3}{G_{\delta}W_{b}}\ln(\frac{3r}{\delta}))^{\frac{1}{2}})G_{\delta}W_{b}\Big{\\}}\leq\frac{\delta}{3r},$

which implies that:

$\displaystyle\\!\\!\\!\mathrm{Pr}\left\\{(\mathbf{B}\hat{\mathbf{x}})_{i}>b_{i}\left(1+\sqrt{\frac{3}{G_{\delta}W_{b}}\ln\Big{(}\frac{3r}{\delta}\Big{)}}\right)G_{\delta},\exists
i\right\\}\leq\frac{\delta}{3r}.\\!\\!\\!\\!\\!$ (44)

By using union bound and (42) and (44), we have that events 1)–3) occur with
probability less than
$\frac{\delta}{3}+m\cdot\frac{\delta}{3m}+r\cdot\frac{\delta}{3r}=\delta$, and
the proof is complete. ∎

##  Appendix C Proof of the Competitive Ratio

We use $OPT$ as the optimal objective value of Problem R-DMLRS, which is also
the optimum to Problem D-R-DMLRS. We let $\hat{\pi}_{i}$ denote the
approximate schedule obtained by Algorithm 2, which inexactly solves Problem
D-R-DMLRS. Let $P_{i}$ and $D_{i}$ be the primal and dual objective values of
Problems R-DMLRS and D-RMLRS after determining the schedule $\hat{\pi}_{i}$ in
Algorithm 1. Let $P_{0}$ and $D_{0}$ be the initial values of Problems R-DMLRS
and D-RMLRS, respectively, where $P_{0}=0$ and
$D_{0}=\sum_{t\in\mathcal{T}}\sum_{h\in\mathcal{H}}\sum_{r\in\mathcal{R}}P_{h}^{r}[0]C_{h}^{r}$.
We also let $P_{I}$ and $D_{i}$ be the final primal and dual objective values
returned by Algorithm 1. We present our main result in Lemma 8.

######  Lemma 8.

If there exists constants $\epsilon\geq 1$, $G_{\delta}>0$ and
$\delta\in(0,1]$ such that
$P_{i}-P_{i-1}\geq\frac{\delta/3G_{\delta}}{\epsilon}(D_{i}-D_{i-1}),\forall
i\in\mathcal{I}$, and if $P_{0}=0$ and $D_{0}\leq\frac{1}{2}OPT$, then
Algorithm 1 is $\frac{6G_{\delta}\epsilon}{\delta}$-competitive.

###### Proof of Lemma 8.

Since $P_{I}=\sum\limits_{i\in\mathcal{I}}(P_{i}-P_{i-1})$, and
$D_{I}-D_{0}=\sum\limits_{i\in\mathcal{I}}(D_{i}-D_{i-1})$, we can have:

$\displaystyle P_{I}$
$\displaystyle=\sum_{i\in\mathcal{I}}(P_{i}-P_{i-1})\geq\frac{\delta/3G_{\delta}}{\epsilon}\sum_{i\in\mathcal{I}}(D_{i}-D_{i-1})$
$\displaystyle=\frac{\delta/3G_{\delta}}{\epsilon}(D_{I}-D_{0}).$

By weak duality theorem [45], we have

$\displaystyle D_{I}\geq OPT\geq P_{I}.$

Thus, it follows that:

$\displaystyle D_{I}-D_{0}\geq\frac{1}{2}OPT,$ $\displaystyle
P_{I}\geq\frac{\delta/3G_{\delta}}{\epsilon}(D_{I}-D_{0})\geq\frac{\delta/3G_{\delta}}{2\epsilon}OPT,$

and the proof is complete. ∎

Next, following similar arguments in [32, 6], we introduce the relationship
between the cost and resource consumption before and after processing one job.
Let $p_{h}^{r,i}[t]$ be the unit cost of type-$r$ resource on server $h$ at
time $t$ after handling job $i$. Let $\rho_{h}^{r,i}[t]$ be the amount of
type-$r$ resource allocated to jobs on server $h$ at time $t$ after processing
the job $i$. For ease of our subsequent analysis, we now define the following
allocation-cost relation that is implied by Algorithm 1:

######  Definition 1.

The allocation-cost relationship for Algorithm 1 with
$\epsilon>1,G_{\delta}>0$ and $\delta\in(0,1]$ is

$\displaystyle
p_{h}^{r,i-1}[t](\rho_{h}^{r,i}[t]-\rho_{h}^{r,i-1}[t])\geq\frac{\delta/3G_{\delta}C_{h}^{r}}{\epsilon}(p_{h}^{r,i}[t]-p_{h}^{r,i-1}[t]),$
$\displaystyle\hskip 144.54pt\forall
i\in\mathcal{I},h\in\mathcal{H},r\in\mathcal{R}.$

The allocation-cost relationship shows that the cost in each time slot for
scheduling a new job is bounded by the increase of term
$C_{h}^{r}p_{h}^{r}[t]$ in Problem D-R-DMLRS, and the possible increment
introduced by randomized rounding in Algorithm 4. This is ensured by the
update of the price function and the rounding scheme, respectively.

######  Lemma 9.

If the allocation-cost relationship holds for $\epsilon\geq 1,G_{\delta}>0$
and $\delta\in(0,1]$, then Algorithm 1 ensures
$P_{i}-P_{i-1}\geq\frac{\delta/3G_{\delta}}{\epsilon}(D_{i}-D_{i-1}),\forall
i\in\mathcal{I}$.

###### Proof of Lemma 9.

For any job $i\in\mathcal{I}$, if job $i$ is rejected, then we have
$P_{i}-P_{i-1}=D_{i}-D_{i-1}=0$ according to Problems R-DMLRS and D-R-DMLRS,
the inequality must hold. If job $i$ is accepted with schedule $\pi_{i}$,
i.e., $x_{\pi_{i}}=1$, then the increment value of the primal objective value
$P_{i}$ is

$\displaystyle P_{i}-P_{i-1}=u_{i}(t_{\pi_{i}}-a_{i}).$

Since $x_{\pi_{i}}=1$, according to Algorithm 1, the constraint (11) is
binding. Then, we can have

$\displaystyle u_{i}(t_{\pi_{i}}-a_{i})$
$\displaystyle=\lambda_{i}+\sum_{t\in\mathcal{T}(\pi_{i})}\sum_{h\in\mathcal{H}(\pi_{i}[t])}\sum_{r\in\mathcal{R}}(\alpha_{i}^{r}w_{ht}^{\pi_{i}}+\beta_{i}^{r}s_{ht}^{\pi_{i}})p_{h}^{r}[t]$
$\displaystyle=\lambda_{i}+\sum_{t\in\mathcal{T}(\pi_{i})}\sum_{h\in\mathcal{H}(\pi_{i}[t])}\sum_{r\in\mathcal{R}}p_{h}^{r}[t](\rho_{h}^{r,i}[t]-\rho_{h}^{r,i-1}[t]).$

Similarly, we can have the increment value of the dual objective value $D_{i}$
as follows:

$\displaystyle
D_{i}-D_{i-1}\\!=\\!\lambda_{i}+\sum_{t\in\mathcal{T}(\pi_{i})}\sum_{h\in\mathcal{H}(\pi_{i}[t])}\sum_{r\in\mathcal{R}}(p_{h}^{r,i}[t]-p_{h}^{r,i-1}[t])C_{h}^{r}.$

Summing up the allocation-cost relationship over all
$t\in\mathcal{T}(\pi_{i}),h\in\mathcal{H}(\pi_{i}[t]),r\in\mathcal{R}$, we
have

$\displaystyle P_{i}-P_{i-1}$
$\displaystyle\geq\frac{\delta/3G_{\delta}}{\epsilon}(D_{i}-D_{i-1}-\lambda_{i})+\lambda_{i}$
$\displaystyle=\frac{\delta/3G_{\delta}}{\epsilon}(D_{i}-D_{i-1})+(1-\frac{\delta/3G_{\delta}}{\epsilon})\lambda_{i}.$

As $\lambda_{i}\geq 0$, $\epsilon,G_{\delta}>0$ and $\delta\in(0,1]$, we have

$\displaystyle
P_{i}-P_{i-1}\geq\frac{\delta/3G_{\delta}}{\epsilon}(D_{i}-D_{i-1}).$

This completes the proof of Lemma 9. ∎

For specific $h\in\mathcal{H},r\in\mathcal{R}$, we define $\epsilon_{h}^{r}$
as the corresponding parameter in the allocation-cost relationship for any job
$i\in\mathcal{I}$ and any time slot $t\in\mathcal{T}$. Then, it holds that
$\epsilon=\max_{h,r}\\{\epsilon_{h}^{r}\\}$. Without loss of generality, we
assume that the resource demand of each worker or parameter server is much
smaller compared to the capacity of that resource on one server, i.e.,
$\alpha_{i}^{r}\ll C_{h}^{r},\beta_{i}^{r}\ll C_{h}^{r}$. This is common in
real-world machine learning system as large percentage of resources in the
whole server. As $\rho_{h}^{r}[t]$ increases from 0 to $C_{h}^{r}$, then we
can claim that $d\rho_{h}^{r}[t]=\rho_{h}^{r,i}-\rho_{h}^{r,i-1}$, and derive
a differential version of the allocation-cost relationship, which is defined
as follows:

######  Definition 2.

The differential allocation-cost relationship for Algorithm 1 with
$\epsilon_{h}^{r}\geq 1$ is

$\displaystyle
p_{h}^{r}[t]d\rho_{h}^{r}[t]\geq\frac{C_{h}^{r}}{\epsilon_{h}^{r}}dp_{h}^{r}[t],\forall
t\in\mathcal{T},h\in\mathcal{H},r\in\mathcal{R}.$

Next we show that a feasible $\epsilon_{h}^{r}$ satisfies the differential
allocation-cost relationship with price function $p_{h}^{r}[t]$ defined in
(12).

######  Lemma 10.

$\epsilon_{h}^{r}=\ln\frac{U^{r}}{L}$, and the price function defined in (12)
satisfy the differential allocation-cost relationship.

###### Proof of Lemma 10.

The derivation of the marginal cost function is

$\displaystyle
dp_{h}^{r}[t]=p_{h}^{r^{\prime}}(\rho_{h}^{r}[t])d\rho_{h}^{r}[t]=L(\frac{U^{r}}{L})^{\frac{\rho_{h}^{r}[t]}{C_{h}^{r}}}\ln(\frac{U^{r}}{L})^{\frac{1}{C_{h}^{r}}}d\rho_{h}^{r}[t].$

The differential allocation-cost relationship is

$\displaystyle
L(\frac{U^{r}}{L})^{\frac{\rho_{h}^{r}[t]}{C_{h}^{r}}}d\rho_{h}^{r}[t]\geq\frac{C_{h}^{r}}{\epsilon_{h}^{r}}L(\frac{U^{r}}{L})^{\frac{\rho_{h}^{r}[t]}{C_{h}^{r}}}\ln(\frac{U^{r}}{L})^{\frac{1}{C_{h}^{r}}}d\rho_{h}^{r}[t],$

which holds for $\epsilon_{h}^{r}\geq\ln(\frac{U^{r}}{L})$. Then, we can set
$\epsilon\\!=\\!\max\limits_{r\in\mathcal{R}}(1,\ln(\frac{U^{r}}{L}))$, which
satisfies the differential allocation-cost relationship. This completes the
proof. ∎

With the aforementioned lemmas, we are now in a position to prove Theorems 5
and 6. Note that the Theorems provide probabilistic guarantees. We first
analyze the performance ratio, which is followed by the probabilistic
feasibility discussion for both cases.

###### Proof of Theorems 5 and 6.

According to Lemma 10, the marginal cost function used in Algorithm 1
satisfies the differential allocation-cost relationship with
$\epsilon=\max_{r}(1,\ln\frac{U^{r}}{L})$. Since the resource demand in a job
$i$ is much smaller than the capacity, we can derive

$\displaystyle d\rho_{h}^{r}[t]=\rho_{h}^{r,i}-\rho_{h}^{i-1}[t],$
$\displaystyle
dp_{h}^{r}[t]=p_{h}^{r^{\prime}}(\rho_{h}^{r}[t])(\rho_{h}^{r,i}[t]-\rho_{h}^{r,i-1}[t])=p_{h}^{r,i}[t]-p_{h}^{r,i-1}[t].$

So, the differential allocation-cost relationship in Definition 2 implies the
allocation-cost relationship in Definition 1 holds for
$\epsilon=\max_{r}(1,\ln\frac{U^{r}}{L})$.

According to Algorithm 1, we note that

$\displaystyle\frac{1}{\mu}\leq\frac{\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i}))\rceil\sum_{r\in\mathcal{R}}(\alpha_{i}^{r}+\beta_{i}^{r})}{T\sum_{h\in\mathcal{H}}\sum_{r\in\mathcal{R}}C_{h}^{r}},\forall
i\in\mathcal{I}.$

Then, the minimum amount of overall resource consumption of job $i$ can be
computed as:

$\frac{T\sum_{h\in\mathcal{H}}\sum_{r\in\mathcal{R}}C_{h}^{r}}{\mu}\\\
\leq\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i}))\rceil\sum_{r\in\mathcal{R}}(\alpha_{i}^{r}+\beta_{i}^{r}).$

Then, it follows that:

$\displaystyle D_{0}=\sum_{t,h,r}LC_{h}^{r}$
$\displaystyle=\sum_{t,h,r}\min_{i\in\mathcal{I},\pi_{i}\in\Pi_{i}}\frac{1/(2\mu)u_{i}(t_{\pi_{i}}-a_{i})C_{h}^{r}}{\sum_{r\in\mathcal{R}}\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i})\rceil(\alpha_{i}^{r}+\beta_{i}^{r})}$
$\displaystyle=\frac{T\sum_{h,r}C_{h}^{r}}{2\mu}\min_{i,\pi_{i}}\frac{u_{i}(t_{\pi_{i}}-a_{i})}{\sum_{r\in\mathcal{R}}\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i})\rceil(\alpha_{i}^{r}+\beta_{i}^{r})}$
$\displaystyle\leq\frac{1}{2}\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i}))\rceil\sum_{r\in\mathcal{R}}(\alpha_{i}^{r}+\beta_{i}^{r})$
$\displaystyle\min_{i\in\mathcal{I},\pi_{i}\in\Pi_{i}}\frac{u_{i}(t_{\pi_{i}}-a_{i})}{\sum_{r\in\mathcal{R}}\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i})\rceil(\alpha_{i}^{r}+\beta_{i}^{r})},\forall
i\in\mathcal{I}.$ $\displaystyle\overset{(a)}{\leq}\frac{1}{2}\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i}))\rceil\sum_{r\in\mathcal{R}}(\alpha_{i}^{r}+\beta_{i}^{r})$
$\displaystyle\frac{u_{i}(t_{\pi_{i}}-a_{i})}{\sum_{r\in\mathcal{R}}\lceil
E_{i}K_{i}(\tau_{i}+2g_{i}\gamma_{i}/(b_{i}^{(e)}F_{i})\rceil(\alpha_{i}^{r}+\beta_{i}^{r})}$
$\displaystyle\leq\frac{1}{2}u_{i}(t_{\pi_{i}}-a_{i})\overset{(b)}{\leq}\frac{1}{2}OPT,$

where (a) follows by selecting
$(i,\pi)=\arg\min_{i\in\mathcal{I},\pi_{i}\in\Pi_{i}}u_{i}(t_{\pi_{i}}-a_{i})$,
and (b) follows from the assumption that the offline optimal solution accepts
at least one job, which is reasonable in real-world machine learning system.
Then we have $OPT\geq\min_{i,\pi}u_{i}(t_{\pi_{i}}-a_{i})$. According to
Lemmas 8 and 9, the competitive ratio is proved.

Recall that the randomized rounding algorithm is a key component in our
algorithm. Toward this end, we show the probability of obtaining a feasible
solution with the proved competitive ratio. Here, we consider the following
two cases:

1)  When $0<G_{\delta}\leq 1$ (Theorem 5): According to Theorem 3, the
probability of violating the cover constraint is no greater than $\delta/3$ at
each randomized rounding iteration. Recall that our Algorithm 1 runs a
predetermined number of $S$ iterations to find a feasible integer solution.
Thus the probability that no feasible integer solution returned after $S$
iterations rounding is at most $(\delta/3)^{S}$. It then follows that the
probability of at least one feasible integer solution found is at least
$1-(\delta/3)^{S}$. Moreover, the number of states $(t,v)$ in the dynamic
programming for each job $i$ is $O(TK_{i}E_{i})$. Therefore, with probability
greater than $(1-(\delta/3)^{S})^{T_{i}K_{i}E_{i}}$, PD-ORS in Algorithm 1
returns a feasible integer solution with the proved competitive ratio.

2)  When $G_{\delta}>1$ (Theorem 6): According to Theorem 4, the probability
of violating the packing constraint is no greater that $\delta/3(HR+1)$ at
each randomized rounding iteration. Following the similar arguments in 1), we
can show that with probability greater than
$(1-(\delta/3(HR+1))^{S})^{T_{i}K_{i}E_{i}}$, PD-ORS in Algorithm 1 returns a
feasible integer solution with the proved competitive ratio, and the proof is
complete. ∎