Source-linked AI summary
Effective Techniques for Message Reduction and Load Balancing in Distributed Graph Computation
Da Yan, James Cheng, Yi Lu, Wilfred Ng
TL;DR
Pregel message passing can concentrate communication at a few vertices and imbalance machine workloads. The paper combines vertex mirroring with message combining and adds a request-respond API to address degree- and logic-driven bottlenecks. Experiments on large real-world graphs report reduced communication cost and computation time.
Problem
Pregel’s message passing can make a few vertices handle far more communication than others, creating bottlenecks from high degree or algorithm logic.
Source: §1 · pp. 1–2Method
The paper uses mirroring with message combining for high-degree bottlenecks and a request-respond API to merge requests by machine and target.
Source: §1 · p. 2Results
The experiments on large real-world graphs report lower communication cost and overall computation time; mirroring reduced Hash-Min on BTC from 26.97 to 9.55 seconds.
Source: §1 · p. 2, §8 · p. 11Takeaways & Limitations
Mirroring addresses degree-driven bottlenecks transparently to programmers, while request-respond addresses logic-driven bottlenecks and simplifies programming of many Pregel algorithms.
Source: §8 · pp. 10–11Experiments ran without resource contention; the authors expect greater gains in public data centers where network bandwidth is lower.
Source: §1 · p. 3
Abstract
from arXiv · showhide
Massive graphs, such as online social networks and communication networks, have become common today. To efficiently analyze such large graphs, many distributed graph computing systems have been developed. These systems employ the "think like a vertex" programming paradigm, where a program proceeds in iterations and at each iteration, vertices exchange messages with each other. However, using Pregel's simple message passing mechanism, some vertices may send/receive significantly more messages than others due to either the high degree of these vertices or the logic of the algorithm used. This forms the communication bottleneck and leads to imbalanced workload among machines in the cluster. In this paper, we propose two effective message reduction techniques: (1)vertex mirroring with message combining, and (2)an additional request-respond API. These techniques not only reduce the total number of messages exchanged through the network, but also bound the number of messages sent/received by any single vertex. We theoretically analyze the effectiveness of our techniques, and implement them on top of our open-source Pregel implementation called Pregel+. Our experiments on various large real graphs demonstrate that our message reduction techniques significantly improve the performance of distributed graph computation.
1. INTRODUCTION
Pregel message passing can concentrate communication at vertices with high degree or vertices targeted by an algorithm’s logic, imbalancing machine workloads. The paper introduces mirroring and request-respond to address these bottlenecks and reduce network messages.
1. INTRODUCTION
- A few bottleneck vertices can send or receive far more messages than others, increasing their machines’ workload and causing workload imbalance.The paper identifies high vertex degree and algorithm logic as sources of these bottlenecks.Source: §1 · p. 1
- On BTC, skewed vertex degrees produce uneven edge counts across workers, and because messages follow edges, communication is uneven as well.Hash-Min assigns vertices by hashing their IDs, so workers can hold similar vertex counts but different numbers of edges.Source: §1 · p. 2
- In S-V on the connected USA road network, all vertices ultimately point to Vertex 0, so Worker 0 handles disproportionate communication despite the network’s maximum degree being only 9.The imbalance comes from algorithm logic rather than skewed vertex degrees.Source: §1 · p. 2
- Hash-Min on BTC fell from 26.97 to 9.55 seconds with mirroring, while S-V on the USA road network fell from 261.9 to 137.7 seconds with request-respond.The figures compare per-worker message counts with and without the respective technique; the reported bars become shorter and more even.Source: Figure 1 · p. 2, §1 · p. 2, Figure 2 · p. 2
- Mirroring places copies of high-degree vertices on other machines, bounding each such vertex’s messages per iteration by min{M, d(v)}.Here, M is the number of machines and d(v) is the vertex’s degree; the paper notes that mirroring can interact with message combining and analyzes which vertices to mirror.Source: §1 · p. 2
- Request-respond lets a vertex request another vertex’s attribute for the next iteration and merges requests to the same target from each machine, bounding responses per vertex by M.In the S-V example, k requesters need at most min{M, k} requests and responses; the API also avoids explicitly coding the request-and-response process across at least three iterations.Source: §1 · p. 2
- The experiments ran without resource contention; the authors expect greater performance gains in public data centers with lower network bandwidth.This qualifies the reported setting and the authors’ expectation about communication reduction.Source: §1 · p. 3
2. BACKGROUND AND RELATED WORK
This section introduces Pregel’s iterative vertex-centric framework and message facilities, then situates Pregel+ alongside Java-based Pregel systems and GraphLab/PowerGraph.
2.1 Pregel
- Pregel uses BSP supersteps: active vertices process prior messages, send messages for the next superstep, and halt; incoming messages can reactivate halted vertices.Programs terminate when all vertices have halted and no messages are pending.Source: §2.1 · p. 3
- Pregel’s combiner merges messages from one machine to the same target vertex when the operation is commutative and associative; its aggregator makes a globally aggregated value available in the next superstep.The combiner example sums messages directed to one vertex.Source: §2.1 · p. 3
2.2 Pregel-Like Systems in JAVA
- Java-based Pregel systems serialize and deserialize vertex data around computation, which can be costly for long adjacency lists; Pregel+ is implemented in C/C++ to avoid that serialization overhead.The passage attributes the Java overhead to object management by the garbage collector and serialization of binary vertex representations.Source: §2.2 · p. 3
- GPS’s LALP resembles mirroring but does not explore its tradeoff with message combining; the authors argue both techniques should be used and motivate a cost model for selecting mirrored vertices.The passage says sender-side combining reduces overall running time in their experiments.Source: §2.2 · p. 3
2.3 GraphLab and PowerGraph
- GraphLab uses asynchronous execution and neighbor-only data pulling, unlike request-respond’s ability to address non-neighbors needed by some pointer-jumping algorithms.The passage names S-V for connected components and the Pregel minimum spanning forest algorithm as examples.Source: §2.3 · p. 3
- GraphLab creates mirrors for every vertex, while PowerGraph partitions by edges to distribute a high-degree vertex’s edges across workers and uses the GAS model.The passage links GraphLab’s all-vertex mirrors to excessive space consumption.Source: §2.3 · p. 4
3. PREGEL ALGORITHMS
This section introduces Pregel algorithms for common graph tasks and explains their operations, including how algorithm logic can create communication bottlenecks. The algorithms provide concepts for the paper’s performance evaluation.
3. PREGEL ALGORITHMS
- The section presents algorithms for connected components, spanning forests, and PageRank as examples for illustrating concepts and evaluating performance.Source: §3 · p. 4
3.1 Attribute Broadcast
- Attribute broadcast builds each vertex’s adjacency list by pairing every out-neighbor with that neighbor’s attribute, using three supersteps to request, return, and collect the attributes.Source: §3.1 · p. 4
3.2 PageRank
- PageRank initializes each vertex’s score, repeatedly sums incoming contributions, applies the 0.15/|V| + 0.85 × sum update, and sends shares to out-neighbors.Source: §3.2 · p. 4
3.3 Hash-Min
- Hash-Min propagates the smallest vertex ID seen so far through neighbors until each vertex’s value equals its connected component’s minimum ID.Source: §3.3 · p. 4
3.4 The S-V Algorithm
- S-V organizes each connected component as a pointer forest and repeats tree hooking, star hooking, and shortcutting until every vertex is in a star.Shortcutting can create large fan-out: in the last round, all vertices in a component request D[w] from its component-ID vertex, creating a bottleneck.Source: §3.4 · pp. 4–5
3.5 Minimum Spanning Forest
- In the MSF algorithm, each vertex picks a minimum-weight edge to form conjoined-trees, then pointer jumping identifies each tree’s supervertex.The smaller-ID vertex in the cycle becomes the supervertex; remaining cross-tree edges are sent to it, where they form an adjacency list.Source: §3.5 · p. 5
4. BASIC COMMUNICATION FRAMEWORK
Pregel+ implements its communication framework with MPI workers and batched message exchange between supersteps. It routes messages through per-vertex incoming buffers and per-worker outgoing buffers, with optional combining before transmission.
- Pregel+ uses one MPI process per worker, with messages added to a channel and sent to target workers in batches before the next superstep.The framework assumes one worker per machine for discussion, while noting that the concepts generalize.Source: §4 · p. 5
- Each worker indexes incoming messages by target vertex, appending them to that vertex’s buffer before compute() processes the messages for active vertices.The incoming-buffer lookup uses a hash table keyed by vertex ID.Source: §4 · p. 5
- Workers maintain one outgoing buffer per worker; messages are batched by destination, and an optional combiner groups messages by target vertex before sending.Source: §4 · pp. 5–6
5. THE MIRRORING TECHNIQUE
Vertex mirroring addresses high-degree communication bottlenecks by sending a vertex’s attribute to mirrors on workers that hold its neighbors, which forward it locally. The paper bounds network messages and gives a degree threshold for when mirroring reduces communication, while integrating the technique with message combining and edge-dependent messages.
The Mirroring Technique
- A high-degree vertex sends its attribute to mirrors on the workers containing its neighbors, which forward it locally without network communication.For j neighbors on one worker and k on another, mirroring sends two network messages rather than j + k.Source: §5 · p. 6
- Mirroring bounds a vertex’s message count by min{M, d(v)} when it delivers one value to all neighbors in a superstep.The bound follows because the vertex sends to at most one mirror per other machine.Source: §5 · p. 6
- Message combining on the normal channel cannot combine a mirror-channel message with messages sent to the same receiver, while the paper’s method integrates mirroring with combining and supports edge fields.The contribution summary contrasts this integration with GPS’s LALP and GraphLab’s mirrors for all vertices.Source: §5 · pp. 6, 8
- Theorem 2 sets the communication-reducing threshold at τ = M · exp{degavg/M}, assuming evenly partitioned vertices, randomly chosen neighbors, and constant average degree.The analysis compares the expected non-wasted messages without mirroring against at most M messages to mirrors.Source: §5 · p. 7
- For M = 100 and degavg ≤50, the threshold is at most 165; the paper reports that the model is effective on real graphs despite its simplifying random-graph assumption.The authors also report that the threshold achieves near-optimal performance in experiments.Source: §5 · pp. 7–8
- Pregel+ applies mirroring above a degree threshold and constructs each mirror with the vertex’s local neighbor list; below the threshold, messages use the normal channel.The interface uses broadcast for vertex-state messages and relay(msg) to update edge-dependent values such as SSSP distances.Source: §5 · pp. 6–8
6. THE REQUEST-RESPOND PARADIGM
The request-respond paradigm adds an API for requesting another vertex’s attribute, available in the next iteration, to address bottlenecks created by algorithm logic. It deduplicates requests by machine and returns attributes through response sets, reducing messages when many vertices request the same value.
The Request-Respond Paradigm
- Each machine stores requests by destination machine in hash sets, eliminating duplicate requests to the same vertex from that machine.After exchange, response sets return requested attributes for access in the next superstep.Source: §6 · p. 8
- For ℓ requesters seeking a(u), request-respond reduces messages from 2ℓ to 2 min(M, ℓ), where M is the number of machines.Each machine sends at most one request to u and receives at most one response from u.Source: §6 · p. 8
- In the worst case, request-respond uses as many messages as Pregel, but for many requesters it bounds messages by the number of machines containing them and eliminates imbalanced workload.The paper notes that many algorithms have bottleneck vertices with numerous requesters.Source: §6 · p. 8
- Explicit responding lets a vertex push its attribute to known requesters, avoiding request exchange and response-set construction; on undirected graphs, responses go to machines rather than individual vertices.The interface also supports pulling through request(v) and retrieving the value with get_resp(v) in the next superstep.Source: §6 · p. 9
7. EXPERIMENTAL RESULTS
The experiments find that request-respond reduces message counts, while mirroring improves performance and workload balance particularly on skewed-degree graphs. Gains depend on the graph and algorithm: mirroring can add overhead when it yields little message reduction.
7. EXPERIMENTAL RESULTS
- Evaluation used five real-world graphs on a 16-machine cluster, running PageRank on directed graphs and Hash-Min on undirected graphs.The reported PageRank runs converged in 89, 89, and 96 supersteps on WebUK, Twitter, and LJ, respectively.Source: §7 · p. 9, §7.1 · p. 9
7.1 Effectiveness of Mirroring
- Mirroring can be ineffective for WebUK PageRank: small message reductions fail to repay the added interception cost, and Pregel-noM performs best.The passage attributes the limited reduction to most WebUK vertices having modest degree.Source: §7.1 · p. 9
- The cost-model threshold achieves the best or near-best performance among tested thresholds, while mirror-construction preprocessing is short compared with computation.The tested thresholds were 1, 10, 100, 1000, and the cost-model threshold.Source: §7.1 · pp. 9–10
- Mirroring reduces running time and message counts, with particularly clear gains on skewed-degree Twitter, LJ, and BTC, demonstrating workload-balancing effectiveness.Figure 12 reports the mirroring results.Source: §7.1 · p. 9, Figure 12 · p. 10
- Pregel+ with mirroring and combining reduces messages and running time more than GPS with LALP; Pregel-noM is also faster than Giraph and GraphLab.The comparisons use Figure 12.Source: Figure 12 · p. 10, §7.1 · p. 10
7.2 Effectiveness of Request-Respond Technique
- 11,015 million to 2,699 million: ReqResq cuts attribute-broadcast messages on WebUK and has the shortest running time except in a few GPS cases and USA minimum spanning forest.The USA exception occurs because its vertices have very low degree, so request-respond overhead is not offset by fewer messages.Source: Figure 13 · p. 10, §7.2 · p. 10
8. CONCLUSIONS
The conclusion presents mirroring and request-respond as complementary techniques for reducing communication bottlenecks. Experiments on large real-world graphs verify reductions in communication cost and overall computation time.
8. CONCLUSIONS
- Mirroring addresses communication bottlenecks caused by high vertex degree and is transparent to programming.The paper presents it as one of two techniques for reducing communication and skewed workload.Source: §8 · pp. 10–11
- The request-respond paradigm targets bottlenecks caused by program logic and simplifies programming for many Pregel algorithms.Source: §8 · p. 11
- Experiments on large real-world graphs verify that the techniques reduce communication cost and overall computation time.Source: §8 · p. 11