# DMGA: A Distributed Shortest Path Algorithm for Multistage Graph

Authors: Huanqing Cui; Ruixue Liu; Shaohua Xu; Chuanai Zhou
Journal: Scientific Programming
 (2021)

---

## Abstract

The multistage graph problem is a special kind of single-source single-sink shortest path problem. It is difficult even impossible to solve the large-scale multistage graphs using a single machine with sequential algorithms. There are many distributed graph computing systems that can solve this problem, but they are often designed for general large-scale graphs, which do not consider the special characteristics of multistage graphs. This paper proposes DMGA (Distributed Multistage Graph Algorithm) to solve the shortest path problem according to the structural characteristics of multistage graphs. The algorithm first allocates the graph to a set of computing nodes to store the vertices of the same stage to the same computing node. Next, DMGA calculates the shortest paths between any pair of starting and ending vertices within a partition by the classical dynamic programming algorithm. Finally, the global shortest path is calculated by subresults exchanging between computing nodes in an iterative method. Our experiments show that the proposed algorithm can effectively reduce the time to solve the shortest path of multistage graphs.


---

## Body

## 1. Introduction

With the continuous development of big data and information technology, graph has been widely applied in many applications, and various graph structures and algorithms have been proposed. Among them, the multistage graph is a special kind of weighted directed graphs, which are widely used in engineering technology, concurrency control, transportation, task schedule in high-performance computing, and other fields. Many coordination or dynamic scheduling problems can be transformed into multistage graph problems [1, 2].Recently, the scale of graph data has grown tremendously, so it is difficult even impossible to store and process such large-scale graphs by a single computer or sequential processing method [3]. At this point, the distributed computing scheme became a must, and lots of dedicated graph-processing systems have been appearing [4–6], such as Pregel [7], PowerGraph [8], GraphX [9, 10], GraphLab [11], and PowerLyra [12]. These graph processing systems extend the computation by dividing the graph into multiple partitions and processing on multiple computing nodes in parallel. High-quality partition can reduce the communication cost and achieve the load balance [13–15], thus the processing time can be minimized subsequently. The current distributed graph processing systems and algorithms are usually designed for general graphs, and they do not consider the special structural properties of multistage graphs, so there are some disadvantages in applying them to multistage graphs, such as high communication cost and long solution time. The purpose of this paper is to present a distributed algorithm DMGA (Distributed Multistage Graph Algorithm) for the shortest path problem of multistage graphs to make full use of their characteristics. The main contributions are as follows:(1)
It presents a partitioning method for multistage graphs on distributed computing systems, which can make best use of characteristics of multistage graphs to achieve the best load balance and reduce the communication cost(2)
It designs a distributed algorithm of the shortest path problem of multistage graphs based on dynamic programming idea(3)
It performs extensive experiments to verify the performance of the proposed algorithm, compared to the classical parallel Dijkstra algorithm and the SSSP (single-source shortest path) algorithm on PregelTable1 gives an overview about the notations used in this paper. The organizations of the rest of the paper are as follows. Section 2 introduces the related works, and Section 3 presents the statements of the shortest path problem of multistage graphs. Section 4 describes the proposed DMGA algorithm, and Section 5 introduces the experiments and analysis. Section 6 concludes the paper.Table 1
Notation overview.
SymbolDefinitionG = (V, E, W)Graph with vertices setV, edges set E, and edge weights’ set WmNumber of stages in a multistage graphViSet of vertices of stagei of a multistage graphniNumber of vertices of stagei of a multistage graphvi,jThejth vertex in the ith stage of a multistage graphwi,j,i+1,kWeight of edge <vi,j, vi+1,k>EiSet of edges fromVi to Vi+1ck,i,l,jCost of the shortest path from vertexvk,i to vl,jCNiTheith partition or computing nodepNumber of partitionsCapCapacity of each computing nodeMLMaximum load of each partitionSumNumber of edges of a partitionskThe first stage of CNkekThe last stage of CNkfsk,i,j,lIndex of the previous vertex ofvj,l in the shortest path from vsk,i to vj,lspsk,j,ek,iList of vertices of the shortest path fromvsk,j to vek,iRSet of partition IDsSPLkSet of the shortest paths storing on CNk


## 2. Related Works

### 2.1. The Shortest Path Algorithms

Finding the shortest path is a classical problem of graph theory, and the well-known sequential algorithms are Dijkstra, Floyd, and Bellman–Ford algorithms [16], which perform well in centralized computing. However, the large-scale graph needs distributed computing algorithms to obtain the shortest paths quickly.The single-source shortest path (SSSP) is one of the most important shortest path problems. Peng et al. [1] defined a new graph model named by single-source-weighted multilevel graph and presented a parallel SSSP algorithm by constructing the vector-matrix multiple model, dividing into parallel tasks, and setting data communication’s method. A·Davidson [17] developed three parallel SSSP algorithms for GPUs (Graphics Processing Unit): Workfront Sweep, Near-Farand, and Bucketing. These algorithms utilize different approaches to balance the trade-off between saving work and organizational overhead. S·Maleki [18] introduced a partially asynchronous parallel DSMR (Dijkstra Strip Mined Relaxation) algorithm for SSSP on shared and distributed memory systems. Busato and Bombieri [19] proposed a parallel Bellman–Ford algorithm based on frontier and active vertices that exploit the architectural characteristics of GPU architectures. Huang [20] gave a distributed Las Vegas algorithm on the classic scaling technique for the all-pairs’ shortest paths on distributed networks. For the dynamic and stochastic graph models of the transportation network, Liu et al. [21] proposed an improved adaptive genetic algorithm by adjusting the encoding parameters to get the dynamic random shortest path. Ghaffari and Li [22] provided a distributed SSSP algorithm with less complexity, and it constitutes the first sublinear time algorithm for directed graphs. For the SSPP-MPN (Shortest Simple Path Problem with Must-Past Nodes), Su et al. [23] proposed a multistage metaheuristic algorithm based on k-opt move, candidate path search, conflicting nodes promotion, and connectivity relaxation.The above algorithms do not consider the structural characteristics of multistage graphs, so they will produce a large amount of communication overhead, resulting in tedious execution time.


### 2.2. Graph Partitioning Algorithms

The basis of the distributed graph processing system is to partition the entire graph into a set of computing nodes. The graph partition algorithms are classified into vertex-cut and edge-cut. Edge-cut partitioning assigns each vertex to a unique partition, and the edge spanning partitions are called cut edge. As shown in Figure1(a), edges <b,d> are cut, and their two endpoints b and d are assigned to different partitions. Vertex-cut partitioning assigns edges uniquely to a certain partition, which results in vertex-cuts across multiple partitions [24]. As shown in Figure 1(b), vertex d is partitioned, both partitions P1 and P2 have copies of vertex d, and their references in each partition are also called mirrors [25].Figure 1
Two graph partitioning methods. (a) Edge-cut partition. (b) Vertex-cut partition.
(a)(b)The distributed graph processing systems often use the vertex-centric programming model [26, 27], where the computing node recursively operates its active vertices according to the user-defined graph function. Each vertex reads the statuses of its adjacent vertices or edges and updates its own status accordingly. In the iterative calculation of a graph, the partitions exchange intermediate results along edges. To some extent, the number of cut edges or mirror vertices can reflect the network communication overhead, which in turn affects the calculation efficiency. On the contrary, the load among computing nodes should be balanced to ensure that the computing nodes can achieve the results synchronously. Hence, both edge-cut and vertex-cut approaches aim to minimize cross-partition dependencies and achieve load balance [25].The existing graph partitioning heuristic solutions are basically divided into offline and online partitioning strategies. The offline partitioning strategy refers to dividing the graph into several subgraphs before being loaded by the distributed system. F·Rahimian et al. [28] introduced the JA-BE-JA algorithm that uses local search and simulated annealing techniques for graph partitioning. The algorithm only needs to use part of the information to process the graph. Akhremtsev et al. [29] presented a multilevel shared-memory parallel graph partitioning algorithm that uses parallel label propagation for both coarsening and refinement, and it can balance the speed and quality of parallel graph partitioning.The online partitioning strategy refers to partitioning the graph during the data loading process, where the input data is usually a vertex stream or an edge stream. Tsourakakis et al. [30] proposed the FENNEL algorithm based on locality-centric measures and balancing goals. Its core idea is to interpolate between maximizing the co-location of neighbouring vertices and minimizing that of nonneighbours. Petroni et al. [15] proposed the high-degree replicated first (HDRF) algorithm according to the characteristics of power-law graphs, which divide the vertices with high degrees in first. Zhang et al. [31] proposed the AKIN algorithm based on the vertex similarity index, which exploits the similarity measure of the vertex degree to collect structure-related vertices in the same partition to further reduce the edge-cut rate. Wang et al. [32] analysed the locality of the graph and proposed the target-vertex sensitive Hash algorithm. The algorithm predivides the target vertices of the edge logically and then partitions the graph in parallel according to the target vertices. Ji et al. [33] proposed a two-stage local partitioning algorithm which introduces the concept of local partitions, emphasizing the impact of changes in the graph structure on the quality of partitions. Slota et al. [34] introduced XtraPuLP based on the scalable label propagation community detection technique. It can solve the multiple constraint and multiple objective graph partition problem on tera-scale graphs. Zhou et al. [35] proposed Geo-Cut which uses a cost-aware streaming heuristic and two partition refinement heuristics to reduce the cost and data transfer time of geo-distributed data centres.The above graph partitioning algorithm are all designed for general graphs, which do not consider the special characteristics of multistage graphs, so it is necessary to design the graph partitioning algorithm for multistage graphs to accelerate the distributed processing.


## 2.1. The Shortest Path Algorithms

Finding the shortest path is a classical problem of graph theory, and the well-known sequential algorithms are Dijkstra, Floyd, and Bellman–Ford algorithms [16], which perform well in centralized computing. However, the large-scale graph needs distributed computing algorithms to obtain the shortest paths quickly.The single-source shortest path (SSSP) is one of the most important shortest path problems. Peng et al. [1] defined a new graph model named by single-source-weighted multilevel graph and presented a parallel SSSP algorithm by constructing the vector-matrix multiple model, dividing into parallel tasks, and setting data communication’s method. A·Davidson [17] developed three parallel SSSP algorithms for GPUs (Graphics Processing Unit): Workfront Sweep, Near-Farand, and Bucketing. These algorithms utilize different approaches to balance the trade-off between saving work and organizational overhead. S·Maleki [18] introduced a partially asynchronous parallel DSMR (Dijkstra Strip Mined Relaxation) algorithm for SSSP on shared and distributed memory systems. Busato and Bombieri [19] proposed a parallel Bellman–Ford algorithm based on frontier and active vertices that exploit the architectural characteristics of GPU architectures. Huang [20] gave a distributed Las Vegas algorithm on the classic scaling technique for the all-pairs’ shortest paths on distributed networks. For the dynamic and stochastic graph models of the transportation network, Liu et al. [21] proposed an improved adaptive genetic algorithm by adjusting the encoding parameters to get the dynamic random shortest path. Ghaffari and Li [22] provided a distributed SSSP algorithm with less complexity, and it constitutes the first sublinear time algorithm for directed graphs. For the SSPP-MPN (Shortest Simple Path Problem with Must-Past Nodes), Su et al. [23] proposed a multistage metaheuristic algorithm based on k-opt move, candidate path search, conflicting nodes promotion, and connectivity relaxation.The above algorithms do not consider the structural characteristics of multistage graphs, so they will produce a large amount of communication overhead, resulting in tedious execution time.


## 2.2. Graph Partitioning Algorithms

The basis of the distributed graph processing system is to partition the entire graph into a set of computing nodes. The graph partition algorithms are classified into vertex-cut and edge-cut. Edge-cut partitioning assigns each vertex to a unique partition, and the edge spanning partitions are called cut edge. As shown in Figure1(a), edges <b,d> are cut, and their two endpoints b and d are assigned to different partitions. Vertex-cut partitioning assigns edges uniquely to a certain partition, which results in vertex-cuts across multiple partitions [24]. As shown in Figure 1(b), vertex d is partitioned, both partitions P1 and P2 have copies of vertex d, and their references in each partition are also called mirrors [25].Figure 1
Two graph partitioning methods. (a) Edge-cut partition. (b) Vertex-cut partition.
(a)(b)The distributed graph processing systems often use the vertex-centric programming model [26, 27], where the computing node recursively operates its active vertices according to the user-defined graph function. Each vertex reads the statuses of its adjacent vertices or edges and updates its own status accordingly. In the iterative calculation of a graph, the partitions exchange intermediate results along edges. To some extent, the number of cut edges or mirror vertices can reflect the network communication overhead, which in turn affects the calculation efficiency. On the contrary, the load among computing nodes should be balanced to ensure that the computing nodes can achieve the results synchronously. Hence, both edge-cut and vertex-cut approaches aim to minimize cross-partition dependencies and achieve load balance [25].The existing graph partitioning heuristic solutions are basically divided into offline and online partitioning strategies. The offline partitioning strategy refers to dividing the graph into several subgraphs before being loaded by the distributed system. F·Rahimian et al. [28] introduced the JA-BE-JA algorithm that uses local search and simulated annealing techniques for graph partitioning. The algorithm only needs to use part of the information to process the graph. Akhremtsev et al. [29] presented a multilevel shared-memory parallel graph partitioning algorithm that uses parallel label propagation for both coarsening and refinement, and it can balance the speed and quality of parallel graph partitioning.The online partitioning strategy refers to partitioning the graph during the data loading process, where the input data is usually a vertex stream or an edge stream. Tsourakakis et al. [30] proposed the FENNEL algorithm based on locality-centric measures and balancing goals. Its core idea is to interpolate between maximizing the co-location of neighbouring vertices and minimizing that of nonneighbours. Petroni et al. [15] proposed the high-degree replicated first (HDRF) algorithm according to the characteristics of power-law graphs, which divide the vertices with high degrees in first. Zhang et al. [31] proposed the AKIN algorithm based on the vertex similarity index, which exploits the similarity measure of the vertex degree to collect structure-related vertices in the same partition to further reduce the edge-cut rate. Wang et al. [32] analysed the locality of the graph and proposed the target-vertex sensitive Hash algorithm. The algorithm predivides the target vertices of the edge logically and then partitions the graph in parallel according to the target vertices. Ji et al. [33] proposed a two-stage local partitioning algorithm which introduces the concept of local partitions, emphasizing the impact of changes in the graph structure on the quality of partitions. Slota et al. [34] introduced XtraPuLP based on the scalable label propagation community detection technique. It can solve the multiple constraint and multiple objective graph partition problem on tera-scale graphs. Zhou et al. [35] proposed Geo-Cut which uses a cost-aware streaming heuristic and two partition refinement heuristics to reduce the cost and data transfer time of geo-distributed data centres.The above graph partitioning algorithm are all designed for general graphs, which do not consider the special characteristics of multistage graphs, so it is necessary to design the graph partitioning algorithm for multistage graphs to accelerate the distributed processing.


## 3. Problem Statements

A multistage graphG=V,E,W is a directed single-source and single-sink weighted connected graph, where V and E are, respectively, the set of vertices and edges and W is the weights of edges. The vertices are divided into disjoint stages, and each edge can only point from the vertex of the previous stage to the vertex of the succeeding stage. Formally, a multistage graph G=V,E,W should satisfy(1)
V=∪i=1mVi,∀i≠jandVi∩Vj=∅, where m is the number of stages.(2)
Vi=vi,j|j=1,2,…,ni, where ni is the number of vertices of the ith stage.(3)
E=vi,j,vi+1,k|i=1,2,…,m−1;j=1,2,…,ni;k=1,2,…,ni+1.(4)
W=wi,j,i+1,k|i=1,2,…,m−1;j=1,2,…,ni;k=1,2,…,ni+1, where wi,j,i+1,k is the weight of edge vi,j,vi+1,k. If vi,j,vi+1,k∉E, wi,j,i+1,k=∞.(5)
V1=v1,1, Vm=vm,1, and v1,1 and vm,1 are, respectively, the source vertex and sink vertex.Figure2 is an example of the multistage graph, where the blue numbers above the graph are the numbers of edges. This paper supposes that the multistage graphs are dense which means(1)Ei≈ni×ni+1,where Ei=vi,j,vi+1,k|j=1,2,…,ni;k=1,2,…,ni+1 is the set of edges between Vi and Vi+1 and Ei is the number of edges of Ei.Figure 2
A multistage graph.The shortest path problem of a multistage graph is to find the minimum cost path from the source vertex to the sink vertex. Letck,i,l,j be the cost of the shortest path from vertex vk,i to vl,j. Obviously, c1,1,m,1 is the cost of the shortest path from the source vertex to the sink vertex, and(2)c1,1,k,i=0,ifk=1andi=1,minvk−1,j,vk,i∈Ec1,1,k−1,j+wk−1,j,k,i,ifk>1andi=1,2,…,nk.Given a large-scale multistage graphG=V,E,W, we need to partition it to a cluster of computing nodes. Each computing node stores a part of G, and each part is called a partition. Let p be the number of partitions, so G is divided into partitions CN1,CN2,…,CNp and CNi is located on the ith computing node.


## 4. DMGA: The Proposed Algorithm

DMGA is run on the homogeneous cluster, which means all computing nodes have the same performance in terms of CPU, memory, and bandwidth. This algorithm partitions the entire graph to the given cluster first, and then, each computing node computes the shortest path of the partition on it. Finally, the computing nodes communicate with each other to obtain the shortest path of the whole graph.Algorithm1 gives the framework of DMGA. The details of each step of Algorithm 1 are given in the followingsections.Algorithm 1: DMGA.
Require: A large-scale multistage graph G=V,E,WEnsure: The shortest path from v1,1 to vm,1(1)
Partition G(2)
Get the local shortest path of each partition(3)
Get the global shortest path of the whole graph


### 4.1. Multistage Graph Partition

In order to determine the graph partition strategy, we should analyse their impacts on the communication overhead after partition. According to the feature of multistage graphs, it is a better scheme to divide the vertices of the same stage into the same partition because it is easy to implement load balance and parallel shortest path solution. SupposeVi and Vi+1 are divided into two different partitions. If we use vertex-cut strategy, the number of mirror vertices is either ni or ni+1. If we use edge-cut strategy, the number of cut edges is Ei. According to (1), ni<Ei and ni+1<Ei, which indicates that the vertex-cut strategy has less communication overhead than edge-cut strategy, so we adopt vertex-cut strategy to partition the graph. Figure 3 is an example. The edge-cut strategy produces 9 cut edges (Figure 3(a)), while the vertex-cut strategy only produces 3 mirror vertices (Figure 3(b)).Figure 3
Two graph partitioning strategies. (a) Edge-cut strategy. (b) Vertex-cut strategy.
(a)(b)Since the multistage graphs studied in this paper are dense, we use the number of edges to represent the load of a partition. Let Cap be the capacity of each computing node, which is also the maximum number of edges that can be stored by a partition, then the number of partitions of a givenG can be estimated as(3)p=ECap.The above equation gives the lower limit of the number of computing nodes. It may lead to the load of the last computing node far lower than those of the other computing nodes. For example, ifE=10100 and Cap=1000, then p=11. If the first 10 computing nodes are fully loaded, then the last computing node only has 100 edges, so the load is imbalanced. Hence, the maximum load of each partition is redefined as(4)ML=Cap,ifp=ECap,γEp,otherwise,where γ is a predefined parameter to keep the load balance for different multistage graphs.The idea of multistage graph partition is to assign the vertices of the same stage to the same partition. Figure4 presents the flow diagram, and Algorithm 2 presents pseudocode. In this algorithm, Sum records the number of edges stored in the current partition. Lines 1 and 2 initialize the variables. Lines 3–16 divide G to a cluster of computing nodes. If Sum≤ML (line 5), lines 6–8 assign the edges of Ei to computing node CNk, and lines 9 and 10, respectively, update ek and i. If Sum>ML (line 11), which means the current partition CNk will be overloaded if we assign the edges of Ei to it, lines 12–14 update variables to prepare for succeeding partition.Figure 4
Flow diagram of graph partition.Algorithm 2: Multistage graph partitioning algorithm.
Require: Multistage graph GEnsure: Partition result of G(1)
Sum⟵0,k⟵1,sk⟵1,i⟵1(2)
CalculateML using (4)(3)
whilei≤m−1 do(4)
Sum⟵Sum+Ei(5)
ifSum≤ML then(6)
for allvi,j,vi+1,l∈Ei do(7)
Assign it toCNk(8)
end for(9)
ek⟵i+1(10)
i⟵i+1(11)
else(12)
Sum⟵0(13)
k⟵k+1(14)
sk⟵i(15)
end if(16)
end while


### 4.2. Local Shortest Path Calculation

After partitioning the graph, each computing node calculates the shortest path of the subgraph stored on it. The shortest path of each partition is referred to as the local shortest path, and the shortest path of the whole graph is referred to as the global shortest path.Theorem 1.
Ifv1,1,v2,i2,…,vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il,…,vm,1 is one of the shortest paths from v1,1 to vm,1, then vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il is one of the shortest paths from vk,ik to vl,il.Proof.
We prove it by using reduction to absurdity. Supposevk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il is not one of the shortest paths from vk,ik to vl,il. There must exist a shortest path from vk,ik to vl,il. Let vk,ik,vk+1,ik+1′,…,vl−1,il−1′,vl,il be one of the shortest paths from vk,ik to vl,il. If we use vk+1,ik+1′,…,vl−1,il−1′ to replace vk+1,ik+1,…,vl−1,il−1 in the path v1,1,v2,i2,…,vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il,…,vm,1, then the cost of v1,1,v2,i2,…,vk,ik,v′k+1,ik+1,…,v′l−1,il−1,vl,il,…,vm,1 is less than v1,1,v2,i2,…,vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il,…,vm,1, so v1,1,v2,i2,…,vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il,…,vm,1 is not one of the shortest paths from v1,1 to vm,1. This is a contradiction.
The above theorem shows that any part of the shortest path is also the shortest path, so the global shortest path is composed of the local shortest paths of all partitions, and(5)c1,1,m,1=min∑k=1m−1ck,i,k+1,j,where i=1,2,…,nk and j=1,2,…,nk+1. Subsequently, we have(6)c1,1,m,1=min∑k=1pcsk,i,ek,j,where i=1,2,…,nsk and j=1,2,…,nek.
Based on the above equation, it is necessary to calculate the shortest paths between any pair of vertices of the first and last stages for each partition. Specifically, each computing node uses the idea of dynamic programming shown as (2) to solve the local shortest path.
Figure5 presents the flow diagram, and Algorithm 3 presents pseudocode. Note that this algorithm is run by each computing node in parallel. On partition CNk, it produces one of the shortest path from vsk,i to vek,j for each vertex vsk,i∈Vsk and each vertex vek,j∈Vek. In this algorithm, fsk,i,j,l records the previous vertex in the shortest path from vsk,i to vj,l, i.e., vsk,i,…,vj−1,fsk,i,j,l,vj,l is one of the shortest path from vsk,i to vj,l. It calculates the shortest paths in a forward way. Lines 1 to 4 initialize c and f for the vertices in the first stage. Lines 5 to 17 calculate the shortest paths by a triple loop. The outermost loop is for all stages Vii=sk+1,sk+2,…,ek (line 5). The middle loop is for all vertices vi,jj=1,2,…,ni of Vi (line 6), and the innermost loop is for all vertices vsk,ll=1,2,…,nsk of Vsk (line 7). Given vi,j and vsk,l, lines 8 to 14 calculate the shortest path from vsk,l to vi,j, where lines 11 and 12, respectively, update csk,l,i,j and fsk,l,i,j. Lines 18 to 29 generate the shortest paths using f in a backtracking way. Given vsk,j and vek,i, it obtains the vertices’ sequence spsk,j,ek,i starting at vertex vek,i.Figure 5
Flow diagram of local shortest path calculation.Algorithm 3: Local shortest path calculation.
Require: The graph partition CNkEnsure:csk,i,ek,j,spsk,i,ek,j|i=1,2,…,nsk;j=1,2,…,nek⊳list of shortest paths(1)
fori⟵1 to nsk do(2)
csk,i,sk,i⟵0(3)
fsk,i,sk,i⟵0(4)
end for(5)
fori⟵sk+1 to ek do ⊳for each stage(6)
forj⟵1 to ni do ⊳for each vertex of stage Vi(7)
forl⟵1 to nsk do ⊳for each vertex of the first stage Vsk(8)
csk,l,i,j⟵∞(9)
for allvi−1,h,vi,j∈E do(10)
ifcsk,l,i,j>csk,l,i−1,h+wi−1,h,i,j then ⊳a shorter path than before(11)
csk,l,i,j⟵csk,l,i−1,h+wi−1,h,i,j(12)
fsk,l,i,j⟵h(13)
end if(14)
end for(15)
end for(16)
end for(17)
end for(18)
fori⟵1 to nek do ⊳backtrack the shortest paths(19)
forj⟵1 to nsk do(20)
l⟵i(21)
h⟵ek(22)
spsk,j,ek,i⟵vh,l(23)
whileh≠sk do(24)
l⟵fsk,j,h,l(25)
h⟵h−1(26)
spsk,i,ek,j⟵vh,l∪spsk,j,ek,i(27)
end while(28)
end for(29)
end for


### 4.3. Global Shortest Path Calculation

After finding the shortest path of each partition, the global shortest path is calculated by message exchanging among computing nodes. Figure6 depicts a sketch of the merging procedure of local shortest paths. This is an iterative procedure. In each iteration, a pair of computing nodes communicates where the “left” computing node sends its local shortest path to the “right” computing node, and the “right” computing node merges these two local shortest paths. Finally, CNp gets the global shortest path.Figure 6
Local shortest paths merging.Figure7 presents the flow diagram, and Algorithm 4 presents the pseudocode. The set R records the indices of computing nodes participating in subresults’ combination in each iteration. Initially, R contains all computing nodes (line 1). Lines 2 to 5 initialize two sets for each computing node. Lines 6 to 30 calculate the global shortest path by message exchanging among computing nodes with the basic idea as Figure 6. Firstly, the “left” computing nodes CNR2i−1 send SPLR2i−1 to the “right” neighbour computing node CNR2i (lines 7 to 9), and this can be run in parallel for each pair of computing nodes. Secondly, the “right” computing nodes merge the two subresults to get a longer subresult (lines 10 to 28). Given a local shortest path of SPLR2i−1 (line 11) and SPLR2i (line 12), if they can be merged, that is to say, the last vertex of the first path is the same as the starting vertex of the second path (line 13), it tries to merge them to be a longer path from va1,b1 to vc2,d2. If the shortest path does not exist, the algorithm generates one and appends it to SPLR2i′ (lines 14 to 17). If the shortest path exists but it is longer than the current one, the algorithm updates the shortest path (lines 18 to 21). After the two innermost loops (lines 11 to 25), the algorithm replaces SPLR2i with SPLR2i′ (line 26) and sets SPLR2i′ to be empty (line27) to prepare for the next iteration. Line 29 removes the “left” computing nodes from R. Finally CNp returns c1,1,m,1andsp1,1,m,1 as the global shortest path (line 31).Figure 7
Flow diagram of global shortest path calculation.Algorithm 4: Global shortest path calculation.
Require: The local shortest paths of each partitionEnsure: The global shortest path and its costc1,1,m,1,sp1,1,m,1(1)
R⟵1,2,…,p(2)
fork⟵1 to p do ⊳all computing nodes do in parallel(3)
SPLk⟵csk,i,ek,j,spsk,i,ek,j|i=1,2,…,nsk;j=1,2,…,nek(4)
SPLk′⟵∅(5)
end for(6)
whileR>1 do(7)
for allCNR2i−1|i=1,2,…,R/2 do ⊳all computing nodes do in parallel(8)
CNR2i−1 sends SPLR2i−1 to CNR2i(9)
end for(10)
for allCNR2i|i=1,2,…,R/2 do ⊳all computing nodes do in parallel(11)
for allca1,b1,a2,b2,spa1,b1,a2,b2∈SPLR2i−1 do(12)
for allcc1,d1,c2,d2,spc1,d1,c2,d2∈SPLR2i do(13)
ifa2=c1∧b2=d1 then(14)
ifspa1,b1,c2,d2=∅ then(15)
ca1,b1,c2,d2⟵ca1,b1,a2,b2+cc1,d1,c2,d2(16)
spa1,b1,c2,d2⟵spa1,b1,a2,b2∪spc1,d1,c2,d2(17)
SPLR2i′⟵SPLR2i′∪ca1,b1,c2,d2,spa1,b1,c2,d2(18)
else ifca1,b1,c2,d2>ca1,b1,a2,b2+cc1,d1,c2,d2 then(19)
ca1,b1,c2,d2⟵ca1,b1,a2,b2+cc1,d1,c2,d2(20)
spa1,b1,c2,d2⟵spa1,b1,a2,b2∪spc1,d1,c2,d2(21)
Updateca1,b1,c2,d2,spa1,b1,c2,d2∈SPLR2i′ with the new value(22)
end if(23)
end if(24)
end for(25)
end for(26)
SPLR2i⟵SPLR2i′(27)
SPLR2i′⟵∅(28)
end for(29)
R⟵R−R2i−1|i=1,2,…,R/2(30)
end while(31)
CNp returns c1,1,m,1,sp1,1,m,1


### 4.4. An Example

Let us take Figure2 as an example to demonstrate the process of the above algorithm. Given Cap=20, we have p=3 according to (3). Set ML=20 according to (4). Based on Algorithm 2, the graph will be partitioned to 3 partitions, as shown in Figure 8.Figure 8
Partition of the example graph. (a) CN1, 15 edges. (b) CN2, 17 edges. (c) CN3, 15 edges.
(a)(b)(c)Next, each computing node calculates the local shortest path of its partition. Let us takeCN2 as an example to show the process.(1)
Initially,c and f of vertices of stage 4 are set to 0.(2)
The vertices ofV5 calculate c and f. For example, c4,1,5,2=c4,1,4,1+w4,1,5,2=2 and f4,1,5,2=1.(3)
The vertices ofV6 calculate c and f. For example, c4,1,6,2=minc4,1,5,1+w5,1,6,2,c4,1,5,2+w5,2,6,2,c4,1,5,3+w5,3,6,2=min∞,8,∞=8 and f4,1,6,2=2.(4)
The vertices ofV7 calculate c and f. For example, c4,1,7,2=minc4,1,6,1+w6,1,7,2,c4,1,6,2+w6,2,7,2=min16,14=14 and f4,1,7,2=2.(5)
Finally, the vertices ofV7 backtrack to get the shortest paths. For example, to get the shortest path from v4,1 to v7,2, it backtracks to v6,2 in first because f4,1,7,2=2. Secondly, it backtracks to v5,2 and then v4,1 finally. Therefore, the shortest path from v4,1 to v7,2 is v4,1,v5,2,v6,2,v7,2, and the cost is 14.Similar to the above process, eachCN calculates the local shortest paths, and these results are as follows:(1)
SPL1=13,v1,1,v2,2,v3,2,v4,1,13,v1,1,v2,2,v3,3,v4,2,13,v1,1,v2,2,v3,3,v4,3(2)
SPL2=13,v4,1,v5,2,v6,1,v7,1,14,v4,1,v5,2,v6,2,v7,2,12,v4,1,v5,2,v6,1,v7,3,10,v4,1,v5,2,v6,2,v7,4,13,v4,2,v5,1,v6,1,v7,1,16,v4,2,v5,1,v6,1,v7,2,12,v4,2,v5,1,v6,1,v7,3,18,v4,2,v5,3,v6,2,v7,4,17,v4,3,v5,2,v6,1,v7,1,18,v4,3,v5,2,v6,2,v7,2,16,v4,3,v5,2,v6,1,v7,3,14,v4,3,v5,2,v6,2,v7,4(3)
SPL3=12,v7,1,v8,3,v9,1,11,v7,2,v8,3,v9,1,8,v7,3,v8,2,v9,1,12,v7,4,v8,2,v9,1The last step is to merge the subresults of all partitions.CN1 sends SPL1 to CN2 to get the shortest paths from v1,1 to vertices of V7. Obviously, c1,1,7,1=minc1,1,4,1+c4,1,7,1,c1,1,4,2+c4,2,7,1,c1,1,4,3+w4,3,7,1=min26,26,27=26, and sp1,1,7,1=sp1,1,4,1∪sp4,1,7,1. Similarly, c1,1,7,2=27, sp1,1,7,2=sp1,1,4,1∪sp4,1,7,2, c1,1,7,3=25, sp1,1,7,3=sp1,1,4,1∪sp4,1,7,3, c1,1,7,4=23, and sp1,1,7,4=sp1,1,4,1∪sp4,1,7,4.CN2 stores the shortest paths after merging, and it sends the new SPL2 to CN3. CN3 calculates c1,1,9,1=minc1,1,7,1+c7,1,9,1,c1,1,7,2+c7,2,9,1,c1,1,7,3+w7,3,9,1,c1,1,7,4+w7,4,9,1=min38,38,33,35=33 and sp1,1,9,1=sp1,1,7,3∪sp7,3,9,1=v1,1,v2,2,v3,2,v4,1,v5,2,v6,1,v7,3,v8,2,v9,1.


## 4.1. Multistage Graph Partition

In order to determine the graph partition strategy, we should analyse their impacts on the communication overhead after partition. According to the feature of multistage graphs, it is a better scheme to divide the vertices of the same stage into the same partition because it is easy to implement load balance and parallel shortest path solution. SupposeVi and Vi+1 are divided into two different partitions. If we use vertex-cut strategy, the number of mirror vertices is either ni or ni+1. If we use edge-cut strategy, the number of cut edges is Ei. According to (1), ni<Ei and ni+1<Ei, which indicates that the vertex-cut strategy has less communication overhead than edge-cut strategy, so we adopt vertex-cut strategy to partition the graph. Figure 3 is an example. The edge-cut strategy produces 9 cut edges (Figure 3(a)), while the vertex-cut strategy only produces 3 mirror vertices (Figure 3(b)).Figure 3
Two graph partitioning strategies. (a) Edge-cut strategy. (b) Vertex-cut strategy.
(a)(b)Since the multistage graphs studied in this paper are dense, we use the number of edges to represent the load of a partition. Let Cap be the capacity of each computing node, which is also the maximum number of edges that can be stored by a partition, then the number of partitions of a givenG can be estimated as(3)p=ECap.The above equation gives the lower limit of the number of computing nodes. It may lead to the load of the last computing node far lower than those of the other computing nodes. For example, ifE=10100 and Cap=1000, then p=11. If the first 10 computing nodes are fully loaded, then the last computing node only has 100 edges, so the load is imbalanced. Hence, the maximum load of each partition is redefined as(4)ML=Cap,ifp=ECap,γEp,otherwise,where γ is a predefined parameter to keep the load balance for different multistage graphs.The idea of multistage graph partition is to assign the vertices of the same stage to the same partition. Figure4 presents the flow diagram, and Algorithm 2 presents pseudocode. In this algorithm, Sum records the number of edges stored in the current partition. Lines 1 and 2 initialize the variables. Lines 3–16 divide G to a cluster of computing nodes. If Sum≤ML (line 5), lines 6–8 assign the edges of Ei to computing node CNk, and lines 9 and 10, respectively, update ek and i. If Sum>ML (line 11), which means the current partition CNk will be overloaded if we assign the edges of Ei to it, lines 12–14 update variables to prepare for succeeding partition.Figure 4
Flow diagram of graph partition.Algorithm 2: Multistage graph partitioning algorithm.
Require: Multistage graph GEnsure: Partition result of G(1)
Sum⟵0,k⟵1,sk⟵1,i⟵1(2)
CalculateML using (4)(3)
whilei≤m−1 do(4)
Sum⟵Sum+Ei(5)
ifSum≤ML then(6)
for allvi,j,vi+1,l∈Ei do(7)
Assign it toCNk(8)
end for(9)
ek⟵i+1(10)
i⟵i+1(11)
else(12)
Sum⟵0(13)
k⟵k+1(14)
sk⟵i(15)
end if(16)
end while


## 4.2. Local Shortest Path Calculation

After partitioning the graph, each computing node calculates the shortest path of the subgraph stored on it. The shortest path of each partition is referred to as the local shortest path, and the shortest path of the whole graph is referred to as the global shortest path.Theorem 1.
Ifv1,1,v2,i2,…,vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il,…,vm,1 is one of the shortest paths from v1,1 to vm,1, then vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il is one of the shortest paths from vk,ik to vl,il.Proof.
We prove it by using reduction to absurdity. Supposevk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il is not one of the shortest paths from vk,ik to vl,il. There must exist a shortest path from vk,ik to vl,il. Let vk,ik,vk+1,ik+1′,…,vl−1,il−1′,vl,il be one of the shortest paths from vk,ik to vl,il. If we use vk+1,ik+1′,…,vl−1,il−1′ to replace vk+1,ik+1,…,vl−1,il−1 in the path v1,1,v2,i2,…,vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il,…,vm,1, then the cost of v1,1,v2,i2,…,vk,ik,v′k+1,ik+1,…,v′l−1,il−1,vl,il,…,vm,1 is less than v1,1,v2,i2,…,vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il,…,vm,1, so v1,1,v2,i2,…,vk,ik,vk+1,ik+1,…,vl−1,il−1,vl,il,…,vm,1 is not one of the shortest paths from v1,1 to vm,1. This is a contradiction.
The above theorem shows that any part of the shortest path is also the shortest path, so the global shortest path is composed of the local shortest paths of all partitions, and(5)c1,1,m,1=min∑k=1m−1ck,i,k+1,j,where i=1,2,…,nk and j=1,2,…,nk+1. Subsequently, we have(6)c1,1,m,1=min∑k=1pcsk,i,ek,j,where i=1,2,…,nsk and j=1,2,…,nek.
Based on the above equation, it is necessary to calculate the shortest paths between any pair of vertices of the first and last stages for each partition. Specifically, each computing node uses the idea of dynamic programming shown as (2) to solve the local shortest path.
Figure5 presents the flow diagram, and Algorithm 3 presents pseudocode. Note that this algorithm is run by each computing node in parallel. On partition CNk, it produces one of the shortest path from vsk,i to vek,j for each vertex vsk,i∈Vsk and each vertex vek,j∈Vek. In this algorithm, fsk,i,j,l records the previous vertex in the shortest path from vsk,i to vj,l, i.e., vsk,i,…,vj−1,fsk,i,j,l,vj,l is one of the shortest path from vsk,i to vj,l. It calculates the shortest paths in a forward way. Lines 1 to 4 initialize c and f for the vertices in the first stage. Lines 5 to 17 calculate the shortest paths by a triple loop. The outermost loop is for all stages Vii=sk+1,sk+2,…,ek (line 5). The middle loop is for all vertices vi,jj=1,2,…,ni of Vi (line 6), and the innermost loop is for all vertices vsk,ll=1,2,…,nsk of Vsk (line 7). Given vi,j and vsk,l, lines 8 to 14 calculate the shortest path from vsk,l to vi,j, where lines 11 and 12, respectively, update csk,l,i,j and fsk,l,i,j. Lines 18 to 29 generate the shortest paths using f in a backtracking way. Given vsk,j and vek,i, it obtains the vertices’ sequence spsk,j,ek,i starting at vertex vek,i.Figure 5
Flow diagram of local shortest path calculation.Algorithm 3: Local shortest path calculation.
Require: The graph partition CNkEnsure:csk,i,ek,j,spsk,i,ek,j|i=1,2,…,nsk;j=1,2,…,nek⊳list of shortest paths(1)
fori⟵1 to nsk do(2)
csk,i,sk,i⟵0(3)
fsk,i,sk,i⟵0(4)
end for(5)
fori⟵sk+1 to ek do ⊳for each stage(6)
forj⟵1 to ni do ⊳for each vertex of stage Vi(7)
forl⟵1 to nsk do ⊳for each vertex of the first stage Vsk(8)
csk,l,i,j⟵∞(9)
for allvi−1,h,vi,j∈E do(10)
ifcsk,l,i,j>csk,l,i−1,h+wi−1,h,i,j then ⊳a shorter path than before(11)
csk,l,i,j⟵csk,l,i−1,h+wi−1,h,i,j(12)
fsk,l,i,j⟵h(13)
end if(14)
end for(15)
end for(16)
end for(17)
end for(18)
fori⟵1 to nek do ⊳backtrack the shortest paths(19)
forj⟵1 to nsk do(20)
l⟵i(21)
h⟵ek(22)
spsk,j,ek,i⟵vh,l(23)
whileh≠sk do(24)
l⟵fsk,j,h,l(25)
h⟵h−1(26)
spsk,i,ek,j⟵vh,l∪spsk,j,ek,i(27)
end while(28)
end for(29)
end for


## 4.3. Global Shortest Path Calculation

After finding the shortest path of each partition, the global shortest path is calculated by message exchanging among computing nodes. Figure6 depicts a sketch of the merging procedure of local shortest paths. This is an iterative procedure. In each iteration, a pair of computing nodes communicates where the “left” computing node sends its local shortest path to the “right” computing node, and the “right” computing node merges these two local shortest paths. Finally, CNp gets the global shortest path.Figure 6
Local shortest paths merging.Figure7 presents the flow diagram, and Algorithm 4 presents the pseudocode. The set R records the indices of computing nodes participating in subresults’ combination in each iteration. Initially, R contains all computing nodes (line 1). Lines 2 to 5 initialize two sets for each computing node. Lines 6 to 30 calculate the global shortest path by message exchanging among computing nodes with the basic idea as Figure 6. Firstly, the “left” computing nodes CNR2i−1 send SPLR2i−1 to the “right” neighbour computing node CNR2i (lines 7 to 9), and this can be run in parallel for each pair of computing nodes. Secondly, the “right” computing nodes merge the two subresults to get a longer subresult (lines 10 to 28). Given a local shortest path of SPLR2i−1 (line 11) and SPLR2i (line 12), if they can be merged, that is to say, the last vertex of the first path is the same as the starting vertex of the second path (line 13), it tries to merge them to be a longer path from va1,b1 to vc2,d2. If the shortest path does not exist, the algorithm generates one and appends it to SPLR2i′ (lines 14 to 17). If the shortest path exists but it is longer than the current one, the algorithm updates the shortest path (lines 18 to 21). After the two innermost loops (lines 11 to 25), the algorithm replaces SPLR2i with SPLR2i′ (line 26) and sets SPLR2i′ to be empty (line27) to prepare for the next iteration. Line 29 removes the “left” computing nodes from R. Finally CNp returns c1,1,m,1andsp1,1,m,1 as the global shortest path (line 31).Figure 7
Flow diagram of global shortest path calculation.Algorithm 4: Global shortest path calculation.
Require: The local shortest paths of each partitionEnsure: The global shortest path and its costc1,1,m,1,sp1,1,m,1(1)
R⟵1,2,…,p(2)
fork⟵1 to p do ⊳all computing nodes do in parallel(3)
SPLk⟵csk,i,ek,j,spsk,i,ek,j|i=1,2,…,nsk;j=1,2,…,nek(4)
SPLk′⟵∅(5)
end for(6)
whileR>1 do(7)
for allCNR2i−1|i=1,2,…,R/2 do ⊳all computing nodes do in parallel(8)
CNR2i−1 sends SPLR2i−1 to CNR2i(9)
end for(10)
for allCNR2i|i=1,2,…,R/2 do ⊳all computing nodes do in parallel(11)
for allca1,b1,a2,b2,spa1,b1,a2,b2∈SPLR2i−1 do(12)
for allcc1,d1,c2,d2,spc1,d1,c2,d2∈SPLR2i do(13)
ifa2=c1∧b2=d1 then(14)
ifspa1,b1,c2,d2=∅ then(15)
ca1,b1,c2,d2⟵ca1,b1,a2,b2+cc1,d1,c2,d2(16)
spa1,b1,c2,d2⟵spa1,b1,a2,b2∪spc1,d1,c2,d2(17)
SPLR2i′⟵SPLR2i′∪ca1,b1,c2,d2,spa1,b1,c2,d2(18)
else ifca1,b1,c2,d2>ca1,b1,a2,b2+cc1,d1,c2,d2 then(19)
ca1,b1,c2,d2⟵ca1,b1,a2,b2+cc1,d1,c2,d2(20)
spa1,b1,c2,d2⟵spa1,b1,a2,b2∪spc1,d1,c2,d2(21)
Updateca1,b1,c2,d2,spa1,b1,c2,d2∈SPLR2i′ with the new value(22)
end if(23)
end if(24)
end for(25)
end for(26)
SPLR2i⟵SPLR2i′(27)
SPLR2i′⟵∅(28)
end for(29)
R⟵R−R2i−1|i=1,2,…,R/2(30)
end while(31)
CNp returns c1,1,m,1,sp1,1,m,1


## 4.4. An Example

Let us take Figure2 as an example to demonstrate the process of the above algorithm. Given Cap=20, we have p=3 according to (3). Set ML=20 according to (4). Based on Algorithm 2, the graph will be partitioned to 3 partitions, as shown in Figure 8.Figure 8
Partition of the example graph. (a) CN1, 15 edges. (b) CN2, 17 edges. (c) CN3, 15 edges.
(a)(b)(c)Next, each computing node calculates the local shortest path of its partition. Let us takeCN2 as an example to show the process.(1)
Initially,c and f of vertices of stage 4 are set to 0.(2)
The vertices ofV5 calculate c and f. For example, c4,1,5,2=c4,1,4,1+w4,1,5,2=2 and f4,1,5,2=1.(3)
The vertices ofV6 calculate c and f. For example, c4,1,6,2=minc4,1,5,1+w5,1,6,2,c4,1,5,2+w5,2,6,2,c4,1,5,3+w5,3,6,2=min∞,8,∞=8 and f4,1,6,2=2.(4)
The vertices ofV7 calculate c and f. For example, c4,1,7,2=minc4,1,6,1+w6,1,7,2,c4,1,6,2+w6,2,7,2=min16,14=14 and f4,1,7,2=2.(5)
Finally, the vertices ofV7 backtrack to get the shortest paths. For example, to get the shortest path from v4,1 to v7,2, it backtracks to v6,2 in first because f4,1,7,2=2. Secondly, it backtracks to v5,2 and then v4,1 finally. Therefore, the shortest path from v4,1 to v7,2 is v4,1,v5,2,v6,2,v7,2, and the cost is 14.Similar to the above process, eachCN calculates the local shortest paths, and these results are as follows:(1)
SPL1=13,v1,1,v2,2,v3,2,v4,1,13,v1,1,v2,2,v3,3,v4,2,13,v1,1,v2,2,v3,3,v4,3(2)
SPL2=13,v4,1,v5,2,v6,1,v7,1,14,v4,1,v5,2,v6,2,v7,2,12,v4,1,v5,2,v6,1,v7,3,10,v4,1,v5,2,v6,2,v7,4,13,v4,2,v5,1,v6,1,v7,1,16,v4,2,v5,1,v6,1,v7,2,12,v4,2,v5,1,v6,1,v7,3,18,v4,2,v5,3,v6,2,v7,4,17,v4,3,v5,2,v6,1,v7,1,18,v4,3,v5,2,v6,2,v7,2,16,v4,3,v5,2,v6,1,v7,3,14,v4,3,v5,2,v6,2,v7,4(3)
SPL3=12,v7,1,v8,3,v9,1,11,v7,2,v8,3,v9,1,8,v7,3,v8,2,v9,1,12,v7,4,v8,2,v9,1The last step is to merge the subresults of all partitions.CN1 sends SPL1 to CN2 to get the shortest paths from v1,1 to vertices of V7. Obviously, c1,1,7,1=minc1,1,4,1+c4,1,7,1,c1,1,4,2+c4,2,7,1,c1,1,4,3+w4,3,7,1=min26,26,27=26, and sp1,1,7,1=sp1,1,4,1∪sp4,1,7,1. Similarly, c1,1,7,2=27, sp1,1,7,2=sp1,1,4,1∪sp4,1,7,2, c1,1,7,3=25, sp1,1,7,3=sp1,1,4,1∪sp4,1,7,3, c1,1,7,4=23, and sp1,1,7,4=sp1,1,4,1∪sp4,1,7,4.CN2 stores the shortest paths after merging, and it sends the new SPL2 to CN3. CN3 calculates c1,1,9,1=minc1,1,7,1+c7,1,9,1,c1,1,7,2+c7,2,9,1,c1,1,7,3+w7,3,9,1,c1,1,7,4+w7,4,9,1=min38,38,33,35=33 and sp1,1,9,1=sp1,1,7,3∪sp7,3,9,1=v1,1,v2,2,v3,2,v4,1,v5,2,v6,1,v7,3,v8,2,v9,1.


## 5. Experiments and Analysis

### 5.1. Experimental Setup

Because there are no public multistage graph datasets, we synthesize 5 datasets using Java on IntelliJ IDEA, where the number of vertices of each stage and the weights of edges are random values satisfying (1). Table 2 presents the basic data of these 5 datasets.Table 2
Multistage graph datasets.
GraphNumber of verticesNumber of edgesNumber of stagesGraph_12000505222Graph_270022100812Graph_340002199702500Graph_41000004997001000Graph_51000020500871010000The shortest paths algorithms are run on Hadoop [36] in conjunction with the Spark [37] computing engine. Spark is a fast and general-purpose computing engine designed for large-scale data processing [38]. The cluster consists of 8 computers, and each computer has a 4-core Intel processor, 8 GB memory, and 1 TB storage. The operating system is CentOS 7, and the distributed environment is built using Hadoop2.7 and Spark2.1.In order to compare the performance of DMGA with existing algorithms, all graphs are partitioned to 8 partitions, which is a little different from Algorithm2 whose number of partitions depends on the scale of the graph. Hence, the ML of each partition is(7)ML=γE8,in the experiments, and line 2 uses (7) to replace (4) in Algorithm 2.


### 5.2. Experimental Results

#### 5.2.1. Partitioning Quality

At present, there is no partition algorithm for multistage graphs, so we compare the partition quality of the DMGA algorithm with the Hash partitioning algorithm. Hash is the default partitioning algorithm in many distributed graph processing systems, and it is the basis of most of existing distributed algorithms for solving the shortest path.For the vertex-cut partitioning method, the number of mirror vertices reflects the communication overhead. The fewer the mirror vertices, the less the communication overhead, and the corresponding calculation time will be reduced. Figure9 shows the number of mirror vertices generated by two graph partitioning algorithms running on the above 5 datasets. It can be seen from the results that the number of mirror vertices of the two graph partitioning algorithms for small datasets is almost the same. With the increase of the scale of the dataset and the number of stages, the number of mirror vertices of the DMGA algorithm is significantly less than the Hash algorithm. Specifically, the number of mirror vertices produced by the Hash algorithm for Graph_3, Graph_4, and Graph_5 is, respectively, 37.25, 24.97, and 69.9 times of the DMGA algorithm.Figure 9
Number of mirror vertices.This shows that the DMGA algorithm has a better partitioning result for multistage graphs than the Hash algorithm. It can be deduced that the number of mirror vertices will be reduced in further if (4) instead of (7) is used to guide partition in Algorithm 2.For the homogeneous cluster which consists of computing nodes with the same configuration, the load of each computing node should be balanced as much as possible in order to reduce the calculation time. Table3 shows the number of edges on each computing node of each dataset generated by two graph partitioning algorithms. The Hash algorithm partitions the graph according to the Hash function defined as(8)k=∑l=1i−1nl+jmod8,which means the vertex vi,j is assigned to partition CNk. Thus, the load of each partition is almost the same. The DMGA algorithm assigns all vertices in the same stage to the same partition, and the numbers of vertices in each stage are different, so the edge distribution is not balanced.Table 3
Edge distribution of graphs on computing nodes.
CNGraph_1Graph_2Graph_3Graph_4Graph_5HashDMGAHashDMGAHashDMGAHashDMGAHashDMGA1658623255322062513825109623956259562504962057327267562835322625139260126267760501625274605519357951226612723251452453562279641016353716461014613758254430212507826105620396120562832461205656155322579266924901251256290163632627591636302661871825722845245902401962410616146194326216147617524263623892483622016628366389462385363909486266292628192924875267816216362158623816627451The average deviation and standard deviation are further analysed to check the loads among partitions, and they are shown in Figure10. In this figure, AVEDEV and STDEV, respectively, represent average deviation and standard deviation. We can see that the average and standard deviations of DMGA are 2.3 to 7.5 times of those of Hash. Graph_3 has the maximum difference among all graphs: the average deviation of DMGA is 6.65 times of that of Hash, and the standard deviation of DMGA is 7.51 times of that of Hash. Graph_1 has the minimal difference among all graphs: the average deviation of DMGA is 2.79 times of that of Hash, and the standard deviation of DMGA is 2.36 times of that of Hash. These results also show that Hash can produce more balanced partition.Figure 10
Average and standard deviations of edge distribution.


#### 5.2.2. Comparison of Running Time

We compare the execution time of four algorithms: single-machine, DMGA, SSSP [7] algorithm embedded in Pregel on Spark, and parallel Dijkstra [39–41]. The single-machine algorithm utilizes the sequential dynamic programming algorithm. The experiments are run 10 times, and the results are the average over these runs.Table4 presents the results, and Figure 11 shows the graphical comparison. It can be seen that DMGA has a higher speedup ratio and time superiority over the sequential algorithm as the scale of dataset increases.(i)
Graph_1: the sequential algorithm is faster than DMGA because the communication among computing nodes of DMGA takes up a lot of time.(ii)
Graph_2: the running time of these two algorithms is almost the same.(iii)
Graph_3 and Graph 4: the running time of the sequential algorithm is, respectively, 1.72 and 3.13 times of that of DMGA.(iv)
Graph_5: the sequential algorithm meets “memory overflow” error, so this graph cannot be solved by a single machine. On the contrary, DMGA can obtain the shortest path.Table 4
Running time of four algorithms (unit: s).
GraphSingle machineDMGASSSPParallel DijkstraGraph_14.065.568.56101.57Graph_217.9115.5211.14452.36Graph_359.1334.461095.232642.58Graph_4381.56122.061856.5915893.45Graph_5–7564.7029324.7385481.36Figure 11
Comparison of running time of four algorithms.In further, parallel Dijkstra has a relatively longer running time than the DMGA and SSSP, due to its high time and space complexity. SSSP uses Pregel’s default graph partition method which does not take the structural feature of multistage graphs into account, so it has a large amount of communication. SSSP needs longer time for the larger scale graph, which implies that the communication overhead increases significantly with the increase of the scale of the graph specifically.(i)
Graph_2: SSSP has the least running time, and DMGA only needs 4 seconds longer than SSSP, but parallel Dijkstra needs more than 30 times of that of DMGA.(ii)
Other graphs: DMGA needs the least running time. The larger the scale of the graph, the more obvious the advantage of DMGA. For example, the running time of DMGA is, respectively, 25.8% and 8.8% of that of SSSP and parallel Dijkstra for Graph_5.The above results show that DMGA makes full use of the special structural characteristic of multistage graphs, and it has extremely low communication overhead.


## 5.1. Experimental Setup

Because there are no public multistage graph datasets, we synthesize 5 datasets using Java on IntelliJ IDEA, where the number of vertices of each stage and the weights of edges are random values satisfying (1). Table 2 presents the basic data of these 5 datasets.Table 2
Multistage graph datasets.
GraphNumber of verticesNumber of edgesNumber of stagesGraph_12000505222Graph_270022100812Graph_340002199702500Graph_41000004997001000Graph_51000020500871010000The shortest paths algorithms are run on Hadoop [36] in conjunction with the Spark [37] computing engine. Spark is a fast and general-purpose computing engine designed for large-scale data processing [38]. The cluster consists of 8 computers, and each computer has a 4-core Intel processor, 8 GB memory, and 1 TB storage. The operating system is CentOS 7, and the distributed environment is built using Hadoop2.7 and Spark2.1.In order to compare the performance of DMGA with existing algorithms, all graphs are partitioned to 8 partitions, which is a little different from Algorithm2 whose number of partitions depends on the scale of the graph. Hence, the ML of each partition is(7)ML=γE8,in the experiments, and line 2 uses (7) to replace (4) in Algorithm 2.


## 5.2. Experimental Results

### 5.2.1. Partitioning Quality

At present, there is no partition algorithm for multistage graphs, so we compare the partition quality of the DMGA algorithm with the Hash partitioning algorithm. Hash is the default partitioning algorithm in many distributed graph processing systems, and it is the basis of most of existing distributed algorithms for solving the shortest path.For the vertex-cut partitioning method, the number of mirror vertices reflects the communication overhead. The fewer the mirror vertices, the less the communication overhead, and the corresponding calculation time will be reduced. Figure9 shows the number of mirror vertices generated by two graph partitioning algorithms running on the above 5 datasets. It can be seen from the results that the number of mirror vertices of the two graph partitioning algorithms for small datasets is almost the same. With the increase of the scale of the dataset and the number of stages, the number of mirror vertices of the DMGA algorithm is significantly less than the Hash algorithm. Specifically, the number of mirror vertices produced by the Hash algorithm for Graph_3, Graph_4, and Graph_5 is, respectively, 37.25, 24.97, and 69.9 times of the DMGA algorithm.Figure 9
Number of mirror vertices.This shows that the DMGA algorithm has a better partitioning result for multistage graphs than the Hash algorithm. It can be deduced that the number of mirror vertices will be reduced in further if (4) instead of (7) is used to guide partition in Algorithm 2.For the homogeneous cluster which consists of computing nodes with the same configuration, the load of each computing node should be balanced as much as possible in order to reduce the calculation time. Table3 shows the number of edges on each computing node of each dataset generated by two graph partitioning algorithms. The Hash algorithm partitions the graph according to the Hash function defined as(8)k=∑l=1i−1nl+jmod8,which means the vertex vi,j is assigned to partition CNk. Thus, the load of each partition is almost the same. The DMGA algorithm assigns all vertices in the same stage to the same partition, and the numbers of vertices in each stage are different, so the edge distribution is not balanced.Table 3
Edge distribution of graphs on computing nodes.
CNGraph_1Graph_2Graph_3Graph_4Graph_5HashDMGAHashDMGAHashDMGAHashDMGAHashDMGA1658623255322062513825109623956259562504962057327267562835322625139260126267760501625274605519357951226612723251452453562279641016353716461014613758254430212507826105620396120562832461205656155322579266924901251256290163632627591636302661871825722845245902401962410616146194326216147617524263623892483622016628366389462385363909486266292628192924875267816216362158623816627451The average deviation and standard deviation are further analysed to check the loads among partitions, and they are shown in Figure10. In this figure, AVEDEV and STDEV, respectively, represent average deviation and standard deviation. We can see that the average and standard deviations of DMGA are 2.3 to 7.5 times of those of Hash. Graph_3 has the maximum difference among all graphs: the average deviation of DMGA is 6.65 times of that of Hash, and the standard deviation of DMGA is 7.51 times of that of Hash. Graph_1 has the minimal difference among all graphs: the average deviation of DMGA is 2.79 times of that of Hash, and the standard deviation of DMGA is 2.36 times of that of Hash. These results also show that Hash can produce more balanced partition.Figure 10
Average and standard deviations of edge distribution.


### 5.2.2. Comparison of Running Time

We compare the execution time of four algorithms: single-machine, DMGA, SSSP [7] algorithm embedded in Pregel on Spark, and parallel Dijkstra [39–41]. The single-machine algorithm utilizes the sequential dynamic programming algorithm. The experiments are run 10 times, and the results are the average over these runs.Table4 presents the results, and Figure 11 shows the graphical comparison. It can be seen that DMGA has a higher speedup ratio and time superiority over the sequential algorithm as the scale of dataset increases.(i)
Graph_1: the sequential algorithm is faster than DMGA because the communication among computing nodes of DMGA takes up a lot of time.(ii)
Graph_2: the running time of these two algorithms is almost the same.(iii)
Graph_3 and Graph 4: the running time of the sequential algorithm is, respectively, 1.72 and 3.13 times of that of DMGA.(iv)
Graph_5: the sequential algorithm meets “memory overflow” error, so this graph cannot be solved by a single machine. On the contrary, DMGA can obtain the shortest path.Table 4
Running time of four algorithms (unit: s).
GraphSingle machineDMGASSSPParallel DijkstraGraph_14.065.568.56101.57Graph_217.9115.5211.14452.36Graph_359.1334.461095.232642.58Graph_4381.56122.061856.5915893.45Graph_5–7564.7029324.7385481.36Figure 11
Comparison of running time of four algorithms.In further, parallel Dijkstra has a relatively longer running time than the DMGA and SSSP, due to its high time and space complexity. SSSP uses Pregel’s default graph partition method which does not take the structural feature of multistage graphs into account, so it has a large amount of communication. SSSP needs longer time for the larger scale graph, which implies that the communication overhead increases significantly with the increase of the scale of the graph specifically.(i)
Graph_2: SSSP has the least running time, and DMGA only needs 4 seconds longer than SSSP, but parallel Dijkstra needs more than 30 times of that of DMGA.(ii)
Other graphs: DMGA needs the least running time. The larger the scale of the graph, the more obvious the advantage of DMGA. For example, the running time of DMGA is, respectively, 25.8% and 8.8% of that of SSSP and parallel Dijkstra for Graph_5.The above results show that DMGA makes full use of the special structural characteristic of multistage graphs, and it has extremely low communication overhead.


## 5.2.1. Partitioning Quality

At present, there is no partition algorithm for multistage graphs, so we compare the partition quality of the DMGA algorithm with the Hash partitioning algorithm. Hash is the default partitioning algorithm in many distributed graph processing systems, and it is the basis of most of existing distributed algorithms for solving the shortest path.For the vertex-cut partitioning method, the number of mirror vertices reflects the communication overhead. The fewer the mirror vertices, the less the communication overhead, and the corresponding calculation time will be reduced. Figure9 shows the number of mirror vertices generated by two graph partitioning algorithms running on the above 5 datasets. It can be seen from the results that the number of mirror vertices of the two graph partitioning algorithms for small datasets is almost the same. With the increase of the scale of the dataset and the number of stages, the number of mirror vertices of the DMGA algorithm is significantly less than the Hash algorithm. Specifically, the number of mirror vertices produced by the Hash algorithm for Graph_3, Graph_4, and Graph_5 is, respectively, 37.25, 24.97, and 69.9 times of the DMGA algorithm.Figure 9
Number of mirror vertices.This shows that the DMGA algorithm has a better partitioning result for multistage graphs than the Hash algorithm. It can be deduced that the number of mirror vertices will be reduced in further if (4) instead of (7) is used to guide partition in Algorithm 2.For the homogeneous cluster which consists of computing nodes with the same configuration, the load of each computing node should be balanced as much as possible in order to reduce the calculation time. Table3 shows the number of edges on each computing node of each dataset generated by two graph partitioning algorithms. The Hash algorithm partitions the graph according to the Hash function defined as(8)k=∑l=1i−1nl+jmod8,which means the vertex vi,j is assigned to partition CNk. Thus, the load of each partition is almost the same. The DMGA algorithm assigns all vertices in the same stage to the same partition, and the numbers of vertices in each stage are different, so the edge distribution is not balanced.Table 3
Edge distribution of graphs on computing nodes.
CNGraph_1Graph_2Graph_3Graph_4Graph_5HashDMGAHashDMGAHashDMGAHashDMGAHashDMGA1658623255322062513825109623956259562504962057327267562835322625139260126267760501625274605519357951226612723251452453562279641016353716461014613758254430212507826105620396120562832461205656155322579266924901251256290163632627591636302661871825722845245902401962410616146194326216147617524263623892483622016628366389462385363909486266292628192924875267816216362158623816627451The average deviation and standard deviation are further analysed to check the loads among partitions, and they are shown in Figure10. In this figure, AVEDEV and STDEV, respectively, represent average deviation and standard deviation. We can see that the average and standard deviations of DMGA are 2.3 to 7.5 times of those of Hash. Graph_3 has the maximum difference among all graphs: the average deviation of DMGA is 6.65 times of that of Hash, and the standard deviation of DMGA is 7.51 times of that of Hash. Graph_1 has the minimal difference among all graphs: the average deviation of DMGA is 2.79 times of that of Hash, and the standard deviation of DMGA is 2.36 times of that of Hash. These results also show that Hash can produce more balanced partition.Figure 10
Average and standard deviations of edge distribution.


## 5.2.2. Comparison of Running Time

We compare the execution time of four algorithms: single-machine, DMGA, SSSP [7] algorithm embedded in Pregel on Spark, and parallel Dijkstra [39–41]. The single-machine algorithm utilizes the sequential dynamic programming algorithm. The experiments are run 10 times, and the results are the average over these runs.Table4 presents the results, and Figure 11 shows the graphical comparison. It can be seen that DMGA has a higher speedup ratio and time superiority over the sequential algorithm as the scale of dataset increases.(i)
Graph_1: the sequential algorithm is faster than DMGA because the communication among computing nodes of DMGA takes up a lot of time.(ii)
Graph_2: the running time of these two algorithms is almost the same.(iii)
Graph_3 and Graph 4: the running time of the sequential algorithm is, respectively, 1.72 and 3.13 times of that of DMGA.(iv)
Graph_5: the sequential algorithm meets “memory overflow” error, so this graph cannot be solved by a single machine. On the contrary, DMGA can obtain the shortest path.Table 4
Running time of four algorithms (unit: s).
GraphSingle machineDMGASSSPParallel DijkstraGraph_14.065.568.56101.57Graph_217.9115.5211.14452.36Graph_359.1334.461095.232642.58Graph_4381.56122.061856.5915893.45Graph_5–7564.7029324.7385481.36Figure 11
Comparison of running time of four algorithms.In further, parallel Dijkstra has a relatively longer running time than the DMGA and SSSP, due to its high time and space complexity. SSSP uses Pregel’s default graph partition method which does not take the structural feature of multistage graphs into account, so it has a large amount of communication. SSSP needs longer time for the larger scale graph, which implies that the communication overhead increases significantly with the increase of the scale of the graph specifically.(i)
Graph_2: SSSP has the least running time, and DMGA only needs 4 seconds longer than SSSP, but parallel Dijkstra needs more than 30 times of that of DMGA.(ii)
Other graphs: DMGA needs the least running time. The larger the scale of the graph, the more obvious the advantage of DMGA. For example, the running time of DMGA is, respectively, 25.8% and 8.8% of that of SSSP and parallel Dijkstra for Graph_5.The above results show that DMGA makes full use of the special structural characteristic of multistage graphs, and it has extremely low communication overhead.


## 6. Conclusion and Future Work

Nowadays, graph models are widely applied in many fields, and the scale of the graph increases significantly. The existing distributed graph computing systems cannot make full use of the special characteristics of the multistage graphs. To this end, this paper proposes DMGA which is used to solve the shortest path problem of large-scale multistage graphs on a distributed computing system. DMGA consists of three phases: graph partition, local shortest path calculation, and global shortest path calculation. The experiment results demonstrate its high-performance. However, the load of DMGA is not balanced, and it only considers of the vertex-cut partitioning method. In future, we will focus on reducing the number of mirror vertices and solution time as much as possible under the premise of load balance and propose the special algorithm based on edge-cut partitioning idea.



---

*Source: 6639008-2021-06-01.xml*