PaperlayerView exact revision on arXiv

Source-linked AI summary

Dorylus: Affordable, Scalable, and Accurate GNN Training with Distributed CPU Servers and Serverless Threads

John Thorpe, Yifan Qiao, Jonathan Eyolfson, Shen Teng, Guanzhou Hu, Zhihao Jia, Jinliang Wei, Keval Vora, Ravi Netravali, Miryung Kim, Guoqing Harry Xu

arXiv:2105.11118v2cs.DCcs.LG

TL;DR

GNN training is constrained by the cost and memory limits of GPU servers, while CPU-only and sampling-based alternatives trade away parallelism, speed, or accuracy. Dorylus separates graph and tensor computation across CPU servers and Lambda threads, then overlaps them with bounded asynchronous pipelining. It scales to billion-edge graphs and improves performance-per-dollar over CPU-only, GPU-only, and sampling-based systems.

  • Problem

    GNN training must handle expensive GPU infrastructure and limited GPU memory for billion-edge graphs, while sampling introduces overhead and can reduce accuracy.

  • Method

    Dorylus separates graph-parallel work on CPU servers from tensor-parallel work on Lambda threads and uses bounded asynchronous pipelining to overlap them.

  • Results

    Dorylus adds 2.75× more performance-per-dollar than CPU-only servers, 4.83× more than GPU-only servers, and is up to 3.8× faster and 10.7× cheaper than sampling-based systems.

  • Takeaways & Limitations

    Dorylus provides scalable, efficient, and low-cost GNN training for small organizations and billion-edge graphs.

  • Takeaways & Limitations

    The evaluation ran GAT on only one small and one large graph because its intensive ApplyEdge computation was monetarily expensive.

Abstract

from arXiv · show

A graph neural network (GNN) enables deep learning on structured graph data. There are two major GNN training obstacles: 1) it relies on high-end servers with many GPUs which are expensive to purchase and maintain, and 2) limited memory on GPUs cannot scale to today's billion-edge graphs. This paper presents Dorylus: a distributed system for training GNNs. Uniquely, Dorylus can take advantage of serverless computing to increase scalability at a low cost. The key insight guiding our design is computation separation. Computation separation makes it possible to construct a deep, bounded-asynchronous pipeline where graph and tensor parallel tasks can fully overlap, effectively hiding the network latency incurred by Lambdas. With the help of thousands of Lambda threads, Dorylus scales GNN training to billion-edge graphs. Currently, for large graphs, CPU servers offer the best performance-per-dollar over GPU servers. Just using Lambdas on top of CPU servers offers up to 2.75x more performance-per-dollar than training only with CPU servers. Concretely, Dorylus is 1.22x faster and 4.83x cheaper than GPU servers for massive sparse graphs. Dorylus is up to 3.8x faster and 10.7x cheaper compared to existing sampling-based systems.

1 Introduction

GNN training must balance GPU cost and memory limits against CPU inefficiency and sampling drawbacks. Dorylus addresses this tension by separating graph and tensor computation across CPU servers and inexpensive Lambda threads, then overlapping them through asynchronous pipelining.

  • Motivation: GPU servers provide parallelism but are expensive and have limited memory for billion-edge graphs.These constraints burden smaller organizations and limit scalability even when multiple GPUs are coordinated.
  • Motivation: CPUs offer lower cost and looser memory restrictions but lack the parallelism needed for efficient GNN training.
  • Motivation: Graph sampling improves scalability by reducing processed data but adds repeated per-epoch overhead, can reduce accuracy, and lacks convergence guarantees for trivial methods.
  • Design insight: Dorylus separates graph-parallel tasks for CPU instances from tensor-parallel tasks for Lambdas, reducing tensor work to lightweight operations suitable for Lambda threads.This separation exploits the fact that large-graph GNN training is dominated by graph computation rather than tensor computation.
  • Design insight: BPAC overlaps graph and tensor tasks to hide Lambda communication latency, while bounded asynchrony controls staleness at parameter updates and neighbor gathering.
  • Takeaway: Dorylus provides a scalable, efficient, and low-cost GNN training scheme for small organizations and billion-edge graphs.

2 Background

GNNs combine graph propagation with neural-network operations over vertex and edge features. In a GCN, each forward layer gathers neighborhood information, applies a neural transformation, and scatters the result, with backward propagation computing parameter gradients.

  • GNNs: GNNs learn from graph structure and vertex features, producing feature vectors for vertices or the whole graph.
  • GNN computation: GNN training combines graph propagation tasks such as Gather and Scatter with neural-network computations.
  • GCN forward pass: A GCN forward layer uses Gather, ApplyVertex, Scatter, and ApplyEdge to transform vertex activations.Gather aggregates incoming neighbor vectors, ApplyVertex performs neural computation, and Scatter propagates updated vectors along outgoing edges.
  • GCN forward pass: For GCN, ApplyVertex multiplies gathered activations by a trainable weight matrix and applies a nonlinear activation, while ApplyEdge is an identity.
  • Training: The backward pass computes gradients for trainable vertex- and edge-level neural-network weights, and repeated epochs update those weights until acceptable accuracy is reached.

3 Design Overview

Dorylus combines graph servers, Lambda threads, and parameter servers in a distributed architecture. Graph operations remain on partitioned CPU servers, while tensor operations interleave through Lambdas and exchange weights with parameter servers.

  • Architecture: Dorylus consists of EC2 graph servers, Lambda threads for tensor computation, and EC2 parameter servers.
  • Architecture: An edge-cut algorithm partitions the input graph across graph servers while balancing partition load.
  • Graph processing: Graph servers communicate across partition boundaries and execute vertex-parallel and edge-parallel graph stages such as Gather and Scatter.
  • Tensor processing: Lambdas perform tensor operations between graph stages, retrieving forward weights from parameter servers and computing updated weights during backpropagation.

4 Tasks and Pipelining

Dorylus decomposes GNN training into fine-grained graph and tensor tasks and connects them in a pipeline. This enables graph servers, Lambdas, and parameter servers to overlap forward and backward work, hiding Lambda communication latency.

  • Task decomposition: Computations involving the graph adjacency structure run as graph operations on graph servers, while tensor-only computations are assigned to Lambdas.
  • Forward tasks: A forward pass contains Gather and Scatter graph tasks plus ApplyVertex and ApplyEdge tensor tasks.
  • Backward tasks: Backward execution adds reverse graph tasks, reverse tensor tasks, and WeightUpdate for aggregating gradients across parameter servers.
  • Pipelining: Dorylus divides vertices into intervals so tasks can be processed concurrently by graph servers, Lambdas, and parameter servers.Intervals are balanced using vertex and inter-interval edge counts to distribute work.
  • Pipelining: 2. Dorylus overlaps graph-parallel and tensor-parallel computations through effective pipelining to hide Lambda communication latency.The paper attributes feasible GNN pipelining to fine-grained tasks and computation separation.

5 Bounded Asynchrony

Dorylus uses bounded asynchronous execution across graph- and tensor-parallel paths to overlap work while controlling stale updates. The design addresses two synchronization points and provides a convergence guarantee under bounded staleness.

  • Bounded Asynchrony: Bounded asynchronous execution lets pipeline tasks proceed without waiting for slower tasks, mitigating stragglers while preserving controlled synchronization.Dorylus introduces asynchrony at parameter updates and neighbor-data gathering, where staleness can be tolerated.
  • Bounded Asynchrony: Dorylus has two synchronization points: weight synchronization at WU tasks and neighbor activation synchronization at Gather tasks.This dual-path structure creates a synchronization challenge absent from systems with only one computation path.
  • 5.1 Bounded Asynchrony at Weight Updates: Weight stashing bounds staleness at parameter updates by preserving the weight version used for each interval’s forward computation.Dorylus assigns an interval to one parameter server for an epoch so subsequent tensor tasks use its stashed weights.
  • 5.2 Bounded Asynchrony at Gather: Bounded staleness at Gather allows intervals to use neighbor activations up to S epochs stale, while requiring them to wait when updates exceed that bound.This permits fast intervals to progress when recent data is available without allowing arbitrarily old activations.
  • 5.3 Convergence Guarantee: Under stated smoothness, Lipschitz, and bounded-gradient assumptions, asynchronous GNN training eventually converges to a local minimum.The guarantee holds for bounded staleness and remains valid regardless of the value of S.
  • 5.3 Convergence Guarantee: Unlike trivial sampling methods, Dorylus’s bounded asynchronous computation is guaranteed to converge while avoiding sampling’s per-epoch overhead and accuracy reduction.The comparison is author-reported and contrasts whole-graph asynchronous training with sampling-based training.

6 Lambda Management

Dorylus manages Lambda execution through controllers, communication optimizations, and adaptive concurrency. These mechanisms reduce invocation and communication costs while tuning Lambda parallelism to CPU workload and graph structure.

  • Lambda Management: Each GS controller launches Lambdas, batches their inputs, monitors health, and routes results back to the GS.Controllers provide the coordination layer for task execution across Lambda threads.
  • Lambda Management: Dorylus launches Lambda threads when their preceding tasks begin, preparing tensor tasks in advance of incoming graph-task results.Lambdas use OpenBLAS with AVX instructions and communicate with GSes and PSes through ZeroMQ.
  • Lambda Management: Deploying Lambdas inside a VPC increases their bandwidth when communicating with EC2 instances, and repeated use makes them warm during training.Initialized Lambdas pull vertex, edge, and weight data from GSes and PSes.
  • Lambda Optimizations: Task fusion merges adjacent final-layer forward and backward operations, reducing thousands of Lambda invocations per epoch and saving a communication round trip.The fused operation combines AV and ∇AV across consecutive passes.
  • Lambda Optimizations: Tensor rematerialization avoids caching intermediate tensor results on GSes when Lambda bandwidth makes recomputation more profitable.This optimization trades additional computation for reduced network communication.
  • Lambda Optimizations: Lambda-internal streaming overlaps computation with retrieval by processing one data half while fetching the second, reducing response time.The optimization hides part of the communication cost within each Lambda.
  • Autotuning Numbers of Lambdas: An autotuner adjusts the number of Lambdas using CPU task-queue behavior because the optimal concurrency depends on graph structure and partitioning.It scales down when the queue grows persistently and scales up when the queue quickly shrinks.

7 Evaluation

Dorylus is evaluated across GCN and GAT, multiple graph scales, instance types, asynchronous variants, and CPU/GPU baselines using performance, cost, and value metrics. Results show that bounded asynchrony and Lambda augmentation improve value, especially for large sparse graphs, while large staleness can hurt convergence and GAT benefits particularly from Lambda parallelism.

  • Evaluation setup: Dorylus evaluates two-layer GCN and GAT models across four graphs, including Friendster with 1.8 billion undirected edges.The first three graphs provide features and labels; Friendster does not.
  • Instance selection: 4.93× higher value makes 8 V100s preferable to 8 K80s for Amazon training.The V100 configuration also reduced runtime from 1578 to 385 seconds and cost from $3.16 to $2.62.
  • Asynchrony: 8%/41% more epochs are required by async (s=0/1) than Dorylus-pipe to reach the same accuracy.Async (s=0) and async (s=1) reduce per-epoch time by approximately 15%, but larger staleness increases convergence cost.
  • Asynchrony: 1.234× faster and 1.288× higher value make async (s=0) preferable to pipe.Async (s=0) is also 1.233× faster and 1.494× higher value than async (s=1).
  • Scope boundary: Cost limits GAT evaluation to one small and one large graph, although the system can scale GAT beyond Amazon when cost is not a concern.The additional AE computation in GAT increases cost.
  • Effects of Lambdas: 2.75× higher value than CPU-only demonstrates the benefit of adding Lambdas for GAT on Amazon.Dorylus with Lambdas combines lower runtime and cost than the CPU-only variant in this comparison.
  • Graph and model effects: Large sparse graphs are where Dorylus adds value over both CPU-only and GPU-only variants.For Amazon and Friendster, GPU scatter is especially slow because transferring ghost data across GPU memories is costly; Lambda threads are more effective for GAT because of its additional per-edge tensor computation.
  • Effects of Lambdas: 2.82× speedup with only 5% more cost raises Dorylus’s value by 2.68× when servers increase from 4 to 16.Dorylus can provide comparable value with roughly half as many servers as the CPU-only variant.

8 Related Work

Dorylus combines fine-grained graph–tensor pipelining with computation separation to address scalability limits in prior GNN systems. It differs from existing parallel and graph-training approaches by separating graph computation from tensor computation.

  • Dorylus is the first system described here to train GNNs with tiny Lambda threads using graph-related optimizations.
  • Dorylus uses finer-grained pipeline tasks than model-parallel systems, allowing graph tasks on graph servers and tensor tasks on separate workers.
  • Asynchronous training reduces epoch time but can require more epochs for a target accuracy because workers use stale parameters; bounded staleness mitigates this issue.
  • Prior scalable GNN systems commonly use graph sampling, while NeuGraph uses multiple GPUs and Roc uses dynamic graph partitioning.
  • Dorylus separates graph computation from tensor computation rather than representing the graph as a matrix for an underlying tensor-training framework.

9 Conclusion

Dorylus combines CPU servers and serverless threads in a distributed architecture that separates graph and tensor computation. The system scales to billion-edge graphs while improving performance-per-dollar over CPU-only, GPU-only, and sampling-based alternatives.

  • 2.75× more performance-per-dollar than CPU-only servers and 4.83× more than GPU-only servers were achieved by adding Lambdas.
  • Up to 3.8× faster and 10.7× cheaper than existing sampling-based systems, Dorylus improves both performance and cost.
  • Dorylus launches graph servers for gather and scatter operations, parameter servers for model weights, and serverless threads for parallel work.
  • Separating graph and tensor components enables a deep asynchronous pipeline that overlaps graph and tensor operations.
  • The artifact includes graph and weight server components, Lambda functions, AWS setup instructions, and a public repository.

A.3.3 Preparing the Data

Preparing Dorylus data requires graph, partition, feature, and label inputs arranged in a dataset directory on an NFS server. After preparation, the system is launched with dataset, Lambda, pipeline, staleness, and backend options.

  • Dorylus requires four main inputs: a graph, partition information, features, and training labels.
  • The graph must be a binary edge list named graph.bsnap with contiguous four-byte vertex identifiers.
  • Partition information uses edge-cut partitioning and records partition assignments line by line.
  • Features and labels are stored in binary files named features.bsnap and labels.bsnap.
  • The prepared dataset is placed under /mnt/filepool/ on an NFS server, then mounted before execution.
  • Dorylus runs with a dataset name, Lambda count, optional asynchronous pipelining, staleness degree, and CPU, GPU, or serverless backend.

B Full Proof

The paper defines asynchronous GNN training as asynchronous SGD combined with asynchronous Gather, contrasting it with synchronous SGD and synchronous Gather.

  • Asynchronous GNN training means asynchronous SGD with asynchronous Gather, while synchronous training uses normal SGD with synchronous Gather.

B.1 Proof of Theorem 1

The proof establishes convergence of asynchronous GNN training by showing that stale activations and gradients remain close to their synchronous counterparts, so gradient bias vanishes.

  • Theorem 1 follows from three steps: bounding approximate activations, bounding approximate gradients, and proving asymptotic convergence.The proof uses Lemmas 1 and 2 to establish the required approximation bounds before deriving convergence.
  • Asynchronous GNN training generates weights that change slowly enough for gradient bias to approach zero, yielding convergence.This is the stated conclusion of Theorem 1.
  • The proof uses standard infinity-norm bounds for matrix products, element-wise products, and sums.These inequalities control approximation errors throughout the activation and gradient analysis.
  • A mixing matrix assigns each element to exactly one of N source matrices.This definition supports the analysis of stale activations assembled from different training states.

B.1.1 A single layer in GCN

The single-layer analysis shows that bounded staleness keeps asynchronous activations close to synchronous activations when inputs and weights change slowly, and this result extends layer by layer to multi-layer GCNs.

  • A single layer in GCN: Under asynchronous training, stale activations are mixed from the latest and preceding S epochs, where S is the staleness bound.The mixing representation captures the data used by asynchronous neighbor gathering.
  • A single layer in GCN: The approximate asynchronous outputs also change by less than Kϵ across training states when weights change slowly.This controls variation among asynchronous outputs needed for the multi-layer argument.
  • A single layer in GCN: When activation functions are Lipschitz and inputs and weights change by at most ϵ, approximate outputs differ from exact outputs by less than Kϵ.The bound applies to both pre-activation matrices Z and activation matrices H.
  • Multi-layer GCNs: Applying the single-layer bound successively across layers bounds multi-layer activation error by the model-weight change rate, regardless of staleness bound S.The result is obtained by applying Proposition 3 layer by layer.
  • Multi-layer GCNs: The gradient analysis defines asynchronous and synchronous losses and gradients, then bounds their difference under Lipschitz and bounded-gradient assumptions.The back-propagation equations propagate activation-error bounds through each layer.

B.1.4 Proof of Theorem 1

Theorem 1 proves convergence for bounded-staleness asynchronous GCN training under smoothness, Lipschitz, and bounded-gradient assumptions, with asymptotically unbiased gradients.

  • The theorem assumes Lipschitz activations, Lipschitz and bounded loss gradients, bounded training gradients, and a smooth loss.These conditions define the scope of the convergence guarantee.
  • The convergence proof uses a warm-up period of LS epochs and bounds the asynchronous gradient error relative to the synchronous gradient.The error is represented as δ_i = g_AS,i(W_i) − g_i(W_i).
  • The proof averages the resulting inequalities over N iterations to establish the asymptotic convergence result.The final step divides the summed inequality by N.
  • After at least L × S iterations, the asynchronous gradient becomes asymptotically unbiased as N approaches infinity.Here L is the number of GNN layers and S is the staleness bound.
Loading 2105.11118v2…