Layph: Making Change Propagation Constraint in Incremental Graph Processing by Layering Graph

Given an input graph $G=(V,E)$, where $V$ is a finite set of vertices and $E\subseteq V\times V$ is a set of edges. The weight of each edge $(u,v)\in E$ is $w_{u,v}$ in a weighted graph or a consistent value 1 in an unweighted graph. In general, an iterative graph algorithm $\mathcal{A}$ that executes in an accumulative model, includes two types of operations, i.e., message generation $\mathcal{F}$ and message aggregation $\mathcal{G}$ [9, 10, 20].

where $m^{i-1}_{u}=\mathcal{G}(\{m^{i-1}_{*,u}|(*,u)\in E\})$.

The message generate operation $\mathcal{F}$ applied on each vertex $u\in V$ prepares the message $m_{u,v}$ for each outgoing edge $(u,v)$ based on the aggregation of received message $m_{u}$ and the edge weight $w_{u,v}$. The aggregation operation $\mathcal{G}$ is applied on each destination vertex $v$. It first aggregates the messages that terminate at $v$ to obtain a new message $m_{v}$, then aggregates the old vertex state $x_{v}$ and the aggregated message $m_{v}$ to update the vertex state $x_{v}$. The two-step process is applied iteratively till convergence (when vertex states become stable). To sum up, an iterative graph computation can be expressed as $\mathcal{A}$$=$$(\mathcal{F},\mathcal{G},X^{0},M^{0})$ where $\mathcal{F}$ and $\mathcal{G}$ are the operations that specify the algorithm logic, and $X^{0}$$=$$\{x_{v}^{0}$$\mid$$v\in V\}$ and $M^{0}$$=$$\{m_{v}^{0}$$\mid$$v\in V\}$ are the initial values of vertex states and root messages respectively. A graph computation on the input graph $G$ can be denoted as $\mathcal{A}(G)$.

Suppose $\mathcal{A}$ can be executed asynchronously, then it can be expressed as Equation (1) naturally, such as $\mathsf{SSSP}$. Otherwise, the synchronous algorithms should be rewritten in accumulative mode and executed asynchronously, such as PageRank. There are some efforts [10, 9] that rewrite a synchronous algorithm in asynchronous accumulative mode.

Equation (1) defines the vertex-centric format of asynchronous iterative computation. On this basis, we can define a set-based iterative computation as follows

$X=\{x_{v}\mid v\in V\}$ is the set of vertex states. $M^{0}=\{m^{0}_{v}\mid v\in V\}$ is the set of root messages of each vertex and $M^{k\neq 0}=\{m^{k}_{v}\}$ is the set of generated messages on all edges. It should be noticed that these are slight meaning changes of $\mathcal{F}$ and $\mathcal{G}$ in set-based format. $\mathcal{F}$ is the message generate operation with edge information embedded so it only needs a single parameter $M$. $\mathcal{G}$ is the group-by aggregator (group by vertex id). Based on this set-based computation, the vertex states set $X$ after $n$ iterations is

where $\mathcal{G}\circ\mathcal{F}(\cdot)=\mathcal{G}(\mathcal{F}(\cdot))$ and $(\mathcal{G}\circ\mathcal{F})^{k}$ denotes $k$ applications of $(\mathcal{G}\circ\mathcal{F})$.

Given an iterative graph computation $\mathcal{A}$ and its incremental counterpart $\mathcal{I}_{\mathcal{A}}$, the problem of incremental computation arises when the input graph $G$ is updated with $\Delta G$. Let $\mathcal{A}(G)$ denote the output of an old graph $G$ with the effect of batch graph algorithm $\mathcal{A}$. The inputs of incremental computation include $\mathcal{A}(G)$ and graph updates $\Delta G$. Then we have

It means that the incremental computation $\mathcal{I}_{\mathcal{A}}(\mathcal{A}(G),\Delta G)$ that is performed based on the old result $\mathcal{A}(G)$ and the graph updates $\Delta G$ is expected to output $\mathcal{A}(G\oplus\Delta G)$, where $G\oplus\Delta G$ denotes applying the updates $\Delta G$ to $G$. It is noticeable that the incrementalization scheme $\mathcal{I}_{\mathcal{A}}$ is algorithm-specific and is deduced from its original algorithm $\mathcal{A}$.

The input batch update $\Delta G$ consists of a set of unit updates. To simplify our discussion, we consider the insertion or deletion of a single edge as a unit update in a sequence, which can simulate certain modifications. For instance, each change to an edge weight can be considered as deleting the edge and followed by adding another edge with the new property. The incremental computation $\mathcal{I}_{\mathcal{A}}$ will identify the changes to the old output $\mathcal{A}(G)$ and make corrections of the previous computation in response to $\Delta G$.

In this section, we first introduce how to extract vertices on the upper layer. Then we provide an automated shortcut calculation method.

The layered graph needs to be updated when $G$ is updated with $\Delta G$. The vertices may move between the two layers, due to the generation or disappearance of dense subgraphs, e.g., the internal vertices of the newly generated subgraph move from $L_{up}$ to $L_{low}$. In order to avoid the expensive overhead caused by repeated subgraph discovery, we incrementally update the dense subgraphs with incremental community detection methods, such as C-Blondel [22] or DynaMo [23]. In practice, the size of $\Delta G$ is very small compared with $G$. A small $\Delta G$ does not have a large effect on existing dense subgraphs. Thus we update the dense subgraphs only when enough $\Delta G$ are accumulated. However, even a very small $\Delta G$ can still change the weight of a number of shortcuts of the layered graph.

The shortcut is built inside each dense subgraph according to the Definition 3. Moreover, from the Equation (6), we can see that the weight of each shortcut on $G_{i}$ only depends on the edges and vertices in $G_{i}$, and the shortcuts on the different subgraphs are independent of each other. Therefore, we only need to update the shortcuts on the subgraphs affected by $\Delta G$, and the shortcuts for each subgraph can be updated in parallel. For the shortcut deletion or addition, they can be done directly within the subgraph by removing or calculating the shortcut. For the weight update, in order to avoid redundant computation, we use an incremental method to update.

According to Equation (6), the weight of the shortcuts is calculated by iterative computations, and the weight of the shortcut from $u$ to $v$ is equal to the aggregate all the messages received by $v$ through all paths from $u$ to $v$. After the edge addition or deletion within the dense subgraph, some messages received by $v$ may become invalid or missing. Thus, the update of the shortcut can adopt the existing incremental computation methods [16, 15, 14]. The compensation and cancellation messages can be deduced based on the memoized information when calculating the old shortcut. These messages will be used to redo and undo the effect of missing and invalid messages on vertex $v$, in which there are some missing and invalid messages in the received messages of $v$ due to the addition and deletion edges within the dense subgraph.

The upper layer $L_{up}$ only contains a subset of vertices, and the internal vertices inside each subgraph on $L_{low}$ do not participate in iterative computation on $L_{up}$. To ensure that all vertices on $L_{up}$ converge with the effects of internal vertices, the iterative computation on $L_{up}$ should collect not only the revision messages deduced by the vertices on $L_{up}$ but also those by internal vertices. Thus, it is required to upload the revision messages deduced by the internal vertices of the dense subgraphs on $L_{low}$ to $L_{up}$. Since the entry/exit vertices of each dense subgraph are on $L_{up}$, messages upload can be done by propagating the revision messages to entry/exit vertices.

Not all the internal vertices within each dense subgraph have connections with the entry/exit vertices, thus, we perform a local iterative computation to propagate internal revision messages to the entry/exit vertices of the subgraph. The iterative computation terminates when the messages received by entry/exit vertices can be ignored. After the upload of the messages, the accumulated messages on the entry vertices $V_{i}^{I}$ and exit vertices $V_{i}^{O}$ can be treated as their initial revision messages i.e.,, $\mathbb{M}^{0}_{V_{i}^{I}\cup V_{i}^{O}}=\mathcal{G}_{V_{i}^{I}\cup V_{i}^{O}}\big{(}\bigcup_{k=1}^{\infty}(\mathcal{G}_{V_{i}}\circ\mathcal{F}_{E_{i}})^{k}(\mathbb{M}^{0}_{V_{i}})\big{)}$. Together with the initial messages of vertices that are not in any dense subgraph on $L^{V}_{up}$, i.e., $\mathbb{M}^{0}_{V-\cup_{i=1}^{N}V_{i}}$, the initial messages of vertices on $L_{up}$ can be expressed as follows

where $\mathbb{M}^{0}_{V_{i}}$ represents the initial revision messages.

After the upload of the messages, the revision messages deduced by internal vertices of the subgraphs on $L_{low}$ have been propagated to $L_{up}$. However, these uploaded messages are only cached in entry and exit vertices of the dense subgraphs according to Equation (7). Iterative computations are required to be performed on $L_{up}$ to propagate the revision messages so that the other vertices on $L_{up}$ can receive all the revision messages to revise their states.

The iterative computations only perform on $L_{up}$, i.e., only $L^{V}_{up}$ and $L^{E}_{up}$ are involved in iterative computations, and the entry and exit vertices of dense subgraphs will participate in the iterative computations because they are on $L^{V}_{up}$. When the entry vertices receive messages, they do not send messages to internal vertices, but propagate messages to exit vertices via shortcuts. After the iterative computations, the states of vertices on $L_{up}$ can be expressed as follows

Based on the following Theorem V-B, We can see that after the iterative computations on $L_{up}$, the vertices converge to the same state as performing the iterative computation on the original graph.

Since the iterative computation is only performed on $L_{up}$, the revision messages will not touch the internal vertices of each subgraph on $L_{low}$, i.e.,, the internal vertices will not receive revision messages from outside. It is essential to launch another step to apply outside messages to internal vertices.

Though the internal vertices do not receive the revision messages from outside, the entry vertices of each dense subgraph have received all the revision messages from vertices in other dense subgraphs and $L_{up}$ according to Theorem V-B. In order to enable the internal vertices to receive outside revision messages, the entry vertices cache the received messages before propagating them to exit vertices via shortcuts during the iterative computations. After many iterations, the entry vertices may cache a large number of messages, and we only store the aggregated messages. The cached messages can be expressed as follows

Finally, we send the messages that have been cached in entry vertices to the internal vertices via shortcuts between entry vertices and internal vertices. The states of the vertices on $L_{low}$ can be expressed as follows

where $\hat{S}_{i}=\{\vec{w}_{u,v}\in S_{i}\mid u\in V_{i}^{I},v\in\hat{V}_{i}\}$ is a set of shortcuts between two layers, $X_{L_{low}^{V}}$ are vertex states on $L_{low}$ after local iterative computation for uploading revision messages to vertices of $L_{up}$, i.e.,

We have the following theorem to guarantee correctness.

We use AliCloud ecs.r6.13xlarge instance (52vCPU, 384GB memory, 64-bit Ubuntu 18.04 with compiler GCC 7.5) for these experiments.

We first compare $\mathsf{Layph}$ with the competitors in response time of each workload executed on different datasets. The Normalized results are reported in Figure 5, where the response time of $\mathsf{Layph}$ is treated as the baseline, i.e., $\mathsf{Layph}$ finishes in unit time 1. In particular, Figure 5e reports the response time for processing vertex updates, while the rest is used for edge updates. We can see that the improvement in handling vertex changes in $\mathsf{Layph}$ is consistent with the improvement in handling edge changes. When updating vertices, the other systems meet runtime errors, thus we only compare Ingress with $\mathsf{Layph}$. It is shown that $\mathsf{Layph}$ consistently outperforms others in most cases. Specifically, $\mathsf{Layph}$ achieves 3.13-15.82$\times$ (8.49$\times$ on average) speedup over KickStarter, 2.54-8.49$\times$ (4.49$\times$ on average) speedup over RisGraph, 2.99-36.66$\times$ (18.99$\times$ on average) speedup over GraphBolt, 2.92-32.93$\times$ (17.53$\times$ on average) speedup over DZiG, and 1.06-7.22$\times$ (2.54$\times$ on average) speedup over Ingress. To explain the reason for the above results, we also report the total number of edge activations in Figure 6. An edge activation represents an $\mathcal{F}$ operation. In most graph workloads, the cost of $\mathcal{F}$ is much greater than that of $\mathcal{G}$ operation, because the number of $\mathcal{F}$ and the unit cost of $\mathcal{F}$ are often both larger than that of $\mathcal{G}$. From Figure 5 and Figure 6, we can see that the normalized number of edge activations is a similar trend to the normalized response time of each system.

Regarding $\mathsf{SSSP}$ and $\mathsf{BFS}$, RisGraph is faster than KickStarter since it allows more parallelism during incremental updates and allows for localized data access. Ingress and RisGraph are comparable because the memoization-path engine in Ingress follows a similar idea. $\mathsf{Layph}$ outperforms the other competitors by leveraging the layered graph. Note that, when performing BFS on WB, RigGraph is slower than $\mathsf{Layph}$ but with fewer edge activations. This is because that RisGraph can identify the safe and unsafe updates to reduce edge activations. It just so happens that most of the updates on WB are safe for $\mathsf{BFS}$. However, the additional cost of identifying the safe or unsafe is relatively expensive since WB is very small. While in SSSP, compared with Ingress, $\mathsf{Layph}$ also requires less response time but with more edge activations. This is because there are some large dense subgraphs in WB, requiring more shortcut updates, which increase the number of edge activations. Since $\mathsf{Layph}$ is parallel-friendly for shortcut updates, it will only have a small effect on the response time.

Regarding $\mathsf{PageRank}$ and $\mathsf{PHP}$, DZiG is faster than GraphBolt since DZiG has a sparsity detection mechanism, based on which it can adjust the incremental computation scheme. Besides, Ingress is faster than DZiG and GraphBolt. This can be attributed to its memoization-free engine which is more efficient than others. $\mathsf{Layph}$ is built on top of Ingress, and can further limit the iterative computation scope with the layered graph, which reduces the number of activation edges, as shown in Figure 6. We find that $\mathsf{Layph}$ exhibits less improvement on WB. The reason is that the subgraphs in WB are much larger than that in other graphs, which increases costs and weakens gains. The reason will be further explained in Section VI-F.

During incremental computation, $\mathsf{Layph}$ consists of four phases: the layered graph update, revision messages upload, iterative computation on the upper layer, and messages assignment. To study the time spent in each phase, we run four algorithms on UK and record the runtime of each phase. The proportion of runtime for different phases is shown in Figure 8. We can see that the iterative computation takes up most of the runtime. The messages assignment is the second most expensive phase. The layered graph update and revision messages upload are both very fast except in $\mathsf{PHP}$. This is because the iterative computation of $\mathsf{PHP}$ is very fast, say 418 ms, which makes those two phases relatively longer. The results indicate that the additional cost in our system, i.e., the maintenance of the layered graph, is lightweight. Based on the above experimental analysis, it is worth adopting the layered graph in incremental graph processing.

We vary the number of execution threads from 1 to 32 to see the runtime reduction. We run $\mathsf{SSSP}$ on UK and compare $\mathsf{Layph}$ with KickStarter, RisGraph, and Ingress. The results are shown in Figure 9a. We can see that as the threads increase, the runtime decreases steadily in all systems as expected. The reduction is smoother when the number of threads is larger than 8. This is because all these systems use atomic operations to guarantee correctness, hence threads will lead to more write-write conflicts which will hurt parallelism. Compared with the runtime with 1 thread, $\mathsf{Layph}$ with 32 threads can achieve 10.1$\times$ speedup, which is higher than KickStarter (4.7$\times$ speedup), RisGraph (6.2$\times$ speedup), and Ingress (9.0$\times$ speedup). We also run $\mathsf{PageRank}$ on UK and compare $\mathsf{Layph}$ with GraphBolt, DZiG, and Ingress. The results are reported in Figure 9b where a base-10 log scale is used for the Y axis. We can observe that GraphBolt, DZiG, and $\mathsf{Layph}$ show better scaling performance than Ingress. The reason is that the problem of the write-write conflict in $\mathsf{PageRank}$ is more serious than that in $\mathsf{SSSP}$. In GraphBolt and DZiG, vertex states need to be recorded during each iteration, which can alleviate the conflict problem with massive space cost in sacrifice. In $\mathsf{Layph}$, both the shortcut update process and the local assignment process contain many independent local computations, making $\mathsf{Layph}$ more parallel-friendly. Therefore, $\mathsf{Layph}$ can benefit more from parallelism.