> For the complete documentation index, see [llms.txt](https://sliu583.gitbook.io/blog/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://sliu583.gitbook.io/blog/specific-work/shivarams-group/group-papers/garaph-efficient-gpu-accelerated-graphprocessing-on-a-single-machine-with-balanced-replication.md). # Garaph: Efficient GPU-accelerated GraphProcessing on a Single Machine with Balanced Replication ## Presentation * Large-scale graph processing * 10^10 pages, 10^12 tokens: page rank * 10^9 nodes, 10^12 edges: social network analysis * Powerful storage & computation technologies * Goal: * Large memory + fast secondary storages * CPU + GPUs * CPU: sequential * GPU: SIMD mode processing * How to efficiently integrate heterogeneity under a unified abstraction ![](/files/-McNDKY70BbQo1o78K3w) ![](/files/-McNDVdb-oLg-3z7RTxO) ![](/files/-McNE-dTOihateXQ0WBr) * Non-distributed platform ![](/files/-McNEFgbfOg7tKMUHxjL) ![](/files/-McNEW0Q44sf1V6A8OF_) * Most time-consuming: gather phase ![](/files/-McNEe2adb0D8lvnTiXh) ![](/files/-McNEit7zu0NuE0gUT71) ![](/files/-McNEx6b_LiRy6-cGNjN) ![](/files/-McNF-OVvvfttQFsZDWN) ![](/files/-McNF1PuNq7zx8l_-wQH) ![](/files/-McNF4VRPaE_aNROLAkd) ![](/files/-McNF6N5Vn88sqdHUv_G) ![](/files/-McNFALS4vsVFYOwY5iH) ![](/files/-McNFC09e7o_gO_M80jX) ![](/files/-McNFDloPlSzd5UZM2Yi) ![](/files/-McNFFeq1Du4-NLbCKXk) ## Paper ### Abstract * Garaph: GPU-accelerated graph processing system on a single machine with secondary storage as memory extension * Contributions * **Vertex replication degree customization scheme** * maximize GPU utilization given vertices' degrees and space constraints * **Balanced edge-based partition** and a **hybrid of notify-pull and pull computation models** * ensure work balance over CPU threads * optimized for fast graph processing on CPU * **Dynamic workload assignment** schemes * Takes into account of the characteristics of processing elements and graph algorithms ### Intro * Distributed graph systems: need fast network and effective partitioning to minimize communication * Alternative: non-distributed * Benefit: users need not to be skilled at managing & tuning * Problem: pressure on **memory** and **computing** power. But it's affordable. * RAM is large * Advances of secondary storage: high access bandwidth close to memory * GPU: massive parallelism to offer high-performance graph processing * Setting: GPU-accelerated, secondary-storage based graph processing * Challenge: * highly skewed degree distribution of natural graphs * Small fraction of vertices adjacent to large fraction of edges --> heavy write contention among GPU threads due to atomic updates of the same vertices * Work imbalance of CPU threads * heterogeneous parallelism of CPU & GPU * CPU: sequential processing * GPU: bulk parallel processing * Propose: Garaph * GPU: vertex replication degree customization * CPU: balanced edge-based partition * Heterogeneity of computation units * Pull computation model: matches the SIMD processing model of GPU * Hybrid of notify-pull and pull computation model: optimizes for fast sequential processing on CPU * Dynamic workload assignment ### System overview #### 2.1 Graph Representation For organizing incoming and outgoing edges: * Compressed Sparse Column (CSC) * Compressed Sparse Row (CSR) * Shard: * split vertices V into disjoint sets of vertices and each set is represented by a shard that stores all incoming edges whose destination is in that set. * Edges in a shard are listed based on increasing order of their indexes of destination vertices. * Allow each shard to be fit into the shared memory for high bandwidth * Maximum offset is 12K, can use 16-bit integer to represent the index of destination vertices * Transfer shards from host memory to GPU memory in batch * Call this as a **page** * Leverages multi-stream feature of GPUs for the overlap of memory copy and kernel execution * Two vertex-centric computation models * Pull model * Every vertex updates its state by pulling the new states of neighboring vertices through incoming edges * Notify-pull * Only active vertices notify their outgoing neighbors to update, who in turn perform local computation by pulling states of their incoming neighbors * More effective in case of few active vertices #### System Architecture * **Dispatcher** * loading graph from secondary storages, distributing the computation over CPU and GPU * Partitions each graph page into equal-size data blocks, which are uniformly distributed over multiple secondary storages with a hash function * Steps * Load data blocks from secondary storage to host memory * Construct pages * After one page is constructed, dispatch to either CPU or GPU * **GPU/CPU computation kernel** * GPU * Process the shards of page in a parallel manner * Only the pull model is enabled on GPU side * Notify-pull can lead to high frequency of non-coalesced memory accesses because of poor locality and warp divergence caused by distinguishing active/inactive vertices * CPU * Enables both pull and notify-pull * Each thread processes one edge set (divide edges of a page into sets of equal size) * Either of the two kernels has processed on page, there will be a synchronization between GPU and CPU * Execution can be done both synchronously and asynchronously * Iter: complete process over all the pages for one time * **Fault Tolerance** * Write vertex data to secondary storages periodically #### Programming API * Modified Gather-Apply-Scatter (GAS) abstraction used in the PowerGraph * Modify scatter function to activate function which sets value if the vertex satisfies the active condition * Atomic (user-provided sum function) + non-atomic operations for GPU and CPU respectively ### GPU-Based Graph Processing ![](/files/-McNUp9U15n3hw_P8NY4) * Global Memory * Up to 24GB in size. * Size of the vertices is usually 4 bytes. Can store up to 6B (or 12B) vertices in global memory. * Global Vertices: allows quick access to values of vertices * Each shard in a page is processed by one GPU block in three phases: initialization, gather, and apply * **Initialization**: * LocalVertices to store accumulate value of each vertex in a shard. * Consecutive threads of a block initialize this array with default vertex values defined by users * **Gather** * Threads of one GPU block process edges of an individual shard. For each edge, one thread fetches vertex & edge data from global memory and increase accumulate value * To have coalesced global memory accesses: consecutive threads of the block read consecutive edges' data in global memory * **Apply** * Each thread of block updates vertex value in shared memory * Async: commit new vertex data to GlobalVertices array, which are immediately visible to the subsequent computation * Sync: values are written to temporary array in global memory, which would be visible in the next iteration * When page has been processed, new vertex values are synchronized between GPU global memory and host memory * Async: transmits updated values of GlobalVertices in the GPU global memory to array storing the most updated values of vertices in the host memory * Sync: stored in temporary space of GPU global memory are transmitted to temporary array in host memory, commit after the iteration ends * Can be overlapped with processing #### Replication-Based Gather * Problem: gather phase with write contention (multi-threads simultaneously modifying the same shared memory address) --> **position conflict** * Frequent for natural graphs (power-law degree distribution) * Strategy: **replication** * which consists of placing R adjoining copies of the partial accumulated value in the shared memory to spread these accesses over more shared memory addresses. * Then these R partial accumulated values are aggregated to calculate the final accumulated value au for a vertex u. * Two-way merge * R: replication factor * Replication factor customization * Too large? GPU underutilization since fewer vertices can be fit in the shared memory * Maximizes the expected performance under given conflict degree and space constraints ### CPU-Based Graph Processing ![](/files/-McNgfZTVT6n7BIdNcvM) * Main points * How it works * Balanced edge-based partition to exploit parallelism * Dual-mode processing model: switches between pull/notify-pull modes according to the density of active vertices in the page * Existing approach: assign each thread of a subset of vertices * Computation imbalance * Random memory access of edge data if adjacent vertices are assigned to different threads * **Edge-centric partition** * Enhance sequential access of edge data and improves work balance over threads * Why mention it in CPU-based?? * Why is it not an issue on the GPU side? * Cuda blocks taking 32 threads * Done with the block, then push another block * CPU engine * GlobalVertices in host memory for quick access to values of vertices * LocalVertices to store accumulate values of destination vertices in the corresponding partition * Each page: initialization, gather, apply * If a page is processed at GPU side, system also synchronizes new vertex values between GPU memory and host memory * Processing is done * Graph state converges (i.e. no active vertices) * Active vertices: vertices with significant state change (use a bitmap to indicate) * Or: a given number of iterations complete * Each page * **Initialization** * Edges of the page is divided into partitions, and each thread processes one partition. * Number of replicas is at most n\_t - 1 * Initialize LocalVertices with vertices' default values * **Gather** * Each partition --> one thread * Edges are processed in a sequential order * For each edge, CPU thread performs gather and updates the accumulate value in LocalVertices with sum function * After each thread, aggregation phase aggregates values of vertices replicated at the partition boundaries * **Apply** * After gather phase of each page is finished, every thread updates vertices' values in the LocalVertices array * For each vertex in the partition, corresponding thread calls activate() to examine if the vertex is active or not and updates the bitmap * **Sync** * after the GPU has processed a page, it sends the corresponding vertex values to the host memory * Then, system calls Activate() of these updated vertices * Async: * enables the updates received from the GPU immediately visible through writing them into the GlobalVertices array of host memory * Then, sends back new vertices updated on CPU side to the GPU and overwrites the corresponding part of the GlobalVertices array in the GPU global memory * Sync * stores updated values in a temporary array, and commits these new values at the end of each iteration * the CPU transmits the new GlobalVertices array to that in the GPU memory at the end of each iteration #### Dual-Mode Processing Engine * **pull mode** * more beneficial when most vertices are activated (dense active vertex set), which avoids the extra costs of modification * **notify/pull mode**: a vertex needs to be updated only when one of its source vertices is active in the previous iteration * more efficient when few vertices are active in the last iteration (sparse active vertex set) * At a given time during the graph processing, the active vertex set **may be dense or sparse** * E.x. starts from sparse, then becoming denser as more vertices being activated, and sparse again when algorithm approaches convergence * Problem of combining two modes where only part of graph can be loaded into the host memory * System entails I/O cost due to sequential and random accesses of outgoing/incoming edges on secondary storage for pull & notify-pull modes respectively (use a different formula to consider the rate of speeds between sequential read and random read of secondary storage) ### Dispatcher * Adaptive scheduling mechanism to exploit the overlap of two engines #### CPU-GPU Scheduling * if T(CPU) \5, <5), don't get 32x speed up