Finding every small shape inside a huge graph — every triangle, every four-node clique — is one of the most expensive things a computer can be asked to do, and the graphs no longer fit on one machine; the existing way to spread the work across a cluster gave each machine an entire search tree and added a general-purpose cache so no piece of the graph was fetched twice, which sounds right but made 64 cores run slower than a single core, because the bookkeeping that tracked which task needed which data cost more than the data ever did; this paper (Purdue) makes the unit of work a SINGLE extension step instead of a whole tree, which allows bounded-memory scheduling and three cheap reuse tricks that need no bookkeeping at all — and it wins by moving roughly THREE TIMES more data.
The argument in one glance
-
Problem
The search explodes, and the graph no longer fits on one machine
Graph pattern mining finds every subgraph of a large graph that matches a shape you specify — every triangle, every four-node clique, every chain of six. The counts are astronomical: a tiny graph with only 7,000 vertices contains more than 30 TRILLION six-node chains. Meanwhile real graphs have grown past 100 billion edges, far beyond the memory of any single machine. So the work must be split across a cluster in a way that scales with processing power AND with memory. That is genuinely hard, because the search constantly needs the neighbour list of some vertex, and after splitting the graph that vertex usually lives on another machine.
-
What people did
Give each machine whole search trees, and cache what it fetches
The established design for partitioned graphs assigns each machine the complete search starting from each of its own vertices, fetching whatever remote data that search needs. Since many searches want the same popular neighbour lists, a general software cache holds fetched lists so none is transferred twice. To know when a cached list can be freed, the system maintains a map from tasks to the lists they depend on. Both parts are reasonable: the caching genuinely avoids redundant transfers, and the map is the honest way to know when data is no longer needed.
-
Why it fell short
The bookkeeping costs more than the network it protects
The authors ran the public implementation. Counting triangles on the Patents graph took 27.2 seconds on eight machines with 64 cores, while a SINGLE-THREADED program finished in 6.2 seconds — 64 cores running 4.4 times SLOWER than one. Two causes. A whole search tree first fetches a multi-hop neighbourhood, which is so large that only about 150 to 300 trees run concurrently, leaving too few outstanding requests to overlap network transfers with useful work. And the map must be updated on every request, the scheduler must repeatedly check whether each task has its data, and the cache must repeatedly check whether every task using a list has finished. Compute threads spent more than 70 percent of their time waiting.
-
This paper's idea
Make one extension step the task, and accept more traffic to delete the bookkeeping
This paper (Purdue — Chen and Qian) shrinks the unit of work from an entire tree to a SINGLE STEP: adding one vertex to one partially-built subgraph. The object it defines, an EXTENDABLE EMBEDDING, is that partial subgraph together with exactly the neighbour lists its next step requires. Fine-grained tasks make two things possible. Same-level tasks are grouped into fixed-size chunks explored breadth-first inside and depth-first between, which bounds memory while keeping thousands of requests outstanding. And three reuse mechanisms replace the general cache, all requiring no bookkeeping: a child points at its parent for shared lists, tasks in one chunk share through a hash table that DROPS collisions rather than chaining, and a static cache that never evicts anything. The result moves about THREE TIMES more data than the general cache did, and runs 19 times faster.
Why this problem is worth solving
The workload is unusual because the output, not the input, is what explodes.
GRAPH PATTERN MINING takes
a large input graph and a small PATTERN — a shape such as a triangle, a
four-node clique, or a chain — and finds every subgraph of the input that has
that shape. Each match is called an EMBEDDING. The number of matches grows
brutally with pattern size: a graph with only 7,000 vertices contains more than
30 trillion six-node chains. Modern systems build matches by construction rather
than by testing candidates, using nested loops in a fixed matching order, one
loop per pattern vertex. Running those loops from one starting vertex traces a
TREE. Each tree node is a partially-built subgraph, each tree level is one loop,
and producing a node’s children means taking the NEIGHBOUR LIST of one or two
vertices already chosen and INTERSECTING them, since the next vertex must be
adjacent to all the right ones. The whole computation is one such tree per vertex
in the input graph, and the shape of each tree depends on the data
Splitting that across machines creates a problem that has no comfortable answer.
A
graph can be given to a cluster two ways. REPLICATION puts a full copy on every
machine, so no lookup ever crosses the network — but the largest graph you can
process is then limited by one machine’s memory, and real graphs have passed 100
billion edges. PARTITIONING splits the vertices across machines, with each
machine holding the edges of its own vertices, so memory scales with the cluster
— but now a neighbour list you need may be elsewhere. Two policies exist for
that. MOVING COMPUTATION TO DATA sends the partially-built subgraph to whichever
machine holds the data; it communicates constantly and cannot reuse anything,
because every partial subgraph transferred is unique. MOVING DATA TO COMPUTATION
fetches the remote list instead, which can be reused if it is needed again. Skew
makes both worse. Real graphs follow a POWER LAW, so a few very high-degree
vertices appear in far more subgraphs than the rest — in one measurement, the
most-accessed 5 percent of the graph accounted for 93 percent of all
communication
What people did before, and where it broke
- Run on one machine and go out of core. Single-machine systems are highly optimised and can spill to disk through memory mapping. But one machine’s processing power caps the throughput, and complex patterns perform poorly.
- Distribute with a replicated graph. Systems such as GraphPi put a full copy everywhere and split the work, avoiding all remote lookups. This performs very well and is the fastest prior approach — but the graph must fit in one machine’s memory, so the largest datasets are simply out of reach.
- Move the computation to the data. Send partial subgraphs to whichever machine holds what they need. The measurements are poor: one such system took roughly 1,000 seconds to count triangles on Friendster with 224 cores, while a single-machine system did it in about 400 seconds with 16 cores.
- Move the data to the computation, with a general cache. G-thinker assigns each machine the complete search from each of its own vertices and fetches remote lists, caching them so popular lists are transferred once. This is the design that should have worked, and it is the only prior partitioned-graph system that is competitive at all.
- But whole trees starve the network of work. Each task first fetches a multi-hop neighbourhood, whose size grows sharply with depth, so only about 150 to 300 trees run at once. With so few outstanding requests, transfers cannot be batched or overlapped with computation.
- And the bookkeeping dominates. The map from tasks to their dependent lists must be updated on every request; the scheduler repeatedly checks whether each task has its data; the cache repeatedly checks whether all users of a list have finished so it can be freed. Compute threads spent over 70 percent of their time waiting. On Patents, 64 cores ran 4.4 times SLOWER than one core.
- The gap. Everyone was minimising network traffic, and the general cache is the correct way to do that. Nobody had asked whether the traffic was still the expensive part once the machinery for avoiding it was counted.
How a smaller unit of work changes everything
Shrink the task to one extension, then let cheap mechanisms replace expensive ones
-
Explore one search tree per vertex
For each vertex of the input graph, run the nested loops that build every matching subgraph starting there. Producing a tree node children means intersecting the neighbour lists of vertices already chosen. With the graph partitioned, many of those lists live on other machines.
-
Make ONE extension the unit of work new
Instead of a whole tree per task, define an extendable embedding — one partially-built subgraph plus exactly the neighbour lists its next step needs. Adding one vertex to one partial subgraph becomes the task. Only the active lists count, and a vertex already used may not be active if the next step does not need it.
-
Group same-level tasks into fixed-size chunks new
A chunk holds a configurable number of tasks at the same tree level. Exploration is breadth-first inside a chunk and depth-first between chunks. Memory is bounded because only one chunk per level from root to current level is live, and a chunk is allocated and freed as one block, so nothing fragments.
-
Shuffle a chunk by source machine and pipeline it new
Before running a chunk, sort its tasks into batches by which machine holds their data, ordered so each machine starts with a different peer rather than all of them requesting the same one. The fetch for one batch overlaps the computation of the previous, and computation is never allowed to stall a fetch.
-
Replace the cache with three reuse tricks that need no bookkeeping new
A child stores only the one new neighbour list and points at its parent for the rest. Tasks within a chunk share through a hash table that simply DROPS a colliding entry instead of chaining. A static cache admits high-degree lists until full and then never evicts, so nothing must ever be tracked or reference-counted.
-
Plug into existing single-machine systems new
The engine exposes one function, EXTEND, in which an existing system expresses its nested loops. Two state-of-the-art single-machine systems were ported for roughly 500 lines of code each, and their user-facing interfaces were left unchanged.
The idea
Task size is a design parameter, not a detail. The prior system chose a whole tree as its task because a tree is the natural unit — one starting vertex, one self-contained search. But that choice determined everything else. A whole tree needs a large multi-hop neighbourhood before it can begin, so its memory is large, so few run at once, so there are too few outstanding requests to batch or overlap, so the network sits idle while compute threads wait. Shrinking the task to a single extension inverts each of those. Thousands of tasks are live, each tiny, each with precisely specified data needs. The paper is careful that “precisely” means what it says: in a partial subgraph, only some vertices are ACTIVE, meaning their neighbour lists are needed for the next step, and a vertex just added is often not among them. Choosing the granularity at which work becomes schedulable, rather than accepting the granularity the problem suggests, is the same decompose-until-it-schedules move covered elsewhere on this site where breaking a monolithic unit into fine pieces is what finally lets a system overlap two resources it had been using one at a time.
Structure gives you reuse that a general mechanism has to discover. The general cache works for any access pattern, which is exactly why it is expensive: knowing nothing, it must record who needs what and check repeatedly. This paper exploits three facts specific to the problem. A child subgraph extends its parent by one vertex, so all but one of its neighbour lists are already in the parent — the child stores one list and follows a pointer for the rest, and only that one list is ever fetched. Tasks in the same chunk are at the same tree level and often want the same popular list, so a small per-chunk hash table catches that, and when two vertices collide the system simply DROPS one rather than building a chain, accepting a little redundant traffic to keep the table free of branching. And popular data stays popular, so a cache that admits high-degree lists until full and then never evicts captures most of the benefit while needing no reference counts, no eviction policy and no scans. Replacing a general mechanism with several narrow ones that exploit known structure is the same specialise-what-you-know reasoning covered elsewhere on this site where a domain-specific design beats a general-purpose library precisely because it can assume things the library cannot.
Optimising communication was the wrong objective. This is the paper’s own stated takeaway, and its evidence is unusually direct. The new system moves about THREE TIMES more data across the network than the general cache did, and is 19 times faster. When the authors compare cache policies, the never-evicting one sometimes moves MORE data than least-recently-used or first-in-first-out, and still wins by roughly an order of magnitude, for two compounding reasons: replacement policies cost computation, and once communication has been reduced enough to hide behind computation, reducing it further buys nothing. They report cases where cutting traffic changed the runtime not at all. Traffic was a proxy for time that stopped tracking time, and every system before this one kept optimising the proxy. Discovering that an objective everyone accepted has stopped predicting the outcome is the same measure-the-true-thing lesson covered elsewhere on this site where a benchmark improves the number being watched while the quantity users experience does not move.
Extending a subgraph across two machines
The paper's running example, on a graph split across three machines
Vertices 0 and 1 live on machine 1; vertices 2, 3 and 4 on machine 2. Following one task through its four states.
Step 1 of 5
Every vertex starts as a one-vertex task
Each machine begins with one task per vertex it owns, all immediately runnable, because the first step needs only the neighbour list of that vertex and every machine holds the lists of its own vertices. Machine 1 therefore starts work on vertex 0 with no network activity at all.
the full computation is one search tree per vertex in the whole graph, and each machine explores the trees rooted at its own vertices only the FIRST step is guaranteed local, which is why almost everything afterwards depends on how remote fetches are scheduled
Why it works
- A task is one extension rather than one tree → thousands are live at once instead of a few hundred → so there are always enough outstanding requests to batch transfers and overlap them with computation, raising the share of time spent computing to about 59 percent on average.
- Chunks are fixed-size, single-level, and allocated as one block → memory is bounded by one chunk per level and freed in one operation → giving breadth-first parallelism without breadth-first memory growth or fragmentation.
- A child differs from its parent by one vertex → all but one neighbour list is reachable through a parent pointer → so exactly one list is fetched per step, which cut traffic for five-clique mining on LiveJournal from 4.4 terabytes to 33.8 gigabytes.
- Access is power-law skewed, and popular data stays popular → admitting high-degree lists until the cache is full approximates keeping the hottest data → cutting triangle-counting traffic on the UK graph from 57.7 terabytes to 487.3 gigabytes, more than 99 percent, for 3.7 times the speed.
- Every reuse mechanism is structural rather than tracked → no map, no reference counts, no eviction scan, no collision chains → so the system pays about three times more network traffic and still runs 19 times faster.
What it costs (trade-offs and scope)
- It deliberately communicates more. Because sharing happens only between a parent and its descendants, within one chunk, or through the static cache, two tasks in different trees usually cannot share. Roughly three times the traffic of the general cache is the price paid, and on a slower network that trade could invert.
- The wins shrink on graphs that are not skewed. On the least skewed dataset tested, extension work is so light that scheduling takes 45 percent of the time and network 23 percent, the per-chunk sharing saves only about 20 percent of traffic, and the single-node port is slower than the system it was ported from.
- Per-pattern startup is not free. Frequent subgraph mining evaluates many candidate patterns, and initialising the engine for each one makes the single-node configuration consistently slower than the unmodified original. Only the distributed configuration wins there.
- The cache policy is knowingly suboptimal. Never evicting cannot follow a changing access pattern, and on several workloads it moves more data than least-recently-used would. It wins on time, not on traffic, which is a bet that communication stays hidden behind computation.
- It is an engine, not an algorithm. It does not improve pattern matching itself, so a competitor using a specialised algorithmic optimisation for triangles and cliques is far faster on those. The authors note that optimisation could be added as a pre-processing step and disable it for fair comparison.
The evaluation
Two distributed systems were built by porting existing single-machine systems on top of the engine, at roughly 500 lines of code each, against an engine of about 5,000 lines. The main cluster had 8 nodes, each with two 8-core Intel Xeon E5-2630 v3 processors and 64GB of memory, joined by 56Gbps InfiniBand. Nine graphs were used, from MiCo at 1.1 million edges to the Web Data Commons graph at 3.5 billion vertices and 128.7 billion edges. Workloads were triangle counting, motif counting, clique counting and frequent subgraph mining. Because the prior partitioned-graph system lacks support for multi-socket machines, its single-socket times were used to compute speedups, which is the conservative choice.
Mining patterns in a graph too large for one machine
| Property | Replicate the graph everywhere | Partition, whole-tree tasks, general cache | Partition, one-extension tasks |
|---|---|---|---|
| Largest graph possible | |||
| Concurrent units of work | |||
| Network traffic | |||
| Bookkeeping cost | |||
| Speed |
Check yourself
Check your understanding
1. The prior system counted triangles on the Patents graph in 27.2 seconds using 64 cores, while a single-threaded program took 6.2 seconds. Predict what makes 64 cores slower than one.
2. The new system moves roughly THREE TIMES more data across the network than the design it replaces, yet runs 19 times faster. Predict how that is possible.
3. Exploration is breadth-first inside a fixed-size chunk but depth-first between chunks. Predict why neither pure strategy would work.
4. The static cache admits high-degree neighbour lists until it is full and then never evicts anything. Predict why this beats least-recently-used by roughly an order of magnitude.
Sources
- Kudu: An Efficient and Scalable Distributed Graph Pattern Mining Engine (arXiv:2105.03789v3) — the full paper, read in full for this explainer. It was published at ASPLOS 2023 under the name Khuzdul; the preprint carries the earlier name Kudu.
- ACM Digital Library (DOI 10.1145/3575693.3575743) — published version.