#  Cassini: Network-Aware Job Scheduling in Machine Learning Clusters

Sudarsanan Rajasekaran† Manya Ghobadi† Aditya Akella‡

†Massachusetts Institute of Technology ‡UT Austin

###### Abstract

We present Cassini, a network-aware job scheduler for machine learning (ML)
clusters. Cassini introduces a novel geometric abstraction to consider the
communication pattern of different jobs while placing them on network links.
To do so, Cassini uses an affinity graph that finds a series of time-shift
values to adjust the communication phases of a subset of jobs such that the
communication patterns of jobs sharing the same network link are interleaved
with each other. Experiments with 13 common ML models on a 24-server testbed
demonstrate that compared to the state-of-the-art ML schedulers, Cassini
improves the average and tail completion time of jobs by up to 1.6$\times$ and
2.5$\times$, respectively. Moreover, we show that Cassini reduces the number
of ECN marked packets in the cluster by up to 33$\times$.

##  1 Introduction

The ever-growing increase in dataset and model sizes of deep learning has
created a massive demand for efficient GPU clusters. Several studies have
demonstrated that as the number of GPUs increases, the communication overhead
of distributed Machine Learning (ML) training workloads quickly takes up a
significant portion of training iteration time [42, 29, 24, 50, 12, 40, 9].
Yet, state-of-the-art ML schedulers, tend to ignore the communication pattern
of ML training jobs when placing workers on GPUs.

In this paper, we develop a simple but effective approach, called Cassini,
that integrates with existing ML schedulers to allow them to efficiently place
multiple ML jobs on network links while minimizing the chances of network
congestion. Our approach requires no special support, such as reservations and
priorities, from switches/NICs and does not require any changes to the
congestion control protocol.

Figure 1: The traffic pattern of different parallelization strategies when
training GPT-1, GPT-2, and GPT-3 models.

We demonstrate that for a specific combination of jobs, introducing a small
time-shift to delay the start of one of the iterations enables Cassini to
interleave the computation and communication patterns of different jobs,
thereby improving the training time. We refer to such combinations of jobs as
compatible and develop Cassini as a pluggable module to augment prior ML
schedulers to consider a novel compatibility metric when determining where to
place jobs and control how jobs compete on network links.

Augmenting ML schedulers to take links and servers into account is inherently
challenging because jobs are likely to traverse multiple links and may compete
with different jobs on different links. To address this challenge, we propose
a geometric abstraction that leverages the periodic communication pattern of
Deep Neural Network (DNN) training workloads. The key idea of our abstraction
is to “roll” time around a circle whose perimeter is proportional to the
training iteration time of ML jobs. To determine the compatibility score of
two (or more) jobs on a link, Cassini places each job on its corresponding
circle and overlays the circles on top of each other. It then uses an
optimization formulation to rotate the circles into a position that maximizes
interleaving. The rotation angle of each job corresponds to a time-shift value
to delay the start of the next immediate training iteration to achieve
compatibility.

Looking beyond a single link and extending to jobs running across a topology,
we generalize the geometric abstraction to cluster-level by introducing a
bipartite Affinity graph where vertices are a subset of jobs and links. An
edge in the Affinity graph indicates a job is traversing a link. We then use a
new graph traversal algorithm to find unique time-shifts for all jobs while
maintaining their compatibility on all links. Using our geometric abstraction
and Affinity graph, we augment Themis [35] and Pollux [45] with $\approx$1000
lines of code.

To evaluate Cassini, we build a testbed with 24 servers, each with one NVIDIA
A100 GPU [6] and one 50 Gbps RDMA NIC. Our experiments with 13 representative
DNN models (VGG11 [22], VGG16 [15], VGG19 [28], ResNet50 [23], WideResNet101
[63], BERT [17], RoBERTa [34], XLM [14], CamemBERT [38], GPT-1 [46], GPT-2
[47], GPT-3 [8], and DLRM [4]) show that Cassini improves the tail completion
time of jobs by up to 2.2$\times$ and 2.5$\times$, compared to Themis [35] and
Pollux [45], respectively. Moreover, we show that Cassini reduces the number
of ECN marked packets in the cluster by up to 33$\times$.

##  2 Background and Motivation

###  2.1 Distributed DNN Training Traffic Pattern

Cassini is designed for large GPU clusters with hundreds of training jobs
distributed with data, pipeline, and/or model parallel training paradigms. To
this end, we study the impact of different parallelization strategies on
network demand using a series of measurements. Each server in our testbed has
one A100 GPU and one ConnectX-5 Mellanox RDMA NIC with 50 Gbps capacity. In
all our experiments, we choose batch sizes such that the GPU utilization is
higher than 80% and that intra-job pipelining is enabled.

Data parallelism. In data parallel training, the DNN model is copied into the
memory of each GPU while the dataset is distributed across them. Figure 1(a)
shows the communication pattern of a GPT-1 [46] model (12 layers, 9 GB memory)
distributed across four GPU servers using data parallelism. The figure shows
the traffic pattern of three back-to-back training iterations. Each iteration
contains a forward pass with near-zero network demand, followed by a period of
high utilization corresponding to the backpropagation and AllReduce phases.

Model/Pipeline parallelism. In model parallel training, the DNN model is
partitioned across workers [30, 25], and parts of the DNN model are computed
on different workers. The two common techniques for model parallelism are
tensor parallelism and pipeline (or layer) parallelism [7]. In pipeline
parallelism, the model is partitioned vertically at the layer boundaries [42,
24]. Figure 1(b) shows the communication pattern of a GPT-2 [47] model (24
layers, 27 GB memory) distributed across two servers using pipeline
parallelism. We partition the model vertically in half (i.e., server1 contains
layer1 to layer12 and server2 contains layer13 to layer24) and use PipeDream’s
approach [42] to divide the batch size into three minibatches. The three small
communication peaks during the forward pass correspond to the activation
parameters of these three minibatches. The heavy communication demand
following the peaks corresponds to the AllReduce operation between the
embedding layers in the model.

Figure 2: Impact of interleaving the Up-Down phases of two VGG19 jobs sharing
link $l_{1}$.

Model/Tensor parallelism. Another variant of model parallel training is tensor
parallelism [53, 54]. Tensor parallelism techniques partition the model
horizontally such that different tensors are distributed across workers [27,
57]. Figure 1(c) shows the communication pattern of a GPT-3 [8] model (96
layers, 35 GB memory) distributed across two servers using tensor parallelism.
We partition the model horizontally in half, where each server contains half
of all the layers. The figure shows that both forward and backpropagation
phases introduce roughly 25 Gbps traffic followed by a short period of near-
zero network demand during data loading.

Hybrid data/pipeline/tensor parallelism. Today’s DNN training systems tend to
use a hybrid of data/pipeline/tensor parallelism to train large DNN models
[41, 29, 18, 59]. Figure 1(d) shows the communication pattern of a GPT-3 [8]
model (96 layers, 155 GB memory) distributed across eight servers using hybrid
data/pipeline/tensor parallelism. We use pipeline parallelism to partition the
model’s layers vertically into two parts. Then, we divide the layers in each
partition horizontally to obtain a total of four partitions. Next, we assign
each of these four partitions to a server. Finally, we replicate the same
process across another group of four servers and use data parallelism to
distribute the data between these two groups of four servers. The figure shows
the communication demand of the forward, backpropagation, and AllReduce phases
where each phase has a different network demand.

Key takeaways. We repeat the above experiments using common DNN models, such
as BERT [17], DLRM [4], WideResNet101 [63], RoBERTa [34], VGG [55] and observe
similar traffic patterns. Our key takeaways are: ($i$) the network demand
repeats itself across all iterations, as long as the training hyper-parameters
remain the same; ($ii$) the network demand of an iteration may consist of
multiple Up and Down phases. The exact magnitude of the network demand during
these Up and Down phases depends on the parallelization strategy and hyper-
parameters. For instance, Figure 1(d) shows each training iteration has six
Up-Down phases, labeled as 1 to 6 . Section 3 describes Cassini’s approach to
capture the duration and bandwidth of Up-Down phases.

###  2.2 Interleaving the Up and Down Phases

Cassini’s goal is to augment ML schedulers to consider the traffic demand of
training jobs when making scheduling decisions. In particular, given the key
takeaways in the previous section, we aim to interleave the bandwidth demand
of Up and Down phases of different jobs to leverage the periodic network
demand of distributed DNN training jobs.

To demonstrate the power of Up-Down network demand interleaving, consider two
data parallel training jobs, $j_{1}$ and $j_{2}$, as shown in Figure 2(a).
Each job has one Up and one Down phase at every training iteration. We run
each job for 1,000 iterations under two scenarios. In the first scenario, two
VGG19 [55] jobs start simultaneously and share $l_{1}$ fairly. The
communication uses the RDMA-based DCQCN congestion control algorithm [68].
Figure 2(b) shows that both jobs achieve roughly 22 Gbps bandwidth (i.e., half
of $l_{1}$’s capacity). In the second scenario, shown in Figure 2(c), we
interleave the Down phase of $j_{1}$ with the Up phase of $j_{2}$ and vice
versa, by shifting the start time of $j_{2}$ by 120 ms (Section 3 describes
how we obtained this value). In this scenario, the jobs do not compete for
bandwidth during their respective Up phases, giving both jobs the entire
available bandwidth. Figure 2(d) plots the CDF of training iteration times for
both scenarios demonstrating that scenario2 accelerates the 90th percentile
tail iteration time of both jobs by 1.26$\times$.

Perfectly interleaving the Up and Down phases of different jobs is not always
possible. For instance, when BERT [17] and VGG19 [55] models share a link, no
suitable time-shift can achieve perfect interleaving. But when WideResNet101
[63] and VGG16 [55] share a link, shifting VGG16 by 150 ms enables perfect
interleaving. Instead of relying on perfectly matching Up and Down phases of
jobs, we define a metric called compatibility rank that captures the potential
degree of interleaving across jobs sharing the network. Section 3 describes a
novel technique to determine the compatibility rank and the amount of required
time-shift to achieve it.

##  3 Geometric Abstraction

Figure 3: Cassini’s unified circles for jobs with different iteration times.

Figure 4: Geometric circle for the job in Fig. 1(d).

Consider a time-series representation of the network demand for a job running
in a dedicated cluster with no congestion. As shown in Section 2, different
training jobs have different Up and Down patterns but the duration and
bandwidth demand of the same job remain more or less the same across training
iterations. The key idea of our abstraction is to roll time around a circle
whose perimeter is equal to the iteration time of a job. Consequently, the Up-
Down phases of all iterations will appear on approximately the same angles of
the circle.

Figure 17(a) illustrates the time-series network demand of a data parallel
VGG16 training job with a training iteration time of 255 ms. Figure 17(b)
shows a circle with perimeter 255 units where the time-series data is plotted
around it. The figure demonstrates that the Up and Down phases of different
iterations cover the same angles of the circle. Our geometric abstraction
captures this property, as shown in Figure 17(c). The perimeter of the circle
is the iteration time, set to 255 units. The Down phase spans 141 units,
represented by the uncolored arc with 200∘ angle, starting at 0∘, on the
x-axis. The Up phase represented by the colored arc occupies the remainder of
the circle.

Rotate the circle to interleave Down and Up phases of different jobs. To
determine the compatibility score of two (or more) jobs on a link, we place
each job on its corresponding circle and overlay the circles on top of each
other. Congestion happens when the total bandwidth demand of a particular
angle is higher than the capacity of the link, as shown in Figure 17(a). To
find the best interleaving, we rotate the circles to a position where the
summation of the bandwidth demands is less than the capacity of the link for
all angles in the circle, as shown in Figure 17(b). If such a rotation is
found, the jobs are fully compatible.

Capturing jobs with different iteration times. The above technique only works
when the perimeter of circles is the same. To generalize to the case where
jobs have different iteration times, we place each job on a unified circle
whose perimeter is equal to the Least Common Multiple (LCM) of the iteration
time of all jobs competing on the link. For instance, consider two jobs
$j_{1}$ and $j_{2}$ competing on a bottleneck link with iteration times 40 ms
and 60 ms, respectively. To determine the compatibility score of the two jobs,
we place them on a circle with a perimeter equal to $LCM(40,60)=120$ units.
Figure 4(a) shows $j_{1}$ on this unified circle. As the perimeter of the
circle is 3$\times$ $j_{1}$’s iteration time, there are three periods of Up
and Down phases in the figure. Similarly, Figure 4(b) shows $j_{2}$ on the
unified circle. We then overlay the unified circles on top of each other
(shown in Figure 4(c)) and rotate the circles to determine the compatibility
score. Figure 4(d) shows that by rotating $j_{1}$ by $\Delta=30^{\circ}$
counter-clockwise, the sum of bandwidth demands on all angles of the unified
circles is lower than the link capacity, giving these two jobs a compatibility
score of 1 (i.e., fully compatible).

Capturing the bandwidth demand of model parallel training jobs. For clarity of
presentation, the examples in this section contained data parallel training
jobs with one Up and one Down phase during each iteration. Cassini’s geometric
abstraction is generic and can capture more complex traffic patterns induced
by various parallelization paradigms. Consider the communication pattern of
the GPT-3 model with hybrid data/pipeline/tensor parallelism shown in Figure
1(d). Here, GPT-3’s communication pattern consists of six Up-Down phases with
different durations and bandwidth demands. The geometric circle of this job
contains six colored arcs where the length and color intensity of each arc
corresponds to the duration and bandwidth demand of each Up-Down phase of the
model, shown in Figure 4. Next, we formalize our geometric representation and
show an optimization formulation that uses the geometric abstraction to find
rotation angles to interleave the Up-Down phases of multiple jobs sharing a
link, irrespective of the parallelization strategy.

Finding rotation angles. Once jobs are placed on their unified circles,
Cassini uses an optimization formulation, shown in Table 1, to find the best
angle of rotation for jobs to maximize their compatibility.

Optimization input. The input is a set of ML jobs $J^{l}=\\{j\\}$ competing on
a link $l$. We profile each job $j$ to construct its unified circle, denoted
by unified_circlej. The perimeter of the unified circle is the LCM of the
iteration times of all jobs $j\in J^{l}$. The data structure of
unified_circlej contains a series of bandwidth demands,
$bw\\_circle_{j}(\alpha)$, where $\alpha\in[0,2\pi]$ identifies an arc of the
circle that corresponds to an Up or Down phase in the communication pattern.
The total capacity of link $l$ is denoted by $C^{l}$.

Optimization objective and output. The optimization goal is to overlay the
unified circles of each job and rotate them such that the excess bandwidth
demand across all angles is minimized. We define the compatibility score as
$score=1-$average($Excess(demand_{\alpha}))$, where $Excess$ is the excess
bandwidth demand of all jobs at a particular angle $\alpha$ (Equation 1). To
make the score a unitless metric, we divide the average excess bandwidth by
the link capacity $C^{l}$. Note that when the excess bandwidth demand is zero,
the compatibility score is 1 (i.e., 100% compatible). However, when there are
many jobs with large excess bandwidth demands, it is possible for the score to
become negative indicating a highly incompatible combination. The optimization
objective is to maximize this compatibility score, and the output of the
optimization is a rotation angle $\Delta_{j}^{l}$ for each job.

Input |  $J^{l}=\\{j\\}$ |  Set of ML jobs $j\in J^{l}$ competing on link $l$.
---|---|---
$\\{$unified_circle${}_{j}\\}$ |  Set of unified circles for $\forall j\in J$. Each circle is a data structure that contains the angles and bandwidth demand of Up or Down phases.
$bw\\_circle_{j}(\alpha)$ |  Bandwidth demand at angle $\alpha$ on unified_circlej
$r_{j}$ |  Number of iterations of $j$ in its unified_circlej.
$A=\\{\alpha\\}$ |  Set of discrete angles $\alpha\in$ [0,2$\pi$]. $|A|$ denotes the number of discrete angles.
$C^{l}$ |  Total link capacity of link $l$.
Output |  $demand_{\alpha}$ |  Total bandwidth demand at angle $\alpha$ when considering the demand of all jobs $j\in J$.
$\Delta_{j}^{l}$ |  Rotation angle of $j\in J$ on link $l$, in radians.
|  $score$ |  Compatibility score of jobs sharing link $l$.

Auxiliary definitions: $\displaystyle Excess(demand_{\alpha})=$
$\displaystyle\begin{cases}demand_{\alpha}-C^{l}&ifdemand_{\alpha}>C^{l}\\\
0&otherwise\end{cases}$ (1) Maximize: $\displaystyle\
score=1-\frac{\sum_{\alpha}Excess(demand_{\alpha})}{|A|C}$ (2) Subject to:
$\displaystyle\forall\alpha:$
$\displaystyle\sum_{j}bw\\_circle_{j}(\alpha-\Delta_{j}^{l})\leq
demand_{\alpha}$ (3) $\displaystyle\forall\Delta_{j}^{l}:$ $\displaystyle\
0\leq\Delta_{j}^{l}\leq\frac{2\pi}{r_{j}}$ (4)

Table 1: Cassini’s optimization formulation.

Optimization constraints. Equation 3 computes the sum of the bandwidth demands
across all the jobs sharing link $l$ at a particular angle $\alpha$ on their
unified circles, rotated by angle ${\Delta_{j}^{l}}$. We bound this value by
the output parameter $demand_{\alpha}$. Equation 4 bounds the rotation angle
$\Delta_{j}^{l}$ between 0 and $\frac{2\pi}{r_{j}}$ because the
unified_circlej contains $r_{j}$ iterations of job $j$. Hence, setting an
upper limit of $\frac{2\pi}{r_{j}}$ ensures that the rotation angle is in the
first iteration and eliminates duplicate solutions.

##  4 Augmenting ML schedulers with Cassini

This section describes how Cassini extends its link-level geometric
abstraction to the entire cluster.

###  4.1 Cassini Affinity Graph

Translating angular rotations to time-shifts. Consider a set of jobs $j\in
J^{l}$ sharing link $l$. Using the formulation in Table 1, Cassini computes a
rotation angle $\Delta_{j}^{l}$ for $\forall j\in J^{l}$ such that the
compatibility score is maximized. Each $\Delta_{j}^{l}$ corresponds to a time-
shift $t_{j}^{l}$ to delay the start time of $j$ to maximize its compatibility
with all other jobs in $J^{l}$. Given that the perimeter of the unified
circle, $p^{l}$, is the LCM of the iteration times of all jobs $j\in J^{l}$,
Cassini computes these time-shifts by multiplying the normalized rotation
angle with $p^{l}$. Formally:

$\forall j\in J^{l},t_{j}^{l}=(\frac{\Delta_{j}^{l}}{2\pi}\times
p^{l})\mod\textit{iter\\_time}_{j}$ (5)

Challenge: ensuring a unique time-shift for each job. In a large-scale
cluster, jobs are likely to traverse multiple links, and they may compete with
different jobs on different links. Consider a case depicted in Figure 5 where
job $j_{1}$ competes with job $j_{2}$ on link $l_{1}$, and $j_{2}$ competes
with job $j_{3}$ on link $l_{2}$. Theoretically, it is possible to migrate the
jobs to pack workers of the same job under the same rack to avoid sharing the
links altogether, but our experiments show that today’s ML scheduling systems
frequently end up with fragmented placements because of the dynamic nature of
their scheduling decisions and job arrival patterns. In fact, no scheduler
guarantees it can maintain perfect placement throughout time without
continuously migrating jobs to defragment the cluster. For the case depicted
in Figure 5, computing the time-shifts of $j_{2}$ using Equation 5 would
result in two time-shift values $t_{j_{2}}^{l_{1}}$ and $t_{j_{2}}^{l_{2}}$.
Given the interdependence between all servers participating in a training job,
Cassini must find a unique time-shift value for each job across links without
compromising the compatibility on any link.

Figure 5: Example illustrating a cluster-scale compatibility challenge:
Cassini must ensure a unique time-shift for $j_{2}$.

Simple approach. A potential approach to address the above challenge is to
simply break the tie by choosing one of the $t_{j}^{l}$ values at random. But
this approach cancels out the benefits of compatibility because it does not
respect the carefully computed time-shifts for different links.

Figure 6: Cassini’s Affinity graph. Traversing left to right incurs a negative
sign on the weight of edges and vice versa.

Complex approach. Another potential approach is to expand the footprint of our
geometric abstraction from link-level to cluster-level. This approach requires
expanding the optimization formulation in Table 1 to include all jobs that
share their paths with any other jobs in the cluster and to encode a unique
$\Delta_{j}$ in the constraints. This approach is not scalable because it
requires expanding the perimeter of the unified circle to become the LCM of
the iteration times of a large number of jobs in the cluster. Thus, finding a
unique rotation angle for each job requires adding an exponential number of
constraints to the optimization formulation which increases the complexity and
overhead of the formulation dramatically.

1:procedure BFSAffinityGraph

2:$\triangleright$ Input $Graph\leavevmode\nobreak\ G=(U,V,E)$: Cassini’s
Affinity graph

3:$\triangleright$ Output time_shiftsG: Time-shifts of jobs in G

4: time_shiftsG = {}

5: for all connected subgraphs $H\in G$, $H=(U_{H},V_{H},E_{H})$ do

6: time_shiftsH = {}

7: $\triangleright$ BFS traversal

8: Mark all vertices $u\in U_{H}$ as not-visited

9: u = randomly_select_vertex($U_{H}$)

10: $t_{u}=0$ and mark $u$ as visited

11: $\triangleright$ Only enqueue vertices from $U$ (jobs)

12: Q.enqueue($u$)

13: while Q is not empty do

14: $j=$ Q.dequeue()

15: $\triangleright$ Find the corresponding links and jobs

16: for all neighbors $l$ of $j$ do

17: for all neighbors $k$ of $l$ do

18: if $k$ is not visited then

19: Q.enqueue($k$) and mark $k$ as visited

20: $\triangleright$ Find the edge from $U$ to $V$

21: $e_{1}=E_{H}(j,l)$

22: $\triangleright$ Find the edges from $V$ to $U$

23: $e_{2}=E_{H}(l,k)$

24: $\triangleright$ Compute the final time-shift

25: $t_{k}=(t_{j}-w_{e_{1}}+w_{e_{2}})\%$ $iter\\_time_{k}$

26: time_shifts${}_{H}[k]=t_{k}$

27: time_shiftsG = time_shiftsG $\cup$ time_shiftsH

28: return time_shiftsG

Algorithm 1 Traversing the Affinity graph

Cassini’s approach. Cassini introduces a bipartite Affinity graph $G=(U,V,E)$,
where $U$ and $V$ are two sets of vertices, and $E$ denotes the edge set
between $U$ and $V$, shown in Figure 6(a). Each vertex $u\in U$ represents a
job that is sharing its path with other jobs somewhere in the network. Each
vertex $v\in V$ represents a link that carries more than one job. An
undirected edge $e=(j,l)\in E$ exists between a job $j\in U$ and a link $l\in
V$ if $j$ traverses $l$. The weight of edge $e=(j,l)\in E$ is the time-shift
of job $j$ on link $l$; i.e., $w_{e}=t_{j}^{l}$.

Traversing the Affinity graph. Cassini uses a graph traversal algorithm to
find unique time-shifts $t_{j}$ for all jobs $j\in J$ while maintaining
compatibility on all links. To consolidate $t_{j}^{l}$ values for each job $j$
and link $l$ into a unique $t_{j}$ value, Cassini first randomly selects one
of the jobs in the Affinity graph as the reference point with $t_{j}=0$ and
then traverses the graph to compute unique time-shifts for all others.
Algorithm 1 describes the pseudocode of our graph traversal. In the general
case, the Affinity graph is not necessarily a connected graph, hence, the
algorithm traverses each connected subgraph separately (line 5). The traversal
algorithm extends the Breadth First Search (BFS) algorithm in two ways. First,
only vertices in $U$ are added to the BFS queue ($Q$) because the time-shifts
correspond to jobs, not links (lines 9\- 19). Second, traversing from jobs
($j\in U$) to links ($l\in V$) incurs a negative sign on the $t_{j}^{l}$
weight on edge $e=(j,l)$, whereas traversing the reverse direction incurs a
positive sign (lines 21-26). As soon as the vertex corresponding to job $j$ is
visited, its unique time-shift is determined by the algorithm (line 26).

######  Theorem 1 (Correctness and Uniqueness Guarantee).

Given a cluster with $J$ jobs and a loop-free Affinity graph, $G=(U,V,E)$,
Algorithm 1 guarantees both correct and unique time-shifts $t_{j}$ for all
jobs $j\in J$.

###### Proof.

The key insight behind this Theorem is that our graph traversal maintains the
same relative time-shift for all job pairs in the Affinity graph. The full
proof uses induction and is provided in Appendix A along with an example
corresponding to the Affinity graph in Figure 6(b). ∎

Figure 7: Using Cassini to augment Themis [35].

###  4.2 Putting It All Together

This section uses Themis [35] as a running example of a scheduler augmented by
Cassini.

Overview of Themis. Themis uses a fairness metric, called finish-time
fairness, to achieve long-term fairness across the entire cluster by
periodically updating the placement of jobs. To achieve fairness, workers in
Themis lease resources and go through periodic auction epochs to help jobs
that are farthest in terms of their fairness metric bid for more resources.
Themis’s central arbiter determines the global winning bids to maximize the
aggregate improvement in the finish-time fair metrics across all bidding jobs.
To capture network overheads, Themis uses a slowdown penalty based on whether
the workers are under the same rack or across racks.

Augmenting Themis with Cassini. Figure 7 shows how Cassini augments Themis.
First, Cassini modifies Themis’s arbiter to return a set of potential
placement candidates instead of a single placement. Then, Cassini selects the
top placement candidate based on its compatibility metric and computes unique
time-shifts for jobs that share the network. Cassini transfers the time-shifts
to Themis’s agent running on servers. Finally, Themis’s agent applies the
time-shifts at the start of the epoch. Note that Cassini respects the hyper-
parameters, such as batch size or the number of workers, decided by Themis (or
other schedulers that Cassini is augmenting). Next, we describe each step in
detail.

1:procedure CassiniModule

2:$\triangleright$ Input $Jobs$: Array of active training jobs in the cluster

3:$\triangleright$ Input $Links$: Array of all links in the cluster

4:$\triangleright$ Input $Candidates$: Array of candidate placements for jobs

5:$\triangleright$ Output $top\\_placement$, $\\{t_{j}\\}$: Top placement and
time-shifts

6: for $c\in Candidates$ do $\triangleright$ (Loop is executed with threads)

7: $\triangleright$ Construct Cassini’s Affinity graph corresponding

8: to this placement (§4.1)

9: $G_{c}$ = $(U_{c},V_{c},E_{c})$

10: for all $j\in Jobs$, $l\in Links$ do

11: if $j$ shares links with other jobs then

12: $U_{c}=U_{c}\cup j$

13: if $l$ carries more than one job then

14: $V_{c}=V_{c}\cup l$

15: if $j$ is traversing $l$ then

16: $e=$ new Edge between $\\{(j,l)\\}$

17: $E=E\cup e$

18: $w_{e}=0$

19: $\triangleright$ Discard this candidate if Affinity graph has a loop

20: if there is a loop in $G_{c}$ then

21: $Candidates.remove(c)$

22: continue

23: $score_{c}$ = {}

24: for $l\in V_{c}$ do $\triangleright$ (Executed with threads)

25: $\triangleright$ List of jobs traversing link $l$

26: $J^{l}=\\{\\}$

27: for all neighbors $j$ of $l$ do

28: $J^{l}=J^{l}\cup j$

29: $\triangleright$ Solve Cassini optimization (Table 1)

30: $score_{l}$ = CassiniOptimization($J^{l}$)

31: $score_{c}=score_{c}\cup score_{l}$

32: $\triangleright$ Set the compatibility score of candidate $c$

33: $c.score=score_{c}$

34: $\triangleright$ Sort placements based on compatibility metric

35: SortCandidates($Candidates$, “$Decreasing$")

36: $top\\_placement=Candidate[0]$

37: $\triangleright$ Find unique time-shifts (Algorithm 1)

38: $\\{t_{j}\\}$ = BFSAffinityGraph($G_{top\\_placement}$)

39: return $\\{t_{j}\\},top\\_placement$

Algorithm 2 Cassini Module’s Pluggable Algorithm

Step 1. Discover placement candidates. In this step, Cassini decouples the
process of finding the number of workers for each job to improve finish-time
fairness from the exact worker placement in the cluster. To do so, instead of
returning the precise job placements at the end of the auction phase, we
configure Themis to return up to $N$ candidate placements. These candidate
placements all achieve the same finish-time fairness, but their worker
placements are different. For instance, consider a case where jobs $j_{1}$ and
$j_{2}$ each place a bid on two additional workers, and they both win, while
job $k_{1}$ is losing one worker, and job $k_{2}$ is losing three. In this
case, there are two ways to distribute workers: $(i)$ $k_{1}$ and $k_{2}$ each
give up one worker to $j_{1}$, and $k_{2}$ gives two workers to $j_{2}$; or
$(ii)$ $k_{1}$ and $k_{2}$ each give up one worker to $j_{2}$, and $k_{2}$
gives two workers to $j_{1}$. Both options are candidate placements. Moreover,
selecting which workers in $k_{1}$ and $k_{2}$ should be reassigned creates
another set of candidate placements. Cassini collects these candidate
placements and feeds them as input to the next step. This process requires
changing only $\approx$300 lines of code in Themis.

Step 2. Find unique time-shifts. This step is listed in Algorithm 2 and
includes Cassini’s key contributions. Cassini first constructs an Affinity
graph $G_{c}$ for each placement candidate $c\in Candidates$ (lines 9-18).
Following Theorem 1, to ensure correctness, we discard placement candidates
with loop(s) in any of their Affinity subgraphs (line 22). Then, Cassini
constructs the unified circles for each job and solves the optimization
formulation in Table 1 for all links in $G_{c}$ to obtain the compatibility
metric for each link in $V_{c}$ (lines 24-31). Given that the placement
candidates are independent of each other, our implementation uses multiple
threads to parallelize this computation. Once the compatibility score of all
candidate placements is determined, Cassini sorts each placement candidate
based on the average compatibility score of its member links to find the top
placement candidate $top\\_placement\in Candidates$ (lines 35-36).111Instead
of averaging, tail or other metrics may also be used. Then, it executes
Algorithm 1 on $top\\_placement$’s Affinity graph $G_{top\\_placement}$ to
obtain unique time-shifts $\\{t_{j}\\},\forall j\in V_{top\\_placement}$ for
jobs that share links with other jobs in this placement (line 38). Finally,
$top\\_placement$ and its corresponding time-shifts are transferred to
Themis’s agent running on the servers (line 39).

Step 3. Apply time-shifts. When a time-shift $t_{j}$ is received by the Themis
agent running job $j$, it delays the start of the next immediate training
iteration by $t_{j}$. However, even though the workers of the same job apply a
unique time-shift, the time-shift values can drift due to noise, stragglers,
and other unpredictable events. Cassini updates the agent on each server to
measure the drift and adjust the time-shifts. Our evaluations show that time-
shift adjustments are rare (§5.7).

##  5 Evaluations

We evaluate Cassini on a 24-server cluster and compare its performance against
other state-of-the-art ML schedulers. First, we describe our evaluation
methodology and setup (§5.1). Then, we compare Cassini’s performance gains
with respect to the state-of-art ML schedulers for a mix of data and model
parallel DNN training jobs (§5.2). Next, we evaluate the impact of data
parallelism (§5.3), model parallelism (§5.4), partial compatibility (§5.5),
and having multiple GPU per server on Cassini’s performance (§5.6). Finally,
we evaluate the frequency of time-shift adjustments and Cassini’s overhead
(§5.7). Cassini’s code will be publicly available.

Figure 8: [Poisson trace] Time series of data parallel and model parallel
jobs.

Figure 9: [Poisson trace] CDF of the iteration times.

###  5.1 Methodology And Setup

Setup. We build a prototype to demonstrate the gains of Cassini in real-world
settings. Our prototype includes 24 ASUS ESC4000A-E10 servers each with one
A100 Nvidia GPU [6] (40 GB of HBM2 memory) and one 50 Gbps Mellanox ConnectX5
NIC. We use RoCEv2 for communication and enable DCB [3] and PFC on these
interfaces to support a lossless fabric for RDMA. The servers run Ubuntu 18.04
LTS. We use PyTorch [31] version 1.8.0, CUDA version 11.1, and NCCL version
2.11.4 in our training framework.

Topology. We use a Tofino switch to construct the logical topology illustrated
in Figure 18 (Appendix B.1) with 13 logical switches and 48 bi-directional
links and 2:1 over-subscription above the ToRs. We use flow table rules that
match on <input port, destination MAC> to forward packets to the correct
output port and physical loopback cables for switch-to-switch links. We use
the default RDMA-based DCQCN congestion control algorithm [68]. ECN is enabled
through WRED with min and max thresholds set to 1000 and 2000 cells. The PFC
skid buffer threshold of each virtual switch is 4000 cells.

DNN Workloads. We experiment with 13 popular DNN models: VGG11 [22], VGG16
[15], VGG19 [28], ResNet50 [23], WideResNet101 [63], BERT [17], RoBERTa [34],
XLM [14], CamemBERT [38], GPT-1 [46], GPT-2 [47], GPT-3 [8], and DLRM [4]. All
Models have an equal probability of occurrence and the training duration time
is randomly selected between 200 - 1,000 iterations. Table 3 (Appendix B.2)
provides details about model configurations and batch sizes used in this
paper.

Parallelization strategy. We use data parallelism to train VGG, ResNet, and
BERT family of models using Pytorch’s DistributedDataParallel framework [33].
This framework distributes the dataset across GPUs and uses RingAllreduce to
update the gradients during each training iteration. We train DLRM and GPT
family of models using a hybrid of data/model parallelism. Following prior
work [59], we use Meta’s opensource codebase for training DLRM [4] where the
embedding tables are partitioned across GPUs, while the rest of the model is
replicated on all GPUs. Finally, we use Microsoft’s DeepSpeed tool [5] to
partition the GPT models across GPUs using hybrid data/model parallelism.

Traces. Following prior work [39, 66, 35, 45], we use three sets of traces in
our evaluations: ($i$) Poisson trace: we use a Poisson distribution for job
arrivals where the job arrival time is determined by the load parameter
defined as the average fraction of GPUs that are serving active jobs in the
cluster. We vary the load between 80% and 100%; ($ii$) dynamic trace: where a
set of DNN training jobs are present in the cluster and a new set of jobs
arrive; ($iii$) snapshot trace: we take several snapshots of the cluster where
all jobs are present at the start of the experiment.

Figure 10: [Dynamic trace] CDF of training iteration times and the number of
ECN marked packets per iteration.

We implement the following schemes in our testbed.

  * •

Themis. We use the default Themis [35] scheduler as one of our baselines. The
bidding period (epoch) is set to 10 mins. Jobs participate in an auction where
they send bid values for different GPU allocations. An assignment of GPU
servers is valid until the period of lease time. When the lease time expires,
the job gives up the server and a new auction is conducted for all the
released servers. When a job arrives, its initial number of requested workers
is randomly selected between 1 to 12 GPUs. As the experiment progresses, the
number of workers is automatically tuned based on Themis’s finish-time-
fairness metric.

  * •

Th+Cassini. Themis augmented with Cassini as described in Section 4.2. In
particular, this scheduler takes up to 10 placement candidates from Themis,
constructs geometric circles and Affinity graphs for each placement to capture
the cluster-level compatibility, solves our optimization formulation to find
time-shifts for jobs that are competing on bandwidth, selects the top
placement candidate based on compatibility ranks, and finally computes a
unique time-shift for jobs. The unique time-shifts and final placement are
given to the Themis agent running on GPUs. Unless otherwise stated, we use 5∘
as the angle discretization precision (Table 1) to compute the time-shifts.

  * •

Pollux. We use Pollux as a second baseline [45]. Pollux considers the system
throughput and statistical efficiency to maximize cluster-wide training
performance. Pollux periodically queries jobs and reassigns GPUs to maximize
the overall goodput of the cluster. Pollux also models migration costs and
avoids frequent job migrations.

  * •

Po+Cassini. We augment Pollux with Cassini using a similar approach described
in Section 4.2 except that Pollux uses overall goodput, instead of finish-
time-fairness, to adjust hyper-parameters during scheduling epochs. Hence, the
number of workers assigned to each job does not always agree with Themis. To
make an apples-with-apples comparison, all Cassini-related parameters in
Po+Cassini and Th+Cassini are the same.

  * •

Ideal. An ideal scheduler that runs each training job on a dedicated cluster.
This scheduler incurs no congestion since the entire cluster is dedicated to
one job and there is no need to take job compatibility into account.

  * •

Random. A random placement scheduler that places workers for each job
randomly. This scheduler has the highest network overhead since it does not
take locality or compatibility into account.

Profiling DNN models. Similar to Themis and Pollux, we profile each DNN using
Pytorch and Infiniband port counters. Our profiling script executes a few
iterations of each job to measure iteration times and collect link utilization
patterns for various batch sizes and numbers of workers. Fine-grained link
utilization data from the port counters enables Cassini to build the geometric
circles and the corresponding bandwidth demands for our optimization
($bw\\_circle_{j}(\alpha)$ in Table 1).

Figure 11: [Dynamic trace, model parallelism] CDF of training iteration times
and the number of ECN marked packets.

###  5.2 Performance Gains

We evaluate Cassini’s performance gains using job arrivals and departures from
our Poisson trace. Figure 9 plots the time series of events in the cluster for
Themis and Th+Cassini. In this experiment, we use a combination of data
parallel and model parallel training jobs. Placement changes are triggered by
job arrivals, departures, and when the lease time of any of the servers
expires. Given the dynamic nature of the trace, the servers are occupied
gradually and their lease times are not synchronized. For instance, at time
$t=72$ mins a data parallel training job for the XLM [14] model arrives at the
cluster and Themis places it such that one of the links is shared with
WideResNet101 [63] without the knowledge that XLM and WideResNet101 are not
compatible jobs. In contrast, Th+Cassini improves the iteration time of XLM by
placing it with compatible jobs. Note that this trace contains different
training instances of the same DNN models where they differ in their hyper-
parameters and number of workers (details in Appendix B.2). For instance, at
time $t=118$ min, a model parallel GPT-2 [47] training job (labeled as
GPT-2-A) arrives at the cluster and without considering the communication
demands, Themis places this job such that it shares a link with another large
GPT-3 [8] model in the cluster. GPT-2-A and GPT-3 models are not compatible
making both training jobs slow down. In contrast, Th+Cassini improves
GPT-2-A’s iteration time by placing it with a compatible GPT-1 model.

Figure 9 plots the CDF of iteration times of all the data points in Figure 9
and shows that compared to Themis, Th+Cassini improves the average and 99th
percentile tail iteration times by 1.4$\times$ and 1.5$\times$, respectively.
We observe similar gains between Po+Cassini and Pollux. The figure also shows
that Th+Cassini achieves similar performance as our Ideal benchmark.

###  5.3 Cassini Reduces Congestion

We next demonstrate the effectiveness of Cassini in reducing the congestion in
the network. We use our dynamic trace to trigger the arrival of DLRM and
ResNet50 to the cluster while the cluster is busy running other jobs. Given
the contrast between the network demand between these two models, this
experiment serves as a stress test to highlight the importance of compatible
job placement on network congestion. In this case, both Pollux and Themis end
up placing DLRM on servers that share network links with other non-compatible
jobs which hurts the iteration times. In contrast, Th+Cassini and Po+Cassini
flip the placements of DLM and ResNet50 to achieve compatibility, thereby
improving the training iteration times, as depicted in Figure 10(a). Compared
to Themis, Th+Cassini improves the average and 99th percentile tail iteration
times by 1.5$\times$ and 2.2$\times$, respectively. Similarly, compared to
Pollux, Po+Cassini improves the average and 99th percentile tail iteration
times by 1.6$\times$ and 2.5$\times$, respectively.

The gains in iteration times are a direct consequence of Cassini’s ability to
reduce network congestion. Figures 10(b) to (d) show the number of ECN-marked
packets per iteration for different models. The figure shows that Th+Cassini
and Po+Cassini consistently maintain a lower number of ECN marks per iteration
across the models. In particular, Figure 10(d) shows that, on average, DLRM is
experiencing 27$\times$ and 33$\times$ more ECN marks in Themis and Pollux,
compared to their Cassini-augmented counterparts.

Figure 12: [Snapshot trace] Link utilization of compatible and partially
compatible snapshots.

###  5.4 Impact of Model Parallelism

To ensure Cassini’s gains are not limited to data parallel jobs, we run a
series of experiments when all jobs in the trace are using model parallelism.
As shown in Section 2.1, model parallel jobs have several Up and Down phases
in each iteration where the duration and bandwidth demand of each phase
depends on the parallelization strategy and hyper-parameters. Similar to the
data parallel case, we use Cassini’s geometric abstraction to capture the
duration and bandwidth demand of Up and Down phases of a series of model
parallel jobs. We then use Cassini’s optimization formulation and Affinity
graph to compute the time-shifts for the jobs sharing the same network links.
We use our dynamic trace to trigger the arrival of multiple GPT and DLRM
models while the cluster is training other model parallel jobs.

Snap-shot ID |  Competing jobs (batch size) |  Th+Cassini |  Themis |  Comp-atibility score |  time-shift (ms)
---|---|---|---|---|---
1 |  WideResNet101 (800) |  138 ms |  205 ms |  1.0 |  0 ms
|  VGG16 (1400) |  148 ms |  199 ms |  |  150 ms
2 |  VGG19 (1400) |  168 ms |  220 ms |  1.0 |  0 ms
|  VGG16 (1700) |  163 ms |  220 ms |  |  158 ms
|  RESNET50 (1600) |  59 ms |  55 ms |  |  46 ms
3 |  VGG19 (1024) |  166 ms |  176 ms |  0.9 |  0 ms
|  VGG16 (1200) |  168 ms |  177 ms |  |  100 ms
4 |  RoBERTa (12) |  164 ms |  210 ms |  0.8 |  0 ms
|  RoBERTa (12) |  180 ms |  208 ms |  |  60 ms
5 |  BERT (8) |  209 ms |  213 ms |  0.6 |  0 ms
|  VGG19 (1400) |  294 ms |  292 ms |  |  42 ms
|  WideResNet101 (800) |  265 ms |  266 ms |  |  191 ms

Table 2: [Snapshot trace] Compatibility score of DNN jobs.

Figure 11(a) shows the CDF of training iteration times. We find that similar
to the data parallel case, Themis ends up placing non-compatible jobs, such as
<GPT-3 and GPT-2> or <GPT-1 and DLRM>, on the same network link, whereas
Th+Cassini places compatible jobs, such as <GPT-1 and GPT-2> or <GPT-3 and
DLRM>, on the same network links. Consequently, Th+Cassini improves the
average and 99th percentile tail iteration times by 1.2$\times$ and
1.6$\times$, respectively. We observe similar gains between Pollux and
Po+Cassini.

Figures 11(b) to (e) depict the number of ECN marked packets per iteration for
the models in this experiment. Depending on the status of congestion,
different models experience different numbers of ECN-marked packets. For
instance, Figure 11(d) shows that compared to Themis, Th+Cassini reduces the
average number of ECN marked packets by 29$\times$.

###  5.5 Impact of Partial Compatibility

An important consideration for practical deployment of Cassini is to evaluate
the impact of placing partially compatible jobs on the same link(s).
Intuitively, the higher the compatibility score, the better interleaving is
achieved. As the compatibility score reduces, the gains diminish. To evaluate
the impact of partial compatibility, we take five snapshots of the cluster, as
shown in Table 2 and compute the compatibility scores and time-shift values
from our optimization formulation (§3) for each snapshot. We then measure the
average communication time of each model under Themis and Th+Cassini. The
table shows that when the compatibility score is 0.6, Cassini’s gain compared
to Themis starts to diminish. Note that Cassini avoids placing jobs with low
compatibility score (e.g., snapshot 5) on the same link.

We demonstrate the reason behind diminishing returns by plotting the link
utilization of each snapshot in Figure 12. When the compatibility score is
high, the opportunity for interleaving is large and jobs end up interleaving
their network usage most of the time, as shown in Figures 12(a)–(d). However,
as the compatibility score is reduced, jobs are forced to share the link most
of the time, as shown in Figure 12(e). Additionally, Figure 12(b) demonstrates
a desirable feature of our optimization formulation where compatibility does
not require strict interleaving. In this snapshot, only VGG19 and VGG16 are
interleaved and ResNet’s communications overlap with both jobs because its
network demand is not significant and can co-exist with the other jobs.

Figure 13: [Dynamic trace] Multi-GPU experiment.

###  5.6 Impact of Multiple GPUs per Server

Although having multiple GPUs per server enables allocating more GPUs within
the same server to a job, today’s large-scale training jobs require hundreds
of workers [59, 40], making it impossible to avoid network congestion entirely
by relying on multi-GPU servers. In such cases, Cassini’s gains are more
pronounced for models that are distributed outside the boundary of a server.

We evaluate Cassini’s gains with multi-GPU servers by removing GPUs from some
of our single-GPU servers and adding them to other servers to compose servers
with two GPUs. We create a topology with six servers, each with two GPUs, as
shown in Figure 13(a). We then use a mix of data parallel and model parallel
jobs and generate a series of job arrivals using our dynamic trace.

Figure 13(b) demonstrates that compared to Themis, Th+Cassini improves the
average and 99th percentile tail iteration times by 1.4$\times$ and
1.9$\times$, respectively. These gains are achieved because some jobs require
more than two GPUs to train. For instance, at a particular instant in our
dynamic trace, the XLM and ResNet50 models each require three GPUs to train.
With the arrival of a network-intensive model DLRM requesting three more GPUs,
Themis decides to place DLRM such that it shares a server with a non-
compatible model (XLM) making both jobs experience congestion. In contrast,
Th+Cassini selects a placement where DLRM shares a link with a compatible
model (ResNet50), thereby improving the training iteration times of both
models.

###  5.7 Adjusting Time-Shifts and Overhead

Figure 14: [Snapshot trace] The frequency of adjusting time-shifts for
snapshots 1–3.

Figure 15: Impact of angle discretization on execution time and time-shift
accuracy.

To maintain Cassini’s interleaving, workers must respect the time-shift values
given to them through the scheduler. Given that our servers are not running
perfectly in sync, we evaluate the frequency of automatic time-shift
adjustments by the Themis (or Pollux) agents running on the servers. Note that
respecting the time-shift is only required for compatible jobs. All other jobs
in the cluster can send packets at any time. A worker triggers an adjustment
when the start of the communication phase deviates by more than five percent
of the ideal iteration time. Figure 15 shows the average frequency of time-
shift adjustments for snapshots 1,2, and 3. As shown, in all cases, the
frequency is less than two adjustments per minute.

Finally, we evaluate the impact of angle discretization precision on Cassini’s
optimization formulation (Table 1). Intuitively, the execution time of a
coarse-grained discretization is fast but such formulation misses interleaving
opportunities, thereby finding imprecise rotation angles. Given that Cassini’s
time-shifts are driven from rotation angles, a coarse-grained formulation
leads to inaccurate time-shifts. On the other hand, having fine-grained
precision leads to more accurate time-shifts at the expense of a longer
execution time. Figure 15 demonstrates this trend and shows that using a
precision of 5∘ is the sweet spot for achieving 100% accuracy for time-shifts
while maintaining a low execution overhead.

##  6 Related Work

Our work builds on several lines of related research.

Compute scheduling approaches. A large number of systems and techniques have
focused on improving the performance of large-scale distributed ML workloads
[16, 13, 10, 19, 31, 60, 64, 20, 56, 37, 49, 62]. Relevant to this paper,
several papers aim to reduce communication overhead using smart scheduling
techniques; e.g., Gandiva [61], Themis [35], Pollux [45], Tiresias [21],
Shockwave [67], and Optimus [43]. These schedulers try to minimize network
sharing by placing workers of the same job as close as possible to each other.
However, these approaches do not consider interleaving the communication
patterns of different training jobs when placing them on servers. Cassini’s
contribution is complementary to these approaches by considering both the
compute resources and the communication demands of different jobs during
scheduling. Moreover, Cassini is designed as a pluggable module to augment
these schedulers.

Figure 16: Cassini’s geometric abstraction.

Figure 17: Rotating the circles enables interleaving the network demand of
$j_{1}$ and $j_{2}$.

Multi-resource sharing. Recently, Muri [66] proposed a scheduling technique to
interleave critical resources (e.g., GPU, CPU, network, storage) of DNN
training jobs. Muri packs jobs that are being executed on the same set of
resources into a group and interleaves their resource requirements using a
Blossom-based scheduler. However, Muri’s approach to resource interleaving
only applies to jobs that share the same set of GPUs, CPUs, memory, and
network resources.222Muri [66] states this limitation: “The algorithm avoids
cross-group packing to minimize the packing overhead.” Hence, Muri can
interleave compute and communication phases of a set of jobs only if the jobs
are sharing the same set of GPUs. In contrast, Cassini is able to interleave
compute and communication phases of different jobs, irrespective of which GPUs
they occupy. For instance, Muri’s algorithm is not applicable to interleave
the resources of $j_{1}$ and $j_{2}$ in Figure 2(a) because $j_{1}$ is
distributed between server1 and server2 while $j_{2}$ is distributed between
server3 and server4; i.e., these two jobs do not belong to the same resource
group in Muri’s algorithm. Muri would have been able to interleave the
resources if both $j_{1}$ and $j_{2}$ were distributed between all four
servers. However, for many of the large models we use in our experiments, GPU-
sharing is not possible because of the memory requirements of the model.
Moreover, even with GPU sharing, in a large-scale cluster, cross-group network
congestion is common. Cassini is able to interleave the Up and Down phases of
different jobs, without requiring them to share the same set of resources.
Similarly, Synergy [39] proposed a multi-resource interleaving scheduling
approach by inferring the sensitivity of DNN jobs to GPU, CPU, and memory
resources using optimistic profiling. Synergy improves the overall cluster
efficiency by performing resource-sensitive allocations instead of a GPU-
proportional allocation. However, Synergy’s approach does not consider the
network bandwidth as a resource and is unable to interleave the communication
phases with other resources. In contrast, Cassini’s focus is on interleaving
the network demand with GPU resources. Cassini is designed to augment both
Muri and Synergy schedulers.

Communication-aware scheduling. A variety of approaches have been developed to
accelerate communication among ML training workers of the same job to reduce
network overhead [50, 65, 19, 2, 11, 58, 40, 29] and to enable more efficient
pipelining strategies [24, 42]. ByteScheduler [44] and Syndicate [36]
accelerate ML training by scheduling and optimizing the order of communication
operations between different GPU servers used by a training job. ByteScheduler
overlaps compute and communication operations within a training job, while
Syndicate provides a solution for planning and scheduling communication
operations for large DNN training. Similarly, TACCL [51], BytePS [26], and
CLOPT [65] improve the communication collective algorithms of large DNN
models. BytePS seeks to find a balance between the Parameter Server [32] and
Ring-AllReduce algorithms for synchronizing the gradients. TACCL proposes a
communication collective algorithm for training large models with data and
model parallelism. CLOPT co-optimizes network topology and communication
schedules for ML training. These approaches optimize communication within a
training job, however, they do not consider congestion and network sharing
across training jobs. In contrast, Cassini’s approach is orthogonal to these
techniques because Cassini focuses on sharing the network resources across
different training jobs.

Difference with prior workshop paper. A prior workshop paper [48] introduced
the idea of using a geometric abstraction to achieve job compatibility at a
single-link level. We extend this workshop paper in a few important ways.
First, the workshop paper considers compute/communication interleaving at a
high level and does not provide a concrete scheduling technique to achieve it.
Specifically, it relies on an unfair congestion control protocol to achieve
interleaving but Cassini does not require any changes to or assumptions about
the congestion control protocol. Second, the workshop paper ignores the impact
of cluster-level interleaving. Third, the workshop paper only considers the
data parallelism paradigm and its geometric abstraction does not generalize to
model parallelism techniques. Finally, our optimization formulation, the
Affinity graph abstraction, the design and implementation of the Cassini
module, and our formal arguments around correctness (Theorem 1), are all new
contributions.

##  7 Conclusion

Cassini is a simple but effective approach that can integrate with existing
cluster schedulers to allow them to accommodate multiple ML jobs’ network
needs. We introduce a novel metric, called compatibility score, to rank
different GPU placements when jobs compete on network links. Our evaluations
show that Cassini improves the average and tail completion time of jobs by up
to 1.6$\times$ and 2.5$\times$, respectively. Moreover, we show that Cassini
reduces the number of ECN marked packets by up to 33$\times$.

## References

  * [1] MLPerf: A broad ML benchmark suite.  https://mlperf.org/.
  * [2] Baidu, 2017.  https://github.com/baidu-research/baidu-allreduce.
  * [3] Data Center Bridging eXchange (DCBX), 2021.  https://man7.org/linux/man-pages/man8/dcb-dcbx.8.html.
  * [4] Deep Learning Recommendation Model for Personalization and Recommendation Systems, 2021.  https://github.com/facebookresearch/dlrm.
  * [5] DeepSpeed version of NVIDIA’s Megatron-LM, 2021.  https://github.com/microsoft/Megatron-DeepSpeed.
  * [6] NVIDIA A100 Tensor Core GPU, 2021.  https://www.nvidia.com/en-us/data-center/a100/.
  * [7] T. Ben-Nun and T. Hoefler.  Demystifying parallel and distributed deep learning: An in-depth concurrency analysis.  ACM Comput. Surv., 52(4), aug 2019.
  * [8] T. B. Brown, B. Mann, N. Ryder, M. Subbiah, J. Kaplan, P. Dhariwal, A. Neelakantan, P. Shyam, G. Sastry, A. Askell, S. Agarwal, A. Herbert-Voss, G. Krueger, T. Henighan, R. Child, A. Ramesh, D. M. Ziegler, J. Wu, C. Winter, C. Hesse, M. Chen, E. Sigler, M. Litwin, S. Gray, B. Chess, J. Clark, C. Berner, S. McCandlish, A. Radford, I. Sutskever, and D. Amodei.  Language models are few-shot learners.  CoRR, abs/2005.14165, 2020.
  * [9] M. A. Chang, A. Panda, D. Bottini, L. Jian, P. Kumar, and S. Shenker.  Network evolution for dnns.  SysML, 2018.
  * [10] T. Chilimbi, Y. Suzue, J. Apacible, and K. Kalyanaraman.  Project adam: Building an efficient and scalable deep learning training system.  In OSDI’14, pages 571–582, 2014.
  * [11] M. Cho, U. Finkler, D. Kung, and H. Hunter.  Blueconnect: Decomposing all-reduce for deep learning on heterogeneous network hierarchy.  SysML Conference, 2019.
  * [12] E. Chung, J. Fowers, K. Ovtcharov, M. Papamichael, A. Caulfield, T. Massengil, M. Liu, D. Lo, S. Alkalay, and M. Haselman.  Accelerating persistent neural networks at datacenter scale.  In Hot Chips, volume 29, 2017.
  * [13] A. Coates, B. Huval, T. Wang, D. Wu, B. Catanzaro, and N. Andrew.  Deep learning with cots hpc systems.  In International conference on machine learning, pages 1337–1345, 2013.
  * [14] A. CONNEAU and G. Lample.  Cross-lingual language model pretraining.  In H. Wallach, H. Larochelle, A. Beygelzimer, F. d'Alché-Buc, E. Fox, and R. Garnett, editors, Advances in Neural Information Processing Systems, volume 32. Curran Associates, Inc., 2019.
  * [15] A. K. Dash.  VGG-16 Architecture.  https://iq.opengenus.org/vgg16/.
  * [16] J. Dean, G. Corrado, R. Monga, K. Chen, M. Devin, M. Mao, A. Senior, P. Tucker, K. Yang, Q. V. Le, et al.  Large scale distributed deep networks.  In Advances in neural information processing systems, pages 1223–1231, 2012.
  * [17] J. Devlin, M. Chang, K. Lee, and K. Toutanova.  BERT: pre-training of deep bidirectional transformers for language understanding.  CoRR, abs/1810.04805, 2018.
  * [18] A. Gholami, A. Azad, K. Keutzer, and A. Buluç.  Integrated model and data parallelism in training neural networks.  CoRR, abs/1712.04432, 2017.
  * [19] P. Goyal, P. Dollár, 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.
  * [20] P. Goyal, P. Dollár, R. B. Girshick, P. Noordhuis, L. Wesolowski, A. Kyrola, A. Tulloch, Y. Jia, and K. He.  Accurate, large minibatch SGD: training imagenet in 1 hour.  CoRR, abs/1706.02677, 2017.
  * [21] J. Gu, M. Chowdhury, K. G. Shin, Y. Zhu, M. Jeon, J. Qian, H. Liu, and C. Guo.  Tiresias: A GPU cluster manager for distributed deep learning.  In 16th USENIX Symposium on Networked Systems Design and Implementation (NSDI 19), pages 485–500, Boston, MA, Feb. 2019. USENIX Association.
  * [22] S. Gupta.  VGG-11 Architecture.  https://iq.opengenus.org/vgg-11/.
  * [23] K. He, X. Zhang, S. Ren, and J. Sun.  Deep residual learning for image recognition.  In Proceedings of the IEEE conference on computer vision and pattern recognition, pages 770–778, 2016.
  * [24] Y. Huang, Y. Cheng, D. Chen, H. Lee, J. Ngiam, Q. V. Le, and Z. Chen.  Gpipe: Efficient training of giant neural networks using pipeline parallelism.  NeurIPS, 2019.
  * [25] Z. Jia, S. Lin, C. R. Qi, and A. Aiken.  Exploring hidden dimensions in accelerating convolutional neural networks.  In J. Dy and A. Krause, editors, Proceedings of the 35th International Conference on Machine Learning, volume 80 of Proceedings of Machine Learning Research, pages 2274–2283, Stockholmsmässan, Stockholm Sweden, 10–15 Jul 2018. PMLR.
  * [26] Y. Jiang, Y. Zhu, C. Lan, B. Yi, Y. Cui, and C. Guo.  A unified architecture for accelerating distributed DNN training in heterogeneous gpu/cpu clusters.  In 14th USENIX Symposium on Operating Systems Design and Implementation (OSDI 20), pages 463–479. USENIX Association, Nov. 2020.
  * [27] C. Karakus, R. Huilgol, F. Wu, A. Subramanian, C. Daniel, D. Cavdar, T. Xu, H. Chen, A. Rahnama, and L. Quintela.  Amazon sagemaker model parallelism: A general and flexible framework for large model training, 2021.
  * [28] A. Kaushik.  VGG-19 Architecture.  https://iq.opengenus.org/vgg19-architecture/.
  * [29] M. Khani, M. Ghobadi, M. Alizadeh, Z. Zhu, M. Glick, K. Bergman, A. Vahdat, B. Klenk, and E. Ebrahimi.  Sip-ml: High-bandwidth optical network interconnects for machine learning training.  In Proceedings of the 2021 ACM SIGCOMM 2021 Conference, SIGCOMM ’21, pages 657–675, New York, NY, USA, 2021. Association for Computing Machinery.
  * [30] S. Lee, J. K. Kim, X. Zheng, Q. Ho, G. A. Gibson, and E. P. Xing.  On model parallelization and scheduling strategies for distributed machine learning.  In Z. Ghahramani, M. Welling, C. Cortes, N. Lawrence, and K. Q. Weinberger, editors, Advances in Neural Information Processing Systems, volume 27, pages 2834–2842. Curran Associates, Inc., 2014.
  * [31] A. Lerer, L. Wu, J. Shen, T. Lacroix, L. Wehrstedt, A. Bose, and A. Peysakhovich.  Pytorch-biggraph: A large-scale graph embedding system.  CoRR, abs/1903.12287, 2019.
  * [32] M. Li, D. G. Andersen, J. W. Park, A. J. Smola, A. Ahmed, V. Josifovski, J. Long, E. J. Shekita, and B.-Y. Su.  Scaling distributed machine learning with the parameter server.  OSDI’14, pages 583–598. USENIX Association, 2014.
  * [33] S. Li, Y. Zhao, R. Varma, O. Salpekar, P. Noordhuis, T. Li, A. Paszke, J. Smith, B. Vaughan, P. Damania, and S. Chintala.  Pytorch distributed: Experiences on accelerating data parallel training.  Proc. VLDB Endow., 13(12):3005–3018, aug 2020.
  * [34] Y. Liu, M. Ott, N. Goyal, J. Du, M. Joshi, D. Chen, O. Levy, M. Lewis, L. Zettlemoyer, and V. Stoyanov.  Roberta: A robustly optimized bert pretraining approach, 2019.
  * [35] K. Mahajan, A. Balasubramanian, A. Singhvi, S. Venkataraman, A. Akella, A. Phanishayee, and S. Chawla.  Themis: Fair and efficient GPU cluster scheduling.  In 17th USENIX Symposium on Networked Systems Design and Implementation (NSDI 20), pages 289–304, Santa Clara, CA, Feb. 2020. USENIX Association.
  * [36] K. Mahajan, C.-H. Chu, S. Sridharan, and A. Akella.  Better together: Jointly optimizing ML collective scheduling and execution planning using SYNDICATE.  In 20th USENIX Symposium on Networked Systems Design and Implementation (NSDI 23), pages 809–824, Boston, MA, Apr. 2023. USENIX Association.
  * [37] L. Mai, C. Hong, and P. Costa.  Optimizing network performance in distributed machine learning.  In 7th USENIX Workshop on Hot Topics in Cloud Computing (HotCloud 15), Santa Clara, CA, 2015. USENIX Association.
  * [38] L. Martin, B. Müller, P. J. O. Suárez, Y. Dupont, L. Romary, É. V. de la Clergerie, D. Seddah, and B. Sagot.  Camembert: a tasty french language model.  CoRR, abs/1911.03894, 2019.
  * [39] J. Mohan, A. Phanishayee, J. J. Kulkarni, and V. Chidambaram.  Looking beyond gpus for dnn scheduling on multi-tenant clusters.  In USENIX Symposium on Operating Systems Design and Implementation (OSDI 2022), July 2022.
  * [40] D. Mudigere, Y. Hao, J. Huang, Z. Jia, A. Tulloch, S. Sridharan, X. Liu, M. Ozdal, J. Nie, J. Park, L. Luo, J. A. Yang, L. Gao, D. Ivchenko, A. Basant, Y. Hu, J. Yang, E. K. Ardestani, X. Wang, R. Komuravelli, C.-H. Chu, S. Yilmaz, H. Li, J. Qian, Z. Feng, Y. Ma, J. Yang, E. Wen, H. Li, L. Yang, C. Sun, W. Zhao, D. Melts, K. Dhulipala, K. Kishore, T. Graf, A. Eisenman, K. K. Matam, A. Gangidi, G. J. Chen, M. Krishnan, A. Nayak, K. Nair, B. Muthiah, M. khorashadi, P. Bhattacharya, P. Lapukhov, M. Naumov, L. Qiao, M. Smelyanskiy, B. Jia, and V. Rao.  Software-hardware co-design for fast and scalable training of deep learning recommendation models, 2021.
  * [41] S. Narasimhan.  NVIDIA Clocks World’s Fastest BERT Training Time and Largest Transformer Based Model, Paving Path For Advanced Conversational AI, Aug. 2019\.  https://devblogs.nvidia.com/training-bert-with-gpus/.
  * [42] D. Narayanan, A. Harlap, A. Phanishayee, V. Seshadri, N. R. Devanur, G. R. Ganger, P. B. Gibbons, and M. Zaharia.  Pipedream: Generalized pipeline parallelism for dnn training.  In Proceedings of the 27th ACM Symposium on Operating Systems Principles, SOSP’19, pages 1–15, New York, NY, USA, 2019. Association for Computing Machinery.
  * [43] Y. Peng, Y. Bao, Y. Chen, C. Wu, and C. Guo.  Optimus: An efficient dynamic resource scheduler for deep learning clusters.  In Proceedings of the Thirteenth EuroSys Conference, EuroSys ’18, New York, NY, USA, 2018. Association for Computing Machinery.
  * [44] Y. Peng, Y. Zhu, Y. Chen, Y. Bao, B. Yi, C. Lan, C. Wu, and C. Guo.  A generic communication scheduler for distributed dnn training acceleration.  In Proceedings of the 27th ACM Symposium on Operating Systems Principles, SOSP ’19, page 16–29, New York, NY, USA, 2019. Association for Computing Machinery.
  * [45] A. Qiao, S. K. Choe, S. J. Subramanya, W. Neiswanger, Q. Ho, H. Zhang, G. R. Ganger, and E. P. Xing.  Pollux: Co-adaptive cluster scheduling for goodput-optimized deep learning.  In 15th USENIX Symposium on Operating Systems Design and Implementation (OSDI 21), pages 1–18. USENIX Association, July 2021.
  * [46] A. Radford, K. Narasimhan, T. Salimans, and I. Sutskever.  Improving language understanding by generative pre-training.  https://s3-us-west-2.amazonaws.com/openai-assets/research-covers/language-unsupervised/language_understanding_paper.pdf.
  * [47] A. Radford, J. Wu, R. Child, D. Luan, D. Amodei, and I. Sutskever.  Language models are unsupervised multitask learners.  2018\.  https://d4mucfpksywv.cloudfront.net/better-language-models/language-models.pdf.
  * [48] S. Rajasekaran, M. Ghobadi, G. Kumar, and A. Akella.  Congestion Control in Machine Learning Clusters.  In Proceedings of the 21st ACM Workshop on Hot Topics in Networks, HotNets ’22, page 235–242, 2022.
  * [49] A. Sapio, M. Canini, C.-Y. Ho, J. Nelson, P. Kalnis, C. Kim, A. Krishnamurthy, M. Moshref, D. Ports, and P. Richtarik.  Scaling distributed machine learning with In-Network aggregation.  In 18th USENIX Symposium on Networked Systems Design and Implementation (NSDI 21), pages 785–808. USENIX Association, Apr. 2021.
  * [50] A. Sergeev and M. D. Balso.  Horovod: fast and easy distributed deep learning in tensorflow.  CoRR, abs/1802.05799, 2018.
  * [51] A. Shah, V. Chidambaram, M. Cowan, S. Maleki, M. Musuvathi, T. Mytkowicz, J. Nelson, and O. Saarikivi.  TACCL: Guiding collective algorithm synthesis using communication sketches.  In 20th USENIX Symposium on Networked Systems Design and Implementation (NSDI 23), pages 593–612, Boston, MA, Apr. 2023. USENIX Association.
  * [52] C. J. Shallue, J. Lee, J. M. Antognini, J. Sohl-Dickstein, R. Frostig, and G. E. Dahl.  Measuring the effects of data parallelism on neural network training.  CoRR, abs/1811.03600, 2018.
  * [53] N. Shazeer, Y. Cheng, N. Parmar, D. Tran, A. Vaswani, P. Koanantakool, P. Hawkins, H. Lee, M. Hong, C. Young, R. Sepassi, and B. Hechtman.  Mesh-tensorflow: Deep learning for supercomputers.  In S. Bengio, H. Wallach, H. Larochelle, K. Grauman, N. Cesa-Bianchi, and R. Garnett, editors, Advances in Neural Information Processing Systems, volume 31. Curran Associates, Inc., 2018.
  * [54] M. Shoeybi, M. Patwary, R. Puri, P. LeGresley, J. Casper, and B. Catanzaro.  Megatron-lm: Training multi-billion parameter language models using model parallelism, 2020.
  * [55] K. Simonyan and A. Zisserman.  Very deep convolutional networks for large-scale image recognition, 2015\.
  * [56] P. Sun, W. Feng, R. Han, S. Yan, and Y. Wen.  Optimizing network performance for distributed dnn training on gpu clusters: Imagenet/alexnet training in 1.5 minutes.  arXiv preprint arXiv:1902.06855, 2019.
  * [57] B. Wang, Q. Xu, Z. Bian, and Y. You.  Tesseract: Parallelize the tensor parallelism efficiently.  In Proceedings of the 51st International Conference on Parallel Processing. ACM, aug 2022.
  * [58] G. Wang, S. Venkataraman, A. Phanishayee, J. Thelin, N. Devanur, and I. Stoica.  Blink: Fast and generic collectives for distributed ml.  In Conference on Machine Learning and Systems (MLSys 2020), March 2020.
  * [59] W. Wang, M. Khazraee, Z. Zhong, M. Ghobadi, Z. Jia, D. Mudigere, Y. Zhang, and A. Kewitsch.  TopoOpt: Co-optimizing network topology and parallelization strategy for distributed training jobs.  In 20th USENIX Symposium on Networked Systems Design and Implementation (NSDI 23), pages 739–767, Boston, MA, Apr. 2023. USENIX Association.
  * [60] Z. Xia, Y. Zhou, F. Y. Yan, and J. Jiang.  Genet: Automatic curriculum generation for learning adaptation in networking.  In Proceedings of the ACM SIGCOMM 2022 Conference, SIGCOMM ’22, page 397–413, New York, NY, USA, 2022. Association for Computing Machinery.
  * [61] W. Xiao, R. Bhardwaj, R. Ramjee, M. Sivathanu, N. Kwatra, Z. Han, P. Patel, X. Peng, H. Zhao, Q. Zhang, F. Yang, and L. Zhou.  Gandiva: Introspective cluster scheduling for deep learning.  In 13th USENIX Symposium on Operating Systems Design and Implementation (OSDI 18), pages 595–610, Carlsbad, CA, Oct. 2018. USENIX Association.
  * [62] M. Yang, A. Baban, V. Kugel, J. Libby, S. Mackie, S. S. R. Kananda, C.-H. Wu, and M. Ghobadi.  Using trio: Juniper networks’ programmable chipset - for emerging in-network applications.  In Proceedings of the ACM SIGCOMM 2022 Conference, SIGCOMM ’22, page 633–648, New York, NY, USA, 2022. Association for Computing Machinery.
  * [63] S. Zagoruyko and N. Komodakis.  Wide residual networks, 2016.
  * [64] H. Zhao, Z. Han, Z. Yang, Q. Zhang, F. Yang, L. Zhou, M. Yang, F. C. Lau, Y. Wang, Y. Xiong, and B. Wang.  HiveD: Sharing a GPU cluster for deep learning with guarantees.  In 14th USENIX Symposium on Operating Systems Design and Implementation (OSDI 20), pages 515–532. USENIX Association, Nov. 2020.
  * [65] L. Zhao, S. Pal, T. Chugh, W. Wang, P. Basu, J. Khoury, and A. Krishnamurthy.  Optimal direct-connect topologies for collective communications, 2022\.
  * [66] Y. Zhao, Y. Liu, Y. Peng, Y. Zhu, X. Liu, and X. Jin.  Multi-resource interleaving for deep learning training.  In Proceedings of the ACM SIGCOMM 2022 Conference, SIGCOMM ’22, page 428–440, New York, NY, USA, 2022. Association for Computing Machinery.
  * [67] P. Zheng, R. Pan, T. Khan, S. Venkataraman, and A. Akella.  Shockwave: Fair and efficient cluster scheduling for dynamic adaptation in machine learning.  In 20th USENIX Symposium on Networked Systems Design and Implementation (NSDI 23), pages 703–723, Boston, MA, Apr. 2023. USENIX Association.
  * [68] Y. Zhu, H. Eran, D. Firestone, C. Guo, M. Lipshteyn, Y. Liron, J. Padhye, S. Raindel, M. H. Yahia, and M. Zhang.  Congestion control for large-scale rdma deployments.  In Proceedings of the 2015 ACM Conference on Special Interest Group on Data Communication, SIGCOMM ’15, page 523–536, New York, NY, USA, 2015\. Association for Computing Machinery.

##  Appendix A Proof of Theorem 1

This section provides the proof of Theorem 1 (Correctness and Uniqueness
Guarantee) in Section 4. To prove uniqueness we need to show that Algorithm 1
assigns a time shift value exactly once to every job $j\in J$ in the cluster
with Affinity graph $G=(U,V,E)$. To prove the correctness, we need to show
that:

$\displaystyle\forall l,\forall(j_{n},j_{m})\in\\{(j_{i},j_{k})$
$\displaystyle|(j_{i},l)\in E\text{ and }(j_{k},l)\in E\\}:$
$\displaystyle(t_{j_{n}}-t_{j_{m}})\%p^{l}=$
$\displaystyle(t^{l}_{j_{n}}-t^{l}_{j_{m}})\%p^{l}$ (6)

where $p^{l}$ is the perimeter of the geometric abstraction for link $l$. In
other words, to guarantee correctness, we prove that for every pair of jobs
sharing a link, the difference in the time-shift values assigned by the
algorithm is equal to the relative time-shift given by Cassini’s optimization
formulation for that link.

We first use induction to prove that both the above statements are true for
any connected and loop-free Affinity graph $G=(U_{1},V_{1},E_{1})$, and later
we extend this to a general Affinity graph with many connected sub-graphs.

Base case: First, we show that both statements hold for a graph $G$ with only
one link $l$. Algorithm 1 first selects one of the jobs $j_{1}$ connected to
the link $l$ and sets $t_{j_{1}}=0$. Using its BFS traversal algorithm for all
the other jobs $j_{i}$ connected to $l$, Algorithm 1 sets the time shift as:

$t_{j_{i}}=-t^{l}_{j_{1}}+t^{l}_{j_{i}}$

As the algorithm uses BFS and visits each job exactly once, the time-shift
value is computed and assigned exactly once for each job. This ensures that
for a given job, there is a unique time-shift value computed by the algorithm.

To show correctness, we need to prove equation 6 for all job pairs connected
to the link $l$. Say $j_{n}$ and $j_{m}$ are two jobs connected to the link
$l$, then:

$(t_{j_{m}}-t_{j_{n}})\%p^{l}=(-t^{l}_{j_{1}}+t^{l}_{j_{m}}-(-t^{l}_{j_{1}}+t^{l}_{j_{n}}))\%p^{l}$
$=(t^{l}_{j_{m}}-t^{l}_{j_{n}})\%p^{l}$

This shows that the time shift assignments are correct for the base case.

Assumption Step: Let us assume that the two statements hold for every
connected and loop-free Affinity graph having $n$ links.

Induction step: We will use the above assumption to prove that the two
statements hold for a connected and loop-free Affinity graph having $n+1$
links. Let $G=(U_{s},V_{s},E_{s})$ be the connected sub-graph with $n$ links,
now we create an affinity graph with $n+1$ links, by adding a new link $l_{n}$
which is already connected to some set of jobs $J=\\{j\\}$. In order to get a
connected and loop-free Affinity graph with $n+1$ links, $l_{n}$ has to be
connected to exactly one job $j_{i}\in U_{s}$. It has to be exactly one
because having an edge with more than one job from the sub-graph $G$ will form
a loop, and not being connected with any of the jobs from the sub-graph $G$
will make the Affinity graph disconnected. Let $j_{i}$ be the job from
subgraph $G$ that is connected to $l_{n}$. Since $j_{i}$ is the only path to
reach the link $l_{n}$ and the jobs $J$ connected to the link, our algorithm 1
will reach link $l_{n}$ through job $j_{i}$. Then from the algorithm, the time
assignments for the jobs in $J$ are given by:

$\forall j_{k}\in J,t_{j_{k}}=t_{j_{i}}-t^{l_{n}}_{j_{i}}+t^{l_{n}}_{j_{k}}$

The uniqueness is guaranteed since BFS visits each job only once. From the
assumption step the correctness constraints for all the links in the subgraph
$G$ are assumed to be valid, we have to only prove equation 6 for the jobs
connected to $l_{n}$.

$\forall(j_{m},j_{n})\in
J,(t_{j_{m}}-t_{j_{n}})\%p^{l}=(t_{j_{i}}-t^{l}_{j_{i}}+t^{l}_{j_{m}}-(t_{j_{i}}-t^{l}_{j_{i}}+t^{l}_{j_{n}}))\%p^{l}$
$=(t^{l}_{j_{m}}-t^{l}_{j_{n}})\%p^{l}$

This shows that both statements hold true for any Affinity graph with $n+1$
links. This concludes the induction proof, hence, Algorithm 1 holds true for
all connected and loop-free Affinity graphs.

Now we extend to an Affinity graph of a cluster having multiple connected sub-
graphs. Since our algorithm solves each connected sub-graph one by one and
assigns a single time-shift value for each job in the sub-graph, uniqueness is
guaranteed. For correctness, since there is no edge connecting jobs and links
from different disjoint sub-graphs there are no constraints across disjoint
graphs that need to be checked for correctness. Hence this concludes the
overall proof.

Figure 18: Logical topology of our testbed.

Example. As an example, traversing the affinity graph in Figure 6(b) results
in the following unique time-shifts for $j_{1}$, $j_{2}$, and $j_{3}$:

$\displaystyle t_{j_{1}}=$ $\displaystyle 0\leavevmode\nobreak\
(\textit{reference point})$ (7) $\displaystyle t_{j_{2}}=$
$\displaystyle(-t_{j_{1}}^{l_{1}}+t_{j_{2}}^{l_{1}})$
$\displaystyle\mod{&\hskip 28.45274pt\textit{iter\\_time}_{j_{2}}}$ (8)
Affinity graph path: $j_{1}\rightarrow$ $l_{1}\rightarrow j_{2}$
$\displaystyle t_{j_{3}}=$
$\displaystyle(-t_{j_{1}}^{l_{1}}+t_{j_{2}}^{l_{1}}-t_{j_{2}}^{l_{2}}+t_{j_{3}}^{l_{2}})$
$\displaystyle\mod{&\hskip 28.45274pt\textit{iter\\_time}_{j_{3}}}$ (9)
Affinity graph path: $j_{1}\rightarrow l_{1}$ $\rightarrow j_{2}\rightarrow
l_{2}\rightarrow j_{3}$

For the correctness of the algorithm, the graph should be loop-free. In
Cassini’s design, we eliminate placement configurations that have loops.
Themis allocates servers using an auction procedure, which involves multiple
jobs in the cluster participating in the auction. This allows multiple
possible placement configurations for the jobs participating in the auction.
Hence it is easy to find many loop-free placement configurations among them.
Similarly, Pollux reallocates resources periodically, involving multiple jobs
and creating many possible placement configurations.

##  Appendix B Testbed Details

###  B.1 Topology

This section describes Cassini’s testbed topology in more detail. Figure 18
illustrates the logical view of our testbed. Our testbed includes ASUS
ESC4000A-E10 servers each with one A100 Nvidia GPU [6] and one Mellanox
ConnectX5 NIC. The NICs are connected to a Tofino switch. The Tofino switch
emulates 13 logical switches and 48 bi-directional links for a 2:1 over-
subscribed topology.

DNN |  Memory requirement (MB) |  Batch size/GPU |  Parallelization strategy |  Type
---|---|---|---|---
VGG11 [22] |  507 |  512-1800 |  Data Parallel |  Vision
VGG16 [15] |  528 |  512-1800 |  Data Parallel |  Vision
VGG19 [28] |  549 |  512-1800 |  Data Parallel |  Vision
WideResNet101 [63] |  243 |  256-1200 |  Data Parallel |  Vision
ResNet50 [23] |  98 |  256-1800 |  Data Parallel |  Vision
BERT [17] |  450 |  8-32 |  Data Parallel |  Language
RoBERTa [34] |  800 |  8-32 |  Data Parallel |  Language
CamemBERT [38] |  266 |  8-32 |  Data Parallel |  Language
XLM [14] |  1116 |  4-32 |  Data Parallel |  Language
GPT1 [46] |  650 - 9000 |  32-80 |  Model Parallel |  Language
GPT2 [47] |  1623- 27000 |  32-80 |  Model Parallel |  Language
GPT3 [8] |  1952- 155000 |  16-48 |  Model Parallel |  Language
DLRM [4] |  890 - 1962 |  16-1024 |  Model Parallel |  Recomm.
Table 3: DNN models used in our experiments.

###  B.2 DNN Models

As mentioned in Section 5.1, we run our experiment with 13 popular DNN models:
VGG11 [22], VGG16 [15], VGG19 [28], ResNet50 [23], WideResNet101 [63], BERT
[17], RoBERTa [34], XLM [14], CamemBERT [38], GPT1 [46], GPT2 [47], GPT3 [8],
and DLRM [4]. Table 3 summarizes the parameters of each model and batch sizes.
Note that the batch sizes are provided as a range because the number of
workers and hyper-parameters change during scheduling epochs. In particular,
in different experiments, we select the batch size according to the hyper-
parameters used in prior work [52, 1, 49, 35, 45]. The memory requirements of
each model reflects the amount of memory each model occupies in the GPU
memory. We adjust the model sizes for different models depending on the
parallelization strategy.

In our experiments in Figure 9, we use different instances of the same model.
We use suffixes on their names to distinguish between the instances, for
example, GPT2-A and GPT2-B are two different training jobs, as shown in the
legend of Figure 9. GPT2-A has a batch size of 24 with a model hidden size of
1536 (as defined by Deepspeed’s codebase [5]), while GPT2-B has a batch size
of 70 with a model hidden size of 1184.

##  Appendix C Number of ECN Marked Packets

Figure 19: Number of ECN marked packets per iteration

Figure 19 plots the number of ECN marked packets per iteration for the models
ResNet and CamemBERT. These measurements are from the experiment of Section
5.3. The ResNet model has relatively lower ECN marks in general than other
models because ResNet has a smaller model size and requires less network
bandwidth for its AllReduce phase.