# GX-Plug: a Middleware for Plugging Accelerators to Distributed Graph
Processing

Kai Zou, Xike Xie∗, Qi Li, Deyu Kong
Data Darkness Lab, University of Science and Technology of China
{slnt, likamo<EMAIL_ADDRESS><EMAIL_ADDRESS>
###### Abstract

Recently, research communities highlight the necessity of formulating a
scalability continuum for large-scale graph processing, which gains the scale-
out benefits from distributed graph systems, and the scale-up benefits from
high-performance accelerators. To this end, we propose a middleware, called
the GX-plug, for the ease of integrating the merits of both. As a middleware,
the GX-plug is versatile in supporting different runtime environments,
computation models, and programming models. More, for improving the middleware
performance, we study a series of techniques, including pipeline shuffle,
synchronization caching and skipping, and workload balancing, for intra-,
inter-, and beyond-iteration optimizations, respectively. Experiments show
that our middleware efficiently plugs accelerators to representative
distributed graph systems, e.g., GraphX and Powergraph, with up-to 20x
acceleration ratio.

###### Index Terms:

Distributed graph systems, Middleware, accelerators

##  I Introduction

Big graph analytics are often with large data volumes, high computation
intensiveness, and diversified applications, such as social networks,
Internets, traffic networks, and biological structures, just to name a few. To
meet the scaling-out challenge [1], an increasing number of distributed graph
systems, including GraphX [2] and PowerGraph [3], are proposed and deployed.
To meet the scaling-up challenge, non-distributed graph systems, such as
Gunrock[4] and ThunderGP[5], incorporate accelerators, including GPUs, multi-
core CPUs, and FPGAs. Beyond merely scaling-out or scaling-up, recent research
spotlights the vision of a scalability continuum [6], where distributed graph
systems and accelerators can be integrated for elastic scaling of big graph
systems deployed in data centers.

The high computational concentration in cloud services makes an appealing case
for accelerating applications on distributed systems. For example, multi-core
processors, like GPUs [7] and multi/many-core CPUs [8, 9], have been deployed
as accelerators in distributed instances of cloud services, such as Amazon
EC2, Google cloud, Microsoft Azure Blob, and HW cloud, for flexibly scaling up
the performance to application demands. Also, Nvidia has announced the plan to
support Spark $3.0$ with GPU acceleration in 2020 [7, 10]. So, it becomes a
natural technology trend for integrating accelerators with distributed graph
systems.

However, it is more challenging for distributed graph systems, because there
exist a large number of system variants [6], due to the diversity and
irregularity of distributed graph processing. They are with different
architectures, runtime environments (Java and C++), and computation
models111BSP (Bulk Synchronous Parallel) is a parallel model that performs
computation in iterative steps, including three steps of computation,
communication, and synchronization. BSP model [11] has been the most
fundamental and popular execution approach on distributed graph systems. GAS
(Gather-Apply-Scatter) model [3] is another basic and widely adopted model for
distributed graph processing, based on BSP [12]. and programming models222It
includes vertex- and edge-centric models[13].. In this work, we propose a
middleware, the GX-Plug, where accelerators can be neatly plugged to
heterogenous distributed graph systems. With such a middleware, users can
economically scale up their graph systems, to avoid the overhead of replanting
to a new accelerator-aided graph systems, and save the efforts of accelerator
accessing and subsequent optimizations.

Figure 1: Middleware Overview (A: Agent, D: Daemon)

A bird’s eye view of the middleware is shown in Figure 1, which follows an
agile framework, called the daemon-agent framework. A daemon is a multi-core
processor, an abstract representation of an accelerator, enabling transparent
integration of accelerators (GPUs or CPUs), to a distributed graph system
(GraphX or PowerGraph), by holding customizable graph programming interfaces.
An agent is resided in a distributed node for bridging upper systems and
daemons, covering data exchanging and daemon life-cycle controlling. A agent
connects one or more daemons, according to the number of accelerators that the
system allocates, for flexible computation resource distribution and workload
balancing. With the daemon-agent framework, the middleware shows flexibility
in supporting different runtime environments, computation models, and
programming models.

For easy accessing to accelerators, daemon provides a graph algorithm
template, based on conventional iterative models, and support transplanting
existing distributed graph algorithms with ease. Agent provides a kit of
interfaces to cooperate with daemons and uppers systems for global graph
computation. Accordingly, it takes only a few lines of code to plug
accelerators to upper systems.

Nevertheless, there arise a series of research challenges for implementing the
middleware, besides the software design and development efforts on the
adaption to different runtimes (Java or C++) and different accelerators (GPUs
or CPUs). First, there exists considerable data transmission overhead for the
middleware in delivering and translating the data payloads into desired
formats, between upper systems and accelerators, causing _intra-iteration_
overhead. Second, the irregular and complex graph structure incurs imbalanced
workloads, as well as latencies in frequent global synchronization, causing
_inter-iteration_ overhead. Third, it is difficult to schedule the workload
and computation resource for different tasks and system configurations,
recognized as beyond-iteration overhead. The overheads can much degrade the
middleware performance.

We tackle the first challenge by incorporating pipeline shuffle for optimizing
the data transferring between daemons and agents. We tackle the second
challenge by optimizing the process of data synchronization, including caching
and skipping, so as to minimize the volumes of data transferring during the
synchronization phase. We tackle the third challenge by making the size of
transferred data blocks self-adaptive to the workloads of distribute nodes,
and therefore the system workload balancing can be improved.

In this work, we focus on the implementation and optimization of the
middleware. We are aware of optimization techniques, either on the accelerator
end, e.g., exploring memory hierarchies [14, 15] for accelerating on-chip data
accessing or reinforcing local GPU processing networks with NVLink and
NVSwitch[16, 17]; or on the upper system end, e.g., using RDMA[18] for faster
distributed system communication and using pull-push model[19] for data
transferring optimization in specific applications. We would like to argue
that optimizations merely on upper system or accelerator end are beyond the
scope of the middleware, and are orthogonal to our work.

Our contributions can be listed as follows.

  * •

We propose, to our best knowledge, the first middleware for arming distributed
graph systems with high-performance accelerators, to meet the needs of
scaling-out and -up in big graph analytics.

  * •

For the middleware, we design a novel daemon-agent framework, which achieves
flexible deployment on different upper systems and easy accessing to
accelerators.

  * •

The middleware is general in supporting different computation models, such as
BSP and GAS. Existing distributed graph algorithms can be transplanted for
accessing accelerators with ease.

  * •

For the middleware optimization, we investigate a series of techniques, such
as pipeline shuffling, synchronization caching and skipping, and workload
balancing, for intra-, inter-, and beyond-iteration optimizations,
respectively.

  * •

We conduct extensive experiments on real datasets to evaluate the efficiency
and scalability of the middleware.

The rest of the paper is organized as follows. Section II shows the overview
of the middleware. Section III investigates optimization techniques used to
improve the internal performance of middleware. Section IV discusses
middleware deployment techniques. Section V reports the results of empirical
studies. Section VI presents related works. Section VII concludes the paper.
Our middleware is open sourced333https://thoh-testarossa.github.io/GX-Plug/.

##  II Middleware Overview

Cloud services are witnessed to evolve from cloud storage services, comprised
of a multitude of distributed nodes/machines/instances, to high-performance
cloud computing services, comprised of accelerator-powered distributed nodes,
as aforementioned. Our middleware is to boost graph computing on such cloud
services, supporting system configuration and application development with
ease. In this section, we investigate the daemon-agent framework, which is the
core of the middleware, in Section II-A. Then, we study the data storage and
the controllers of the middleware, in Sections II-B and II-C, respectively.

###  II-A Daemon-Agent Framework

The structures of daemons and agents, and their interactions are shown in
Figure 2. In general, daemons are in connection with accelerators, and agents
are in connection with upper systems. The communication of the two parts is
done via the System V IPC.

Figure 2: Daemon-Agent Framework

####  II-A1 Daemon

A daemon represent an accelerator, where graph algorithms are executed. A
daemon thus holds an algorithm template and the iteration logic controlling,
as shown in Figure 2. The design of a daemon is towards transparent hardware
management for upper systems. At the runtime, an instance of the algorithm
template is implemented for daemons. For accelerating distributed graph
algorithms, algorithm engineers only focus on the implementation of the APIs
of the algorithm template. The connection with accelerators are established
during the initialization phase, and details are hidden to system developers
after that.

####  II-A2 Agent

An agent represent a distributed node of an upper system and makes a bridge
for upper systems and daemons. Essentially, an agent covers a set of operation
interfaces between upper systems and daemons, on data exchanging, subfunction
execution, and daemon lifecycle controlling. With the operation interfaces on
agents, upper systems can substantially configure and control daemons,
including specifying the number of accelerators and mixing and matching
different types of accelerators in a system. The structure of an agent is
shown in Figure 2.

In the local environment of a distributed node, there should be at least one
agent, and one or more daemons, representing different accelerators. Also, as
shown in Section III, agents are equipped with a series of optimization
techniques to reduce the overhead caused by data transferring, which is the
major source affecting system performances.

###  II-B Data Flows & Management

The challenges in the data management of the middleware are two-fold: 1) upper
systems and accelerators can be of different runtime environments (C++ and
Java); 2) data in different upper systems may follow different (vertex- or
edge-centric) storage strategies. More, the data transferring should be
efficient in order to meet the system runtime requirements.

The data flow in the middleware is shown in Figure 3. To tackle the efficiency
challenge, the graph data is neither stored in the agent side, nor in the
daemon side. Instead, data is stored in the shared memory space based on the
System V IPC.

Figure 3: Data Flows & Management

Initially, the graph data are partitioned to distributed nodes by upper
systems. Then, for each distributed node, the data are fed to daemons for
acceleration via agents. However, data accessed by an agent cannot be directly
accessed by a daemon, since they belong to different processes with no common
memory space, as discussed in Section IV-C. More, conventional inter-process
communication incurs extra data transferring, degrading the system runtime
performance. In our implementation, we use kernel functions aided by the UNIX
System V to create a shared memory space for daemons and agents. In our
middleware, a daemon has a unique System V key pointing to its specific shared
memory space, while an agent has multiple keys to communicate with all daemons
attached to it.

The benefits are on three aspects. First, the common shared memory space
enables the mutual data accessing between upper systems and accelerators that
are of different environments. Second, data accessing can be done via the
common memory space, avoiding the intermediate data copying between the two
ends. Third, any data updates in the agent or daemon end can be immediately
perceived by the other end without extra sensing efforts, thus facilitating
the control logic of the system.

Based on the common memory space, we proceed to discuss the data management in
the middleware. A agent uses a vertex table and an edge table to manage the
graph data of a distributed node. The structure is also general in supporting
different (vertex- or edge-centric) storage strategies in upper systems. For
daemons, the edge-centric strategy is adopted, because it is commonly accepted
to be more effective in workload balancing, than its counterpart [20, 3],
especially for real graphs following power-law distributions [21, 22]. For
efficient processing in accelerators, a daemon uses a series of data blocks,
including vertex blocks and edge blocks, to be fed to accelerators. Each edge
block contains a fixed number of edges. Also, each edge block is associated
with a paired vertex block, where both source and destination vertices of an
edge can be found. There is a vertex-edge mapping table, for transforming the
data stored in the vertex and edge tables of the agent end to the vertex and
edge blocks of the daemon end. Thus, to construct an edge block, an agent
selects a vertex and retrieves its outer edges, with vertex-edge mapping
table. The corresponding vertex block is constituted by incorporating
destination vertices, as well as their attributes, for the edges in the edge
block.

So, at each iteration of computation, the middleware packages up the vertex
and edge blocks for accelerators, by repeatedly selecting vertices or edges
that are needed. After an iteration of computation, updated vertex and edge
blocks are synchronized back to the vertex and edge tables.

###  II-C Controllers

For the middleware, daemons are in charge of orchestrating different
computational components. We introduce two components located in the daemon,
making the middleware adaptable to different computation models and optimizing
the iterative processing for upper systems.

####  II-C1 Runtime Control

The runtime control component is for controlling the execution order of
implemented APIs, including the runtime information collected from
accelerators and sending/receiving flags for iteration controlling. By
controlling the execution order of implemented APIs, the middleware can easily
be integrated into different computation models, as discussed in Section
IV-B2.

####  II-C2 Iteration Control

The iteration control component is for controlling and coordinating the entire
iteration. Since the middleware separates the runtimes of upper systems and
the computation, it is necessary to connect both parties for in-between data
synchronization and the computation processing cycle. Several optimizations,
such as pipeline shuffling, synchronization caching, and skipping, are
implemented in Section III. The component collaboratively works with agents
for retrieving information from upper systems, and works with the runtime
control component for exchanging flag information, to fulfill the iteration
controlling.

The main goal of the middleware is to shield the system developers and graph
algorithm programmers from the heterogeneity in different systems and
accelerators. It is achieved at the expense of internal overheads in the
middleware. In the sequel, we devise a series of optimizations to alleviate or
even eliminate the overhead originating from the middleware.

##  III Runtime Optimization

In this section, we introduce three optimization techniques, pipeline
shuffling for improving intra-iteration processing, synchronization caching
and skipping for eliminating unnecessary data transferring for inter-iteration
processing, and workload balancing for beyond-iteration optimization.

###  III-A Intra-Iteration Optimization: Pipeline Shuffling

####  III-A1 Motivation

For the basic daemon-agent framework aforementioned, an ordinary workflow of
graph processing acceleration consists of five steps, data downloading (from
upper systems), agent-to-daemon data transferring, computing, daemon-to-agent
data transferring, and data uploading (to upper systems). However, a tightly
coupled execution of the five steps, where the output of one step is streamed
as the input of another step, leads to many waiting-and-suspending states and
therefore the underutilization of computation resources.

For example, the computing step must wait for agent-to-daemon data
transferring to start, so that the computing step would be suspended during
other steps. To alleviate the predicament and to improve the computation
resource utilization, we investigate a pipeline parallelism mechanism,
Pipeline shuffle, to the middleware.

####  III-A2 Overview

Figure 4: Pipeline Shuffle Algorithm 1 Pipeline Shuffle - Daemon Side

Input: Computer Device $com\\_dev$, Data area pointer $n$, $c$, $u$

1:while In Iteration do

2: Block_Recv($agent$, $msg$)

3: if $msg$ = “ExchangeFinished” then

4: Rotate($n$ $\rightarrow$ $c$ $\rightarrow$ $u$ $\rightarrow$ $n$)

5: Send($agent$, “RotateFinished”)

6: else if $c$ contains contents to compute then

7: $com\\_dev$.Load($*c$)

8: $com\\_dev$.Compute()

9: $*c$ $\leftarrow$ $com\\_dev$.data

10: Send($agent$, “ComputeFinished”)

11: else

12: Send($agent$, “ComputeAllFinished”)

13: End_Iteration()

Algorithm 2 Pipeline Shuffle - Agent Side

Input: Upper system interface $USI$, Data area pointer $n$, $c$, $u$

1:$*n$ $\leftarrow$ $USI$.Download()

2:Send($daemon$, “ExchangeFinished”)

3:while In Iteration do

4: Block_Recv($daemon$, $msg$)

5: if $msg$ = “RotateFinished” then

6: $upload$, $download$ = new Thread()

7: for Thread upload do

8: $USI$.Upload($*u$)

9: for Thread download do

10: $*n$ $\leftarrow$ $USI$.Download()

11: else if $msg$ = “ComputeFinished” then

12: if $upload$.isTerminated() then

13: if $download$.isTerminated() then

14: Send($daemon$, “ExchangeFinished”)

15: else if $msg$ = “ComputeAllFinished” then

16: if $upload$.isTerminated() then

17: if $download$.isTerminated() then

18: End_Iteration()

The idea is to construct a multi-layer pipeline for better parallelism. First,
we replace the original $5$-step data transferring into a $3$-step data
transferring, data downloading, computing, and data uploading, which are
handled by $3$ threads, Thread.Download, Thread.Compute and Thread.Upload,
respectively. Compared to the $5$-step setting, the $3$-step setting
eliminates the two steps of agent-to-daemon and daemon-to-agent data
transferring. Second, based on the $3$-step setting, we construct a
$3$-layered pipeline to reduce the suspension time of the computing step, and
finally improve the accelerator utilization. The overview of pipeline shuffle
is shown in Figure 4. The detailed process is depicted by Algorithms 1 and 2.

##### Pipeline Parallelism

The pipeline consists of $3$ layers in correspondence to the $3$ steps
mentioned above, as shown in Figure 5. With the pipeline, an iteration can be
decomposed into a sequence of $3$ processing cycles, corresponding to the $3$
layers. For all three pipeline layers, we use “edge triplets” as the
intermediate data structure, which includes an edge and its source and
destination vertices, by efficiently joining the edge and vertex tables. With
the data structure of edge triplets, the pipeline has homogenous data
structures for all layers, for avoiding unnecessary data format transformation
and enabling grained-granularity data retrieval. Essentially, the triplet is
the basic processing unit of an iteration, which serves as both the source of
computation input and the carrier of output. Within an iteration, there is no
data dependencies between triplets.

For each layer, triplets are grouped into a set of blocks, as shown in Figure
5. The blocks are assigned to the $3$ threads for processing. Thus, pipeline
parallelism can be established which significantly improves the system
performance.

Figure 5: Pipelined Process Flow with Pipeline Shuffle

##### Shuffle for data transfer reduction

There are intermediate data layers existing in an ordinary pipeline. The
$i$-th data layer stores the result of $i$-th pipeline process cycle, and is
used for $(i+1)$-th downstream pipeline process cycle. Within a cycle, data
are transferred between threads for fulfilling the processing. However,
frequent data copying incurs considerable system overhead.

To handle that, we design a shuffle mechanism for efficient data transferring
in pipeline parallelism. First, we allocate equal sized memory chunks for
threads. Each memory chunk is associated with a pointer, for the reference of
the front block currently being processed inside the chunk. Second, inter-
thread data copying is replaced by the pointer copying. For example, in Figure
5, data blocks of $3$ layers can be represented by $n$-block, $c$-block, and
$u$-block, indicating blocks for new data retrieved from upper systems, blocks
for computing, and blocks for uploading to upper systems, respectively. When a
pipeline cycle is finished, the $3$ pointers are shuffled in a rotation
manner: the pointer to $n$-block switches to $c$-block, the pointer to
$c$-block switches to $u$-block, and so on. With the pipeline shuffle
mechanism, there is no more data copying between threads, since it is
completed in situ.

##### Block Size Selection

In our work, we found the size of a block has a profound effect over the
parallelism performance. We assume that there is a sub-dataset distributed to
a agent-daemon pair for processing which contains $d$ entities need to be
processed in the current iteration. Also, agent divides the dataset into $s$
blocks evenly, $b=\frac{d}{s}$. Let $T_{n}(b)$, $T_{c}(b)$, $T_{u}(b)$ be the
process time of one data block in Thread.Download, Thread.Compute and
Thread.Upload, respectively. $T_{n}$ and $T_{u}$ are proportional to the data
block size. We can estimate pipeline processing time $T_{total}$.

$\displaystyle T_{total}=$ $\displaystyle T_{n}(b)+\max(T_{n}(b),T_{c}(b))$
(1) $\displaystyle+$ $\displaystyle(s-2)\max(T_{n}(b),T_{c}(b),T_{u}(b))$
$\displaystyle+$ $\displaystyle\max(T_{c}(b),T_{u}(b))+T_{u}(b)$

$T_{c}$ refers to the time cost of Thread.Compute, and consists of calling
computation devices, loading data to devices, and computation. Their
corresponding time costs are represented by $T_{call}$, $T_{comp}$, and
$T_{copy}$, respectively. The operation of calling devices takes constant
time, while computation and data copying time are related to data size. So,
$T_{c}(b)$ can be modeled as follows.

$\displaystyle T_{c}(b)=T_{call}+T_{comp}(b)+T_{copy}(b)$

When $s$ increases, block size $b$ decreases, so do $T_{n}$ and $T_{u}$.
$T_{c}$ also decreases when $b$ decreases, but will never be less than
$T_{call}$, which means $T_{total}$ starts increasing when $s$ is large enough
and keep increasing. On the other hand, $T_{n}$ and $T_{u}$ increase when $s$
is being smaller, since $b$ is being larger. Thus, both the function
$T_{total}(b)$ and $T_{total}(s)$ should tend to become a U-turn form. Thus,
$s$ and $b$ should be deliberately configured for achieving fine-tuned system
performance. We will try to calculate the value of $b$ in order to provide
optimization suggestion to overall system.

The calculation follows two assumptions: 1) The sub-dataset distributed to the
agent-daemon pair has $d$ entities, and is divided into $s$ blocks evenly,
which have the size $b=\frac{d}{s}$. 2) we assume both $T_{n}$, $T_{u}$ are
directly proportional to the block size; and 3) In $T_{c}$ shown in Equation
III-A2, $T_{call}$ has a fixed number, while $T_{comp}$ and $T_{copy}$ are
also directly proportional to $b$. We use $k_{1}$, $k_{2}$ and $k_{3}$ to
represent download, computation, and upload time cost per unit of data entity.
Thus, Equation 1 can be further simplified as follows:

$\displaystyle T_{total}=k_{1}b+\max(k_{1}b,a+k_{2}b)$ (2)
$\displaystyle+(s-2)\max(k_{1}b,a+k_{2}b,k_{3}b)$
$\displaystyle+\max(a+k_{2}b,k_{3}b)+k_{3}b$

Lemma 1 shows the derivation of optimal $b$ for the pipeline shuffle
mechanism, as follows.

######  Lemma 1.

If a distributed node stores $d$ data entities, the optimal block size
$b_{opt}$ and corresponding $T_{total_{min}}$ can be calculated as follows,
where $Q=\sqrt{\frac{ad}{k_{1}+k_{3}}}$.

$\small b_{opt}=\left\\{\begin{aligned}
\frac{a}{k_{1}-k_{2}}&,&\left(\begin{aligned} k_{max}=k_{1}&,\&\\\
\frac{a}{k_{1}-k_{2}}<Q&\\\ \end{aligned}\right)\\\
\frac{a}{k_{3}-k_{2}}&,&\left(\begin{aligned} k_{max}=k_{3}&,\&\\\
\frac{a}{k_{3}-k_{2}}<Q&\\\ \end{aligned}\right)\\\
\sqrt{\frac{ad}{k_{1}+k_{3}}}&,&otherwise\\\
\end{aligned}\right.\text{~{}~{}~{}~{}~{}~{}, and}$ $\small
T_{total_{min}}=\left\\{\begin{aligned}
\frac{a(k_{1}+k_{3})}{k_{1}-k_{2}}+k_{1}d&,&\left(\begin{aligned}
k_{max}=k_{1}&,\&\\\ \frac{a}{k_{1}-k_{2}}<Q&\\\ \end{aligned}\right)\\\
\frac{a(k_{1}+k_{3})}{k_{3}-k_{2}}+k_{3}d&,&\left(\begin{aligned}
k_{max}=k_{3}&,\&\\\ \frac{a}{k_{3}-k_{2}}<Q&\\\ \end{aligned}\right)\\\
k_{2}d+2\sqrt{(k_{1}+k_{3})ad}&,&otherwise\\\ \end{aligned}\right.$ (3)

###### Proof.

Following Equation 2, we have $3$ cases to consider.

Case 1: $T_{n}=k_{1}b$ is the maximum value. This case is true only if $k_{1}$
is the maximum of the $3$ parameters, $k_{1}$, $k_{2}$, and $k_{3}$.
Accordingly, $b$ should satisfy the follows.

$\displaystyle k_{1}b\geq a+k_{2}b\Rightarrow b\geq\frac{a}{k_{1}-k_{2}}$

Thus, Equation 2 can be transformed into:

$\displaystyle T_{total}=sk_{1}b+\max(a+k_{2}b,k_{3}b)+k_{3}b$
$\displaystyle=k_{1}d+\max(a+k_{2}b,k_{3}b)+k_{3}b$

Notice that $a$ and $\\{k_{i}\\}_{i\leq 3}$ are all positive, and both
$\max(a+k_{2}b,k_{3}b)$ and $k_{3}b$ increase when $b$ increases. Thus, when
$b=\frac{a}{k_{1}-k_{2}}$, we have the minimum value of $T_{total}$ as
follows.

$\displaystyle
T_{total}=k_{1}d+\max(a+\frac{ak_{2}}{k_{1}-k_{2}},\frac{ak_{3}}{k_{1}-k_{2}})+\frac{ak_{3}}{k_{1}-k_{2}}$
$\displaystyle=k_{1}d+\max(\frac{ak_{1}}{k_{1}-k_{2}},\frac{ak_{3}}{k_{1}-k_{2}})+\frac{ak_{3}}{k_{1}-k_{2}}$
$\displaystyle=k_{1}d+\frac{(k_{1}+k_{3})a}{k_{1}-k_{2}}$

Case 2: $T_{c}=(a+k_{2}b)$ is the maximum value. First, we have this equation
below, where $s=\frac{d}{b}$.

$\displaystyle T_{total}=k_{1}b+s(a+k_{2}b)+k_{3}b$
$\displaystyle=(k_{1}+k_{3})b+k_{2}d+\frac{ad}{b}$

In this equation, we can have the minimum $T_{total}$, if $b$ equals
$\sqrt{\frac{ad}{k_{1}+k_{3}}}$, which is $Q$. Notice that $b$ may not equal
$Q$, as constrained by $\\{k_{i}\\}$. Accordingly, we discuss $T_{total}$ in
$3$ subcases, based on the value of $k_{2}$.

($k_{2}$ is the minimum one. ) In this situation, both $(k_{1}-k_{2})$ and
$(k_{3}-k_{2})$ are positive. Thus, we have:

$\displaystyle a+k_{2}b\geq\max(k_{1},k_{3})\cdot b\Rightarrow
b\leq\min(\frac{a}{k_{1}-k_{2}},\frac{a}{k_{3}-k_{2}})$

Assume that $k_{1}\geq k_{3}$, we have $b\leq\frac{a}{k_{1}-k_{2}}$. Thus, we
have the minimum $T_{total}$:

$\small T_{total}=\left\\{\begin{aligned}
k_{2}d+2\sqrt{(k_{1}+k_{3})ad}&,&\frac{a}{k_{1}-k_{2}}\geq Q\\\
\frac{a(k_{1}+k_{3})}{k_{1}-k_{2}}+k_{2}d+(k_{1}-k_{2})d&,&\frac{a}{k_{1}-k_{2}}<Q\\\
\end{aligned}\right.$

Also, we have the minimum $T_{total}$ when $k_{3}\geq k_{1}$:

$\small T_{total}=\left\\{\begin{aligned}
k_{2}d+2\sqrt{(k_{1}+k_{3})ad}&,&\frac{a}{k_{3}-k_{2}}\geq Q\\\
\frac{a(k_{1}+k_{3})}{k_{3}-k_{2}}+k_{2}d+(k_{3}-k_{2})d&,&\frac{a}{k_{3}-k_{2}}<Q\\\
\end{aligned}\right.$

($k_{2}$ is the middle one. ) In this situation, we should notice the change
of the inequality, because some terms of Equation 4 can be negative. Without
loosing generality, we assume $k_{3}\leq k_{2}\leq k_{1}$. In this case,
$(k_{1}-k_{2})$ is positive, and $(k_{3}-k_{2})$ is negative. Thus, we can
have $b\leq\frac{a}{k_{1}-k_{2}}$, since

$\displaystyle\frac{a}{k_{3}-k_{2}}<0\leq
b\leq\frac{a}{k_{1}-k_{2}}\Rightarrow 0\leq b\leq\frac{a}{k_{1}-k_{2}}$ (4)

Then, we have the minimum value of $T_{total}$ as shown in Equation III-A2.

$\small T_{total}=\left\\{\begin{aligned}
\frac{a(k_{1}+k_{3})}{k_{1}-k_{2}}+k_{2}d+(k_{1}-k_{2})d&,&\frac{a}{k_{1}-k_{2}}<Q\\\
k_{2}d+2\sqrt{(k_{1}+k_{3})ad}&,&otherwise\\\ \end{aligned}\right.$

On the other hand, if $k_{3}\geq k_{2}\geq k_{1}$ holds, the minimum value of
$T_{total}$ is:

$\small T_{total}=\left\\{\begin{aligned}
\frac{a(k_{1}+k_{3})}{k_{3}-k_{2}}+k_{2}d+(k_{3}-k_{2})d&,&\frac{a}{k_{3}-k_{2}}<Q\\\
k_{2}d+2\sqrt{(k_{1}+k_{3})ad}&,&otherwise\\\ \end{aligned}\right.$

($k_{2}$ is the maximum one. ) In this situation, both $k_{1}-k_{2}$ and
$k_{3}-k_{2}$ are negative. Since $b>0$, $b$ is also greater than
$\frac{a}{k_{1}-k_{2}}$ and $\frac{a}{k_{3}-k_{2}}$. Thus, we have the minimum
$T_{total}=k_{2}d+2\sqrt{(k_{1}+k_{3})ad}$, when
$b=\sqrt{\frac{ad}{k_{1}+k_{3}}}=Q$.

Case 3: $T_{u}=k_{3}b$ is the maximum value. This case is true only if $k_{3}$
is the maximum of the $3$ parameters in $k_{1}$, $k_{2}$, and $k_{3}$.

Following the discussion in Equation III-A2, we simply have the conclusion
that $T_{total}$ has the minimum value
$k_{3}d+\frac{(k_{1}+k_{3})a}{k_{3}-k_{2}}$ when $b=\frac{a}{k_{3}-k_{2}}$.

Discussion. Following the previous discussion, and the order of $k_{1}$,
$k_{2}$ and $k_{3}$, we have $3$ cases to calculate $b_{opt}$.

Case (i). $k_{1}$ is the maximum one: if
$\frac{a}{k_{1}-k_{2}}\geq\sqrt{\frac{ad}{k_{1}+k_{3}}}=Q$,
$b=\sqrt{\frac{ad}{k_{1}+k_{3}}}=Q$, and $T_{total}$ have the minimum value
$k_{2}d+2\sqrt{(k_{1}+k_{3})ad}$. Otherwise, $b=\frac{a}{k_{1}-k_{2}}$, and
$T_{total}$ have the minimum value
$k_{1}d+\frac{(k_{1}+k_{3})a}{k_{1}-k_{2}}$.

Case (ii). $k_{2}$ is the maximum one: $b=\sqrt{\frac{ad}{k_{1}+k_{3}}}=Q$,
and $T_{total}$ have the minimum value $k_{2}d+2\sqrt{(k_{1}+k_{3})ad}$.

Case (iii). $k_{3}$ is the maximum one: if
$\frac{a}{k_{3}-k_{2}}\geq\sqrt{\frac{ad}{k_{1}+k_{3}}}=Q$,
$b=\sqrt{\frac{ad}{k_{1}+k_{3}}}=Q$, and $T_{total}$ have the minimum value
$k_{2}d+2\sqrt{(k_{1}+k_{3})ad}$. Otherwise, $b=\frac{a}{k_{3}-k_{2}}$, and
$T_{total}$ have the minimum value
$k_{1}d+\frac{(k_{1}+k_{3})a}{k_{3}-k_{2}}$.

Then, Equations in Lemma 1 can be directly derived from the above discussion,
by grouping $b$ and $k$. ∎

Notice that both $s$ and $b$ must be integers. If $b_{opt}$ or
$s_{opt}=\frac{d}{b_{opt}}$ is not an integer, we choose $2$ values $\lfloor
s_{opt}\rfloor$ and $\lceil s_{opt}\rceil$ for $s$, and $2$ values $\lfloor
b_{opt}\rfloor$ and $\lceil b_{opt}\rceil$ for $b$, so that Equation 2 can be
used for estimating the minimum $T_{total}$.

###  III-B Inter-Iteration Optimization: Synchronization Caching and Skipping

####  III-B1 Motivation

For a distributed graph system, there is inevitable data synchronization
between iterations, for ensuring the data correctness in every distributed
node. However, it is often costly to do such synchronizations, since it would
trigger considerable data copying between two successive iterations. Also, in
a naïvely integrated scale-out and -up system, the data copying involves
memory accessing from distributed system environments to external computation
accelerators, which is costly. Thus, it is necessary to reduce the load of
data synchronization, either for the number of times that synchronization
triggers, or for the data volume transferred. To this end, we introduce
techniques of synchronization caching and synchronization skipping for inter-
iteration optimization.

####  III-B2 Synchronization Caching

Figure 6: Synchronization Caching

Figure 6 shows the main process of synchronization caching. The idea is to use
local cache of agents to reduce unnecessary data transferring between daemons
and upper systems. It has two parts, _LRU-based caching_ and _lazy uploading_.

##### LRU-based Caching

Think twice about the data transferring process. At the beginning of an
iteration, an agent downloads data to be computed from upper systems. A vertex
would have to be repeatedly downloaded from upper systems, if it is involved
in the computation iterations, even though its corresponding attributes are
never updated.

To save the overhead, the agent can cache a set of vertices in a temporary
vertex table, and the cache is organized in a least recently used (LRU in
short) manner. Initially, when entering the cache, every vertex has a weight,
whose value decreases with the passage of iterations, and increases if being
used for computation. When the daemon requires a specific vertex for
computation, the agent first checks its local cache for the vertex. If not
found, the agent downloads it from upper systems to cache, and evicts the
vertex with the highest weight. When the agent collects computation results
for updating to upper systems, it first checks if corresponding vertices are
cached. If so, the agent updates the attributes of the vertex, and upgrades
its weight. Otherwise, the agent chooses vertices with the lowest weights, and
replaces them by vertices in the computation result. If the chosen vertices
were updated in previous iterations, corresponding information will be
uploaded to upper systems. The updated vertices in the cache are marked, for
lazy uploading, as discussed below.

##### Lazy Uploading

Also, there is no need for immediately uploading an updated vertex, until it
is involved in the computation of other distributed nodes. For example, if
there are many copies of a vertex generated before the synchronization, only
the vertex copy with optimal updated value needs to be uploaded, meaning that
other vertex copies are obsolete. Thus, to prevent unnecessary uploading, we
make the strategy of “lazy uploading”.

So, we design two queues for the lazy uploading, _global query queue_ and
_global data queue_. After all computation results are updated to the cache,
the agent first constructs a list of vertex IDs which are needed by the
distributed node for the next iteration. Then, all agents push their local
lists to the upper system. The union of local lists formulates the global
query queue and is broadcast to all agents. Each agent compares its cache with
the global query queue, and uploads the required vertices to the global data
queue. This way, data uploading is triggered only if necessary. Algorithm 3
shows the details of lazy uploading.

Algorithm 3 Lazy Uploading

Input: Updated Dataset $s$, Global query queue $gqq$, Global data queue $gdq$

1:$s_{q}$ $\leftarrow$ $s$.GetQueriedEntity()

2:Send($gqq$, $s_{q}$)

3:Wait for other agents

4:$s_{u}$ $\leftarrow$ Find($gqq$, $s$.GetUpdatedEntity())

5:Send($gdq$, $s_{u}$)

6:Wait for other agents and upper system synchronization

7:$s$.Update(Fetch($gdq$, $s_{q}$))

####  III-B3 Synchronization Skipping

Following the characteristics of distributed graph processing, it happens that
some iterations can be skipped, so that the synchronization overheads of these
iterations can be saved. The observation is that, there is no need to trigger
the global synchronization, if there is no de facto “conflicts” among
distributed nodes for being synchronized, i.e., updated data of a node is not
needed by all other nodes. We therefore design a mechanism called
“synchronization skipping” based on synchronization caching, to detect if the
current iteration synchronization can be skipped.

Figure 7: Synchronization Skipping

As shown in Figure 7, at the end of cache updating, an agent checks if each
updated vertex and its outer edges are in the same node. If it is true for an
agent, it means the agent can continue with the next iteration using its local
data. If it is true for all agents, it means that there is no need for any
inter-node data transferring. Thus, the upper system process can be skipped
and next computation iteration can be directly started. This way, multiple
computation iterations can be equivalent to a logically combined iteration,
and therefore unnecessary synchronization for intermediate iterations can be
skipped.

###  III-C Beyond-Iteration Optimization: Workload Balancing

####  III-C1 Motivation

As a “software glue”, the middleware connects different accelerators and
different upper systems. For instance, upper systems may adopt various graph
partitioning strategies for assigning subgraphs to distributed nodes, which
may cause storage imbalance. On the other hand, different accelerators may
have different computation powers, which may cause computation imbalance.
Therefore, it is important for the middleware to have a mechanism to detect
and react to the workload balancing, so that the performance of the
parallelism can be maximized.

####  III-C2 Analysis

To this end, we introduce a simple yet effective workload estimation model for
the middleware to predict the performance of data processing of a local node.

Suppose there are in total $D$ data entities which are partitioned into $m$
distributed nodes, satisfying $\sum_{j=1}^{m}d_{j}=D$. According to pipeline
shuffle mechanism in Section III-A, the total processing time consists of
three parts. For ease of discussion, for distribute node $j$, we set the total
time cost $T^{j}_{total}$ taken by Thread.Download, Thread.Compute, and
Thread.Upload as $T^{j}_{n}$, $T^{j}_{c}$, and $T^{j}_{u}$, respectively,
satisfying $T^{j}_{total}=T^{j}_{n}+T^{j}_{c}+T^{j}_{u}$.

Noticed that both ${T^{j}_{n}}$ and ${T^{j}_{u}}$ are proportional to $d_{j}$.
${T^{j}_{c}}$’s computation time and data copying time are also proportional
to $d_{i}$, and calling time $T^{j}_{call}$ is proportional to the number of
blocks $s$. We can have:

$\displaystyle T^{j}_{total}={T^{j}_{n}}+{T^{j}_{c}}+{T^{j}_{u}}=c_{j}\cdot
d_{j}+s\cdot T^{j}_{call},$

where $c_{j}$ is the coefficient associated with node $j$ to represent the
relation between data size and process time. Since there is no relationship
between $s$ and $d_{j}$, there is no need to consider $s$ in this situation.
Then, given a set of $m$ distributed nodes, the objective of workload
balancing can be represented by:

$\small\min(\max\limits_{j\leq m}(c_{j}\cdot d_{j}))$ (5)

Here, we call $\frac{1}{c_{j}}$ as computation capacity factor, since $c_{j}$
shows “time cost per unit amount of data”, and $\frac{1}{c_{j}}$ shows “data
processed per unit time”.

####  III-C3 Mechanism

With the objective function (Equation 5), we come up with two metrics for the
middleware to detect and react to the workload imbalance. The first one offers
benchmarks for upper systems to adjust partitioning strategies, given a
specific configuration of accelerators for distributed nodes. The second one
offers benchmarks for upper systems to supervise the assignment of
accelerators to distributed nodes, under a specific graph partitioning
strategy. In other words, our estimation model can be applied for $2$ cases.
The first case is on the tuning of $\\{d_{j}\\}$ with fixed $\\{c_{j}\\}$. The
second case is on the tuning of $\\{c_{j}\\}$ with fixed $\\{d_{j}\\}$.

Case 1: tuning $\\{d_{j}\\}$ under fixed $\\{c_{j}\\}$. Lemma 2 makes the
theoretical basis for getting optimal values of $d_{j}$.

######  Lemma 2.

Given $D$ data entities which are partitioned to $m$ distributed nodes, where
each node holds a data fragment $d_{j}$, satisfying $\sum_{j=1...m}d_{j}=D$,
the balancing target is to minimize function $G(.)$, which represents the
maximum time cost of a distributed node.

$\displaystyle G(d_{1},...,d_{m})=\max_{j=1...m}(c_{j}d_{j})$

It holds that function $G(.)$ achieves its minimum value, iff every element
$d_{j}$ of its $m$-dimensional input variable $\\{d_{j}\\}_{j\leq m}$
satisfies:

$\displaystyle d_{j}=\frac{\frac{1}{c_{j}}}{\sum^{m}_{j=1}\frac{1}{c_{j}}}D$

###### Proof.

First, if every $d_{j}$ meets the condition, we have:

$\displaystyle
G(d_{1},...,d_{m})=\max^{m}_{j=1}\\{c_{j}\cdot\frac{\frac{1}{c_{j}}}{\sum^{m}_{j=1}\frac{1}{c_{j}}}D\\}=\frac{D}{\sum^{m}_{j=1}\frac{1}{c_{j}}}$

Second, we prove that for any possible $d_{j}$, we have
$G\geq\frac{D}{\sum^{m}_{j=1}\frac{1}{c_{j}}}$. We prove this assertion by
contradiction. We first assume it holds that
$G=max^{m}_{j=1}(c_{j}d_{j})<\frac{D}{\sum^{m}_{j=1}\frac{1}{c_{j}}}$. Then
for every $d_{j}$, we have:

$\displaystyle
d_{j}<\frac{\frac{1}{c_{j}}}{\sum^{m}_{j=1}\frac{1}{c_{j}}}D\Rightarrow
D=\sum^{m}_{j=1}d_{j}<\frac{\sum^{m}_{j=1}\frac{1}{c_{j}}}{\sum^{m}_{j=1}\frac{1}{c_{j}}}D=D$

Here, contradiction occurs. Thus, function $F$ reaches the minimum value
$\frac{D}{\sum^{m}_{j=1}\frac{1}{c_{j}}}$, if and only if for all $d_{j}$,
$d_{j}=\frac{\frac{1}{c_{j}}}{\sum^{m}_{j=1}\frac{1}{c_{j}}}D$. The lemma is
hence proved. ∎

Lemma 2 shows that $\frac{\frac{1}{c_{j}}}{\sum^{m}_{j=1}\frac{1}{c_{j}}}$ can
serve as balancing factors for selecting appropriate partitioning strategies.
For example, given a set of partitioning strategies, the one that achieves
minimum $F(.)$ is to be selected.

Case 2: tuning $\\{c_{j}\\}$ under fixed $\\{d_{j}\\}$. It is possible for
upper systems to elastically select demanding number of accelerators (e.g.,
from a GPU cloud), given that the graph partitioning results are fixed (graph
partitioning is more I/O and computational intensive than other processing
phases). If so, the middleware can adjust the computation capacity factor
$1/c^{{}^{\prime}}_{j}$ for balancing the workload, according to Lemma 3.

######  Lemma 3.

Given $D$ data entities which are partitioned to $m$ distributed nodes, where
each node holds a data fragment $d_{j}$, satisfying $\sum^{m}_{j=1}d_{j}=D$,
and given the maximum available computation capacity factor $f$
($f\geq\max_{j=1...m}\frac{1}{c_{j}}$), our target is to minimize function
$G^{\prime}(.)$, which indicates the maximum time cost of a distributed node.

$\displaystyle
G^{{}^{\prime}}(\frac{1}{c_{1}},...,\frac{1}{c_{m}})=\max^{m}_{j=1}(c_{j}d_{j})$

Function $G^{\prime}(.)$ achieves its minimum value, if every element
$\frac{1}{c_{j}}$ of its $m$-dimensional input variable
$\\{\frac{1}{c_{j}}\\}_{j\leq m}$ satisfies:

$\small\begin{aligned} \frac{1}{c_{j}}=\frac{f\cdot
d_{j}}{d_{*}}\end{aligned}\textit{,~{}~{}~{}~{}~{}~{}~{}where
$d_{*}=max_{j\leq m}(d_{j})$}$

###### Proof.

Let $\frac{1}{c_{*}}$ be $max_{j\leq m}\frac{1}{c_{j}}$. Since
$\frac{1}{c_{*}}\leq f$, we have:

$\displaystyle\frac{d_{*}}{f}\leq c_{*}d_{*}\leq
F^{{}^{\prime}}=\max^{m}_{j=1}(c_{j}d_{j})$

To make $G^{{}^{\prime}}=\frac{d_{*}}{f}$, all other $c_{j}d_{j}$ must be not
greater than $\frac{d_{*}}{f}$. Thus, to minimize $\frac{1}{c_{j}}$, we have:

$\displaystyle\frac{1}{c_{j}}=min\\{\frac{1}{c_{j}},\textnormal{where
}c_{j}d_{j}\leq\frac{d_{*}}{f}\\}=\frac{fd_{j}}{d_{*}}$

Thus, the lemma is proved. ∎

According to Lemma 3, the middleware can dynamically allocate idle
accelerators to generate more daemons for the node demanding computation
powers, as long as conditions of computation capacity factor of each partition
are met.

##  IV System Implementation

We show details on key implementation of the middleware.

###  IV-A APIs

It is important for the middleware to create a series of easy-to-use
interfaces to make accelerators plugged to upper systems easy and coder-
friendly. In our implementation, we design an iteration-based graph algorithm
template and a set of operations interfaces to connect with upper systems.

####  IV-A1 Algorithm Template

The APIs of algorithm template follow a unified iterative model and support
C++-based code integration, including OpenMP, OpenCL, MPI, and CUDA. There are
$3$ steps with computation of an iteration for general multiworker systems,
Message Passing, Combining and Aggregating, in which external computation
resources can be utilized for computation optimization. In correspondence to
the above $3$ steps, our algorithm template has $3$ APIs, MSGGen(), MSGMerge()
and MSGApply(). MSGGen() function is the computation function for calculating
the initial results with vertex and edge blocks and transforming them into
initial messages. MSGMerge() function delivers the initial messages to
corresponding graph partitions. MSGApply() fetchs message sets for the current
partition, and applies them to corresponding vertices and edges. Accordingly,
one can design a graph algorithm by implementing the 3 interfaces of the
algorithm template. Examples of implementing graph algorithms can be found in
our code repository${}^{\ref{source}}$.

With the help of the daemon-agent framework, runtime details, such as data
transferring, runtime orders, interactions with upper systems, and extra
resource management, are hidden to algorithm engineers , thus they can focus
on the implementation of three APIs for specific graph algorithms.

More, with the separate maintenance of thee functions, upper system developers
can arrange the API calls in different orders, so that the middleware is
adaptable to various graph computation models, such BSP, GAS, and asynchronous
model, as shown in Section IV-B2.

####  IV-A2 Operation Interfaces

To make upper system calls easier to be adapted to agents, the agent accessing
is organized into three functions, including two functions for data
transferring, i.e., transfer() and update(), and function requestX() for
computation lifecycle controlling. Here, $X$ can be any of the three APIs,
MSGGen(), MSGMerge() or MSGApply(). For an upper system, a call sequence of a
computation iteration is: connect() $\rightarrow$ update() $\rightarrow$
{requestX()} $\rightarrow$ update() $\rightarrow$ disconnect(). Upper system
developers only need to access corresponding functions by inputting proper
parameters to get the full control of the daemon runtime. Also, it takes
merely a few lines of code for the agent to connect to upper systems.

In summary, with such interface functions of agents, computation daemons can
be integrated and cooperated for the global computation invoked by upper
systems.

###  IV-B Environment Accessing

####  IV-B1 GraphX (JVM)

JVM is a uniform environment separated from the local environment of
distributed nodes to execute Java programs. However, it makes things
complicated when GraphX needs external tools or libraries for computation. JNI
provided in JVM suffers from additional costs in invoking native target
functions, due to JNI callbacks, which can be eased by the native memory [23].
To solve the problem, we design $2$ components to efficiently break the
barrier between the JVM runtime and local environment.

JNI Transmitter. We use JNI to achieve reflection between GraphX native method
interfaces and external functions connected with agent. However, naively
invoking JVM methods at runtime incurs significant transmission lags. Hence,
we utilize a series of techniques such as POSIX-based shared memory data
exchanging, batch data transferring, in JNI transmitter in order to reduce JNI
calls.

Data Packager. Data packager solves the inconsistency of data structures of
the two ends of JVM and local environment. It uses techniques of bit data
organization and space reserving for data transformation without extra space
usage and redundant data copying. Preliminary testing shows that about $3$ to
$10$ times of improvement can be achieved, compared to direct target function
invoking.

####  IV-B2 PowerGraph

PowerGraph[3] follows another computation model, called Gather-Apply-Scatter
(GAS in short), which is widely used by many distributed graph processing
frameworks. Although BSP and GAS are of different graph computation models,
they are common in basic iterative characteristics [13], which can be view as
different orders of iterative operations, etc. It paves the road for
theoretical and technical foundation of a general middleware design in
supporting different computation models.

For example, when connecting to PowerGraph, MSGGen(), MSGMerge() and
MSGApply() is used to represent scatter, gather, and apply steps in GAS model.
The execution order in PowerGraph thus follows Merge() $\rightarrow$ Apply()
$\rightarrow$ Gen(), which differs from BSP model, i.e., Gen() $\rightarrow$
Merge() $\rightarrow$ Apply(). In an iteration, PowerGraph calls agent
interfaces, in the order of requestMerge(), requestApply() and requestGen(),
so that GAS model can be supported by the middleware without any extra code
modifications. It shows the generality of our middleware in adapting to
different computation models.

###  IV-C Runtime Isolation

If an agent is naïvely designed as a parent process of daemons, the device
environment of accelerators would be re-initialized multiple times during the
iterative graph processing. Because the launching and ceasing of an agent must
be triggered multiple times by the upper system, and so do their associated
daemons. The frequent re-initialization incurs considerable system overheads,
since the initialization process of daemons (with internal function calls) and
associated computational devices is time-consuming. To overcome the dilemma,
our daemon-agent framework detaches the initialization process from direct
function calls. Daemons and agents work as independent processes, and they
communicate with each other by message exchange. This way, a daemon never
triggers re-initialization during the iterative graph processing.

##  V Evaluation

(a) algos @ Twitter2010

(b) algos @ Orkut

(c) algos @ LiveJournal

(d) algos @ Wiki-topcats

Figure 8: Results on Different Datasets

(e) Scalability w.r.t. GPUs. (Orkut)

(f) Results on Twitter & UK-2007

(g) Scalability w.r.t. Algos

(h) Mix & Match (CPU and GPU)

Figure 9: Results on Scalability

Figure 10: Performance of Pipeline Shuffle

(a) Caching on Systems

(b) # of Skipped Iterations

Figure 11: Sync Cacheing & Skipping Performance

(a) Fixed Comp Resource

(b) Fixed Data Partitioning

Figure 12: Workload Balancing Performance

Figure 13: Runtime Isolation

(a) PowerGraph, Orkut

(b) GraphX, Orkut

Figure 14: Middleware Cost Ratio

(a) Estimation vs Real Figure 15: Estimating $s_{opt}$

###  V-A Setup

We conduct experiments on a set of representative graph algorithms, such as
Bellman-Ford (SSSP-BF), PageRank (PR), and Label Propagation algorithm (LP),
by varying datasets of different distributions and scales444For SSSP-BF, we
use 4 vertices as source vertices and calculate their SSSPs simultaneously to
make it more compute-intensive. For LP, we limit the iterations to 15 times to
avoid unlimited computation on specific datasets.. For experiments, we use a
series of $6$ real datasets which are commonly used for graph systems testing,
as shown in Table I. By default, Orkut is used, since it has the highest
vertex degree among the $6$555The workload of a distributed node is
proportional to the number of edges stored in it [20, 3]. Accordingly, a
dataset with high vertex degrees offers high computation workload per unit
amount of data..

TABLE I: Datasets Dataset | Vertex | Edge | Type
---|---|---|---
Orkut[24] | 3.07M | 117.18M | Social
Wiki-topcats[25] | 1.79M | 28.51M | Network
LiveJournal[26] | 4.84M | 68.99M | Social
WRN[27] | 23.9M | 28.9M | Road
Twitter[28, 29] | 41.65M | 1.468B | Social
UK-2007-02[30] | 110.1M | 3.945B | Social

For testing the scalability, we build a GPU cluster with 6 physical nodes,
each of which is equipped with CPU Xeon E5-2698 v4 ($2.20$GHz, $20$ cores) and
$2$ NVIDIA V$100$ GPUs ($16$GB GPU memory on each GPU). Other experiments are
run at a NVIDIA DGX workstation with CPU E5-2698 ($2.20$GHz, $20$ cores), and
$4$ NVIDIA V$100$ GPUs. For the middleware accelerator abstraction, we treat
CPU in one node as an accelerator which has a 20-thread multithread processing
model, and we treat each GPU as an accelerator which has 1024-thread
multithread processing model.We build our system with Ubuntu 16.04.5 LTS and
deploy a Nvidia-docker framework for simulating the distributed environment
with heterogeneous processors. We construct a cross compilation solution by
using maven, sbt and cmake to manage and config the global project
dependencies. For Spark runtime, we use Java 8 and Scala 2.11. For local C++
and CUDA programming, g++$7$ and CUDA $10.0$ are used. The source code is
available in GX-Plug repository. Also, code for GraphX integration can be
found at GraphXwithGPU repository[31], and code for PowerGraph integration can
be found PowerGraph-GPU repository.

###  V-B Results

####  V-B1 Results on real graphs

Figure 15 compares the performance of GraphX and PowerGraph with non-
acclerator(no prefix), CPU-integrated (prefix CPU+) and GPU-integrated (prefix
GPU+) The y-axis is in log scale. In relatively computation-dense applications
such as LP and SSSP-BF, acceleration can be observed in total time.

On GraphX, compared with GraphX, GPU+GraphX achieves up to 7x acceleration in
SSSP-BF, and up to 20x acceleration in LP algorithm. CPU+GraphX also achieves
up to 4x acceleration in SSSP-BF algorithm and up to 5x acceleration in LP
algorithm. On PowerGraph, GPU+PowerGraph achieves up to 25x acceleration in
SSSP-BF algorithm and up to 15x acceleration in LP algorithm. CPU+PowerGraph
also achieves up to 5x acceleration in SSSP-BF algorithm and up to 10x
acceleration in LP algorithm. These results verify the effectiveness of the
middleware.

The results on the scalability are shown in Figure 15. First, we compare three
competitors, PowerGraph+GX-plug, Gunrock [4], and Lux [19], by varying the
number of GPUs in Figure 15 (a-b). Gunrock is a single-node single-GPU graph
system, and Lux is a multi-node multi-GPU graph system. In Figure 15 (a), the
result (Orkut, PageRank) shows that the runtime cost decreases w.r.t. the
number GPUs. Gunrock performs the best on the single-GPU setting, but the
multi-GPU setting is not supported. When there are more than $2$ GPUs, the
performance of PowerGraph+GX-plug is better than Lux, and the lead is growing,
showing better scalability w.r.t. the number of GPUs. We proceed to evaluate
the scalability on larger datasets, i.e., Twitter and UK-2007, in Figure 15
(b). Gunrock gets overflowed on the two datasets, because it only supports the
single-GPU setting and the graph data cannot be accommodated by a single GPU.
PowerGraph+GX-plug performs better than Lux on the two datasets. For example,
PowerGraph+GX-plug is about $40$% faster than Lux when processing Twitter with
$4$ GPUs. The technology pathways of Lux and GX-plug are different. The former
focuses on exploiting GPU internal mechanisms, while the latter explores more
optimizations on the upper system end, e.g., synchronization skipping, which
may become more critical for the scalability on large datasets. There is no
result for using $4$ GPUs on UK-2007, for all methods, because the system GPU
memory capacity is exceeded. Then, we examine the scalability of
PowerGraph+GX-plug on different graph algorithms in Figure 15 (c). It can be
observed that the sublinear speedup in computation is achieved. In particular,
the runtime cost of SSSP-BF decreases from $14$s to $7$s, if the number of
GPUs is increased from $2$ to $4$. The result of mixing and matching different
accelerators, i.e., CPUs and GPUs, is shown in Figure 15 (d). It shows that
the runtime cost decreases if the computation power increases. We will discuss
the workload balancing on the heterogeneous system.

####  V-B2 Effect of Pipeline Shuffle

Figure 15 shows the experiment results of the performance of pipeline shuffle
mechanism. We consider $3$ competitors, “Pipeline*”, “Pipeline” and “Without
pipeline”, in corresponding to the results with optimal blocksize, fixed
blocksize, and the one without pipeline parallelism. Experiment shows that
“Pipeline*” can achieve 30%-50% acceleration rate compared with
“withoutPipeline”. Also, “Pipeline*” can improve pipeline performance as
20%-30%, compared with “Pipeline”.

####  V-B3 Effect of Synchronization Caching & Skipping

Figure 15 (a) shows the performance of the synchronization caching mechanism.
We use both synthetic and real graph datasets as input, and use SSSP-BF
algorithm for testing the workload. The experiment result shows that we can
get 2-3x acceleration in GraphX integrations. For the results on PowerGraph,
it is much more efficient than GraphX. We can get up to 150% acceleration in
both synthetic and real datasets.

Figure 15 (b) shows the performance of synchronization skipping mechanism. We
use SSSP-BF algorithm for testing the workload, and count the number of
iterations skipped on both synthetic and real datasets. We also compare the
result with the number of iterations when synchronization skipping mechanism
is disabled. For real datasets, the synchronization skipping mechanism
achieves 60%-90% decrease of the number of iterations. However, the effect on
synthetic dataset is insignificant, where the data are more uniform, due the
random generation of nodes and edges. For real datasets, there tends to be
more clusters of dense partitions, leading to better partitioning results that
triggers synchronization skipping.

####  V-B4 Effect of Workload Balancing

Figure 15 shows the results of workload balancing. We compare the difference
of system performance with and without workload balancing. Also, we plot the
best performance can be achieved in accordance to our estimation model as
discussed in Section III-C.

Figure 15 (a) shows the scenario in which the hardware configuration of
distributed nodes are fixed, and the partitioning strategy can be tuned (Case
1, Section III-C). We construct two distributed nodes for the experiment. One
node contains $1$ GPU + $1$ CPU, and the other contains $3$ GPUs + $1$ CPU. We
evenly partition the graph dataset to all nodes, which is the default setting
of distributed graph systems, and is denoted as “Not Balanced”. We compare it
to the one with our balancing strategy as discussed in Section III-C, which is
denoted as “Balanced”. It shows that the workload balancing can significantly
improve the system performance. Also, the balanced result is very close to the
theoretically optimal result.

Figure 15 (b) shows the scenario in which the partitioned results are fixed,
and the hardware configuration can be tuned (Case 2, Section III-C). We
construct $2$ distributed nodes with the same hardware configuration. We vary
the data load of distributed nodes to observe the effect of hardware
configuration tuning. Without balancing techniques, both distributed nodes are
with $1$ GPU, denoted as “Not Balanced”. With balancing techniques, we can
estimate the number of GPUs needed in accordance to the data load and
dynamically allocate appropriate number GPUs, denoted as “Balanced”. It shows
that the workload balancing can significantly improve the system performance.
Also, the balanced result is very close to the theoretically optimal result,
demonstrating the merits of workload balancing strategies.

####  V-B5 Effect of Runtime Isolation

We hereby examine the performance of computation daemon on the runtime
isolation by designing a comparative test to compare the influence of GPU
initialization between daemon-agent based solution and direct GPU call
solution. A larger number of iterations corresponds to a higher number of
times of CPU-GPU runtime environment switching. Results in Figure 15 (11
iterations) show that our solution significantly reduces unnecessary
initialization costs. The benefits would be amplified when the number of
iterations is increased.

####  V-B6 Middleware Scalability

We examine the scalability of our middleware, by varying the number of
distributed nodes, in Figure 15. It plots the ratio of time cost taken by the
middleware to the cost of the entire system, for different graph tasks on
different distributed systems, e.g., PowerGraph and GraphX. It can be observed
that, for all graph tasks, the time ratio of the middware decreases w.r.t. the
increase of number of distributed nodes. The downhill trend reflects good
scalability of the middleware in a larger scaled distributed computing
environment, where the cost can be dominated by the gradually enlarged
synchronization overhead of distributed system side. Also, the time ratios of
middleware are mostly between 10% and 20%, especially for algorithms with high
operational intensities. Particularly, PageRank takes only about 10% of total
cost in a distributed system with 32 nodes. LP is different, since it is a
fully iterative algorithm, corresponding to a low operational intensity.

In summary, the low cost ratio and the downhill trend demonstrate good
scalability of our middleware.

####  V-B7 Block Size Selection

To examine the effect of the block size selection, we measure the pipeline
performance in different with different amount of blocks $s$, in Figure 15. We
also compare the estimated $s_{opt}=\frac{d}{b_{opt}}$ with the real
result666Coefficients are tested as follows: for SSSP: ($k_{1}$, $k_{2}$,
$k_{3}$, $a$) = (0.03, 0.51, 0.09, 84671); for PR: ($k_{1}$, $k_{2}$, $k_{3}$,
$a$) = (0.02, 0.58, 0.1, 1970); for SSSP: ($k_{1}$, $k_{2}$, $k_{3}$, $a$) =
(0.003, 0.59, 0.006, 498).. For both LP and PageRank algorithms, we use the
first iteration as the testing data. For SSSP algorithm, we use $6$-th
iteration as the testing data, since the computation workload is the maximum
during the entire execution. We can find that when $s$ increases, iteration
time cost first decreases, and then increases. Thus, for $b=\frac{d}{s}$, when
$b$ increases, iteration time cost also tends to first decrease, and then
increase.

We also give our estimated $s$ following the analysis here for the $3$
different algorithms. It shows that when real $b$ and $s$ are close to the
estimated result, the pipeline performance is also close to the estimated one,
showing the accuracy of our estimation. Also, the optimal performance can be
reached when real $b$ and $s$ are close to our estimation.

##  VI Related Works

Distributed CPU-based Systems. With the prosperity of distributed system,
people investigate common operator sets inside diverse graph primitives for
scaling out in the distributed environment. As a forerunner, Pregel[11] is
proposed by Google on large-scale graph computing, following the BSP model. In
BSP, graph computation are divided into iterations and intermediate results
can be globally synchronized at barriers called super-steps. GraphLab[32]
allows asynchronous computation and dynamic asynchronous scheduling, whose
programming model also isolates the user-defined algorithm from data movement.
To achieve better workload balancing on natural graphs, PowerGraph[3] uses a
more flexible GAS abstraction for power-law graphs.

There are also many embedded graph processing systems built on existing
distributed systems to gain the benefits of task scheduling and data
management. GraphX[2] is one of the most successful representative built on
top of Apache Spark[33]. HaLoop[34] is a similar distributed graph processing
system, in particular, extended from Hadoop[35].

However, most works pay little attention to the computation intensiveness of
large-scale graph processing. Efforts on scheduling balancing and data
accessing also incur extra cost in computation, making the system even slower
than single-node solutions. It is thus desired to have a scale-up solution for
distributed graph systems.

Single-node Parallel Graph Algorithms. There also exist hardwired graph
primitive implementations for the single-node environment. Merrill et al.
propose linear parallelization of BFS algorithm on GPU [36]. Soman et al.
studies graph algorithms based on two PRAM connected-component [37]. Several
parallel Betweenness Centrality implementations are available on GPU based on
the work of Brandes et al. [38]. Davidson et al.[39] propose a work-efficient
Single-Source Shortest Path algorithm on GPU.

Low-level graph parallel solutions can have best performance only on specific
computation tasks, but are not general for diversified graph applications.
Also, the hardwired primitives are challenging to even skilled algorithm
engineers, making such solutions hard for being deployed in real systems and
applications.

High-level GPU Programming Model. There are also existing works on high-level
graph operations for GPU. Zhong and He devise Medusa [40] on a high-level GPU-
based system for parallel graph processing using a message-passing model,
which is arguably the earliest work for GPGPU development for graphs. CuSha
[41], targeting a GAS abstraction, avoids non-coalesced memory accessing and
avoids irregular memory accessing. Gunrock [4] implements a novel data-centric
abstraction centered on operations on a vertex or edge frontier rather than
designing an abstraction for computation. Recently, there are a few works on
GPU graph processing system built on distributed systems, among which Lux [19]
is one of the representatives. Users can use GPUs in multiple physical nodes
for efficient computation. However, without the support of mature distributed
systems, Lux faces a series of challenges, such as robust distributed data
management, scheduling balancing, effective fault recovery, and efficient data
synchronization with physical layers, and thus falls short in addressing
technical issues arise in large-scale graph data management and analytics.

##  VII Conclusion

In this paper, we propose a middleware for the integration of heterogenous
distributed graph systems and accelerators. Our middleware is versatile in the
sense that it supports different programming models, computation models, and
runtime environments. For reinforcing the middleware performance, we devise a
series of techniques, such as pipeline shuffle, synchronization caching and
skipping, partitioning, and parameter configuration for intra-, inter, and
beyond iteration optimization. Extensive experiments show that our middleware
achieves good performance in large-scale graph processing.

## References

  * [1] K. Bouchard, J. Aimone, M. Chun, T. Dean, M. Denker, M. Diesmann, D. Donofrio, L. Frank, N. Kasthuri, C. Koch, O. Ruebel, H. Simon, F. Sommer, and M. Prabhat, “High-performance computing in neuroscience for data-driven discovery, integration, and dissemination,” _Neuron_ , vol. 92, pp. 628–631, 11 2016.
  * [2] J. E. Gonzalez, R. S. Xin, A. Dave, D. Crankshaw, M. J. Franklin, and I. Stoica, “Graphx: Graph processing in a distributed dataflow framework,” in _OSDI_ , J. Flinn and H. Levy, Eds., 2014, pp. 599–613.
  * [3] J. E. Gonzalez, Y. Low, H. Gu, D. Bickson, and C. Guestrin, “Powergraph: Distributed graph-parallel computation on natural graphs,” in _OSDI_ , 2012, pp. 17–30.
  * [4] Y. Wang, A. A. Davidson, Y. Pan, Y. Wu, A. Riffel, and J. D. Owens, “Gunrock: a high-performance graph processing library on the GPU,” in _PPoPP_ , 2016, pp. 11:1–11:12.
  * [5] X. Chen, H. Tan, Y. Chen, B. He, W. Wong, and D. Chen, “Thundergp: Hls-based graph processing framework on fpgas,” in _FPGA ’21: The 2021 ACM/SIGDA International Symposium on Field Programmable Gate Arrays, Virtual Event, USA, February 28 - March 2, 2021_ , L. Shannon and M. Adler, Eds. ACM, 2021, pp. 69–80.
  * [6] S. Sakr, A. Bonifati, H. Voigt, A. Iosup, K. Ammar, R. Angles, W. G. Aref, M. Arenas, M. Besta, P. A. Boncz, K. Daudjee, E. D. Valle, S. Dumbrava, O. Hartig, B. Haslhofer, T. Hegeman, J. Hidders, K. Hose, A. Iamnitchi, V. Kalavri, H. Kapp, W. Martens, M. T. Özsu, E. Peukert, S. Plantikow, M. Ragab, M. Ripeanu, S. Salihoglu, C. Schulz, P. Selmer, J. F. Sequeda, J. Shinavier, G. Szárnyas, R. Tommasini, A. Tumeo, A. Uta, A. L. Varbanescu, H. Wu, N. Yakovets, D. Yan, and E. Yoneki, “The future is big graphs: a community view on graph processing systems,” _Commun. ACM_ , vol. 64, no. 9, pp. 62–71, 2021.
  * [7] NVIDIA, “Gpu-accelerated apache spark,” https://www.nvidia.com/en-us/deep-learning-ai/solutions/data-science/apache-spark-3/, 2020\.
  * [8] M. D. Linderman, J. D. Collins, H. Wang, and T. H. Meng, “Merge: a programming model for heterogeneous multi-core systems,” in _Proceedings of the 13th International Conference on Architectural Support for Programming Languages and Operating Systems, ASPLOS 2008, Seattle, WA, USA, March 1-5, 2008_ , S. J. Eggers and J. R. Larus, Eds. ACM, 2008, pp. 287–296.
  * [9] A. Haidar, C. Cao, A. YarKhan, P. Luszczek, S. Tomov, K. Kabir, and J. J. Dongarra, “Unified development for mixed multi-gpu and multi-coprocessor environments using a lightweight runtime environment,” in _2014 IEEE 28th International Parallel and Distributed Processing Symposium, Phoenix, AZ, USA, May 19-23, 2014_. IEEE Computer Society, 2014, pp. 491–500.
  * [10] Databrick, “Deep dive into gpu support in apache spark 3.x,” https://databricks.com/session_na20/deep-dive-into-gpu-support-in-apache-spark-3-x, 2020\.
  * [11] G. Malewicz, M. H. Austern, A. J. C. Bik, J. C. Dehnert, I. Horn, N. Leiser, and G. Czajkowski, “Pregel: a system for large-scale graph processing,” in _SIGMOD_ , 2010, pp. 135–146.
  * [12] K. Ammar and M. T. Özsu, “Experimental analysis of distributed graph systems,” _Proc. VLDB Endow._ , vol. 11, no. 10, pp. 1151–1164, 2018.
  * [13] M. T. Özsu, “Graph processing: A panoramic view and some open problems,” _VLDB_ , 2019.
  * [14] X. Chen, “Graphcage: Cache aware graph processing on gpus,” _CoRR_ , vol. abs/1904.02241, 2019.
  * [15] P. Wang, J. Wang, C. Li, J. Wang, H. Zhu, and M. Guo, “Grus: Toward unified-memory-efficient high-performance graph processing on GPU,” _ACM Trans. Archit. Code Optim._ , vol. 18, no. 2, pp. 22:1–22:25, 2021\.
  * [16] A. Li, S. L. Song, J. Chen, J. Li, X. Liu, N. R. Tallent, and K. J. Barker, “Evaluating modern gpu interconnect: Pcie, nvlink, nv-sli, nvswitch and gpudirect,” _IEEE Transactions on Parallel and Distributed Systems_ , vol. 31, no. 1, pp. 94–110, 2020.
  * [17] NVIDIA, “Nvlink and nvswitch: The building blocks of advanced multi-gpu communication,” https://www.nvidia.com/en-us/data-center/nvlink/, 2021\.
  * [18] H. Fu, M. G. Venkata, S. Salman, N. Imam, and W. Yu, “Shmemgraph: Efficient and balanced graph processing using one-sided communication,” in _18th IEEE/ACM International Symposium on Cluster, Cloud and Grid Computing, CCGRID 2018, Washington, DC, USA, May 1-4, 2018_ , E. El-Araby, D. K. Panda, S. Gesing, A. W. Apon, V. V. Kindratenko, M. Cafaro, and A. Cuzzocrea, Eds. IEEE Computer Society, 2018, pp. 513–522.
  * [19] Z. Jia, Y. Kwon, G. M. Shipman, P. S. McCormick, M. Erez, and A. Aiken, “A distributed multi-gpu system for fast graph processing,” _VLDB_ , vol. 11, no. 3, pp. 297–310, 2017.
  * [20] R. Albert, H. Jeong, and A.-L. Barabási, “Error and attack tolerance of complex networks,” _Nature_ , vol. 406, no. 6794, pp. 378–382, 2000.
  * [21] D. Donato, L. Laura, S. Leonardi, and S. Millozzi, “Large scale properties of the webgraph,” _The European Physical Journal B_ , vol. 38, no. 2, pp. 239–243, 2004.
  * [22] D. Kong, X. Xie, and Z. Zhang, “Clustering-based partitioning for large web graphs,” in _38th IEEE International Conference on Data Engineering, ICDE 2022, Kuala Lumpur, Malaysia, May 9-12, 2022_. IEEE, 2022.
  * [23] N. A. Halli, H. Charles, and J. Méhaut, “Performance comparison between java and JNI for optimal implementation of computational micro-kernels,” _CoRR_ , vol. abs/1412.6765, 2014.
  * [24] A. Mislove, M. Marcon, K. P. Gummadi, P. Druschel, and B. Bhattacharjee, “Measurement and Analysis of Online Social Networks,” in _Proceedings of the 5th ACM/Usenix Internet Measurement Conference (IMC’07)_ , San Diego, CA, October 2007.
  * [25] H. Yin, A. R. Benson, J. Leskovec, and D. F. Gleich, “Local higher-order graph clustering,” in _Proceedings of the 23rd ACM SIGKDD International Conference on Knowledge Discovery and Data Mining, Halifax, NS, Canada, August 13 - 17, 2017_. ACM, 2017, pp. 555–564.
  * [26] L. Backstrom, D. P. Huttenlocher, J. M. Kleinberg, and X. Lan, “Group formation in large social networks: membership, growth, and evolution,” in _Proceedings of the Twelfth ACM SIGKDD International Conference on Knowledge Discovery and Data Mining, Philadelphia, PA, USA, August 20-23, 2006_ , T. Eliassi-Rad, L. H. Ungar, M. Craven, and D. Gunopulos, Eds. ACM, 2006, pp. 44–54.
  * [27] R. A. Rossi and N. K. Ahmed, “The network data repository with interactive graph analytics and visualization,” in _AAAI_ , 2015.
  * [28] P. Boldi and S. Vigna, “The webgraph framework I: compression techniques,” in _Proceedings of the 13th international conference on World Wide Web, WWW 2004, New York, NY, USA, May 17-20, 2004_ , S. I. Feldman, M. Uretsky, M. Najork, and C. E. Wills, Eds. ACM, 2004, pp. 595–602.
  * [29] P. Boldi, M. Rosa, M. Santini, and S. Vigna, “Layered label propagation: a multiresolution coordinate-free ordering for compressing social networks,” in _Proceedings of the 20th International Conference on World Wide Web, WWW 2011, Hyderabad, India, March 28 - April 1, 2011_ , S. Srinivasan, K. Ramamritham, A. Kumar, M. P. Ravindra, E. Bertino, and R. Kumar, Eds. ACM, 2011, pp. 587–596.
  * [30] D. dataset, “uk-2007-02 dataset,” https://law.di.unimi.it/webdata/uk-2007-02/, 2007.
  * [31] Q. Li, K. Zou, D. Kong, H. Guan, and X. Xie, “Gpugraphx: A gpu-aided distributed graph processing system,” in _WISE 2021 - 22nd International Conference on Web Information Systems Engineering_ , vol. 13081\. Springer, 2021, pp. 501–509.
  * [32] Y. Low, J. Gonzalez, A. Kyrola, D. Bickson, C. Guestrin, and J. M. Hellerstein, “Distributed graphlab: A framework for machine learning in the cloud,” _VLDB_ , vol. 5, no. 8, pp. 716–727, 2012.
  * [33] “Spark,” http://spark.apache.org, 2019.
  * [34] Y. Bu, B. Howe, M. Balazinska, and M. D. Ernst, “The haloop approach to large-scale iterative data analysis,” _VLDB J._ , vol. 21, no. 2, pp. 169–190, 2012.
  * [35] “Hadoop,” http://hadoop.apache.org, 2019.
  * [36] D. Merrill, M. Garland, and A. S. Grimshaw, “Scalable GPU graph traversal,” in _PPoPP_ , 2012, pp. 117–128.
  * [37] J. Greiner, “A comparison of parallel algorithms for connected components,” in _SPAA_ , 1994, pp. 16–25.
  * [38] U. Brandes, “A faster algorithm for betweenness centrality,” _Journal of Mathematical Sociology_ , vol. 25, 2001.
  * [39] A. A. Davidson, S. Baxter, M. Garland, and J. D. Owens, “Work-efficient parallel GPU methods for single-source shortest paths,” in _IPPS_ , 2014, pp. 349–359.
  * [40] J. Zhong and B. He, “Medusa: A parallel graph processing system on graphics processors,” _SIGMOD Record_ , vol. 43, no. 2, pp. 35–40, 2014.
  * [41] F. Khorasani, K. Vora, R. Gupta, and L. N. Bhuyan, “Cusha: vertex-centric graph processing on gpus,” in _HPDC_ , 2014, pp. 239–252.