Documentation
¶
Overview ¶
Package engine is the iteration layer of the graph analytics engine (ADR-0229 §SD2): a frontier (Subset) kept sparse or dense, an edge map whose direction is chosen per sweep, and chunked parallelism whose result is bit-identical to the serial one.
The shape is Ligra's (Shun & Blelloch, PPoPP 2013) with GraphIt's separation of direction from algorithm. Two rules make the parallel result a function of the topology alone:
- A dense sweep pulls: destinations are split into contiguous chunks and every write in a chunk lands on a destination that chunk owns, so no two workers write one slot and the fold within a slot runs in id order.
- A reduction across vertices folds fixed-size chunk partials in chunk order, so the summation order does not depend on the worker count.
A sparse push sweep runs serially. It is chosen only when the frontier's out-edges are a small fraction of the arcs, so its cost is bounded by construction, and a serial push needs no atomics to be deterministic.
Index ¶
- Constants
- Variables
- func Backward(g *csr.Graph, d int32, dir DirectionE, buf *[]int32) []int32
- func ContextDone(ctx context.Context) bool
- func Forward(g *csr.Graph, s int32, dir DirectionE, buf *[]int32) []int32
- func MergeSorted(a, b, buf []int32) []int32
- type DirectionE
- type EdgeFuncs
- type EdgeMapOptions
- type Engine
- func (e *Engine) ChunkedJobs(n, chunk int, body func(chunkIdx, lo, hi int))
- func (e *Engine) EdgeMap(g *csr.Graph, frontier *Subset, f EdgeFuncs, opts EdgeMapOptions) (next *Subset, sweep SweepE)
- func (e *Engine) Fanout(k int, body func(w int))
- func (e *Engine) ParallelFor(n int, body func(worker, lo, hi int))
- func (e *Engine) ReduceFloat64(n int, body func(lo, hi int) float64) float64
- func (e *Engine) Workers() int
- type Subset
- type SweepE
Constants ¶
const DefaultDenseFraction = 1.0 / 20
DefaultDenseFraction is Ligra's threshold: the sweep pulls once the frontier's out-edges exceed this fraction of the arcs.
const MinParallel = 4096
MinParallel is the range length below which Engine.ParallelFor runs inline: goroutine fan-out costs more than a few thousand cheap iterations.
const ReduceChunk = 4096
ReduceChunk is the fixed chunk length of Engine.ReduceFloat64. It is a constant, not a function of the worker count, so the fold order — and therefore the bits of the result — do not change with the machine.
Variables ¶
var PackageProps = packageprops.Props{ WASMWASI: packageprops.WASMCompiles, WASMJS: packageprops.WASMCompiles, WASMFreestanding: packageprops.WASMCompiles, }
PackageProps records this package's curated properties (ADR-0080).
Functions ¶
func Backward ¶
Backward returns the vertices whose sweep-direction edge reaches d — the sources a pull over d must consult. buf is used as in Forward.
func ContextDone ¶
ContextDone reports whether ctx is cancelled or past its deadline, without blocking.
func Forward ¶
Forward returns the neighbours of s in the sweep direction, ascending. For DirectionBoth on a directed graph the two rows are merged into *buf, which grows to fit and is kept for the next call; otherwise the graph's own row is returned and buf is untouched.
func MergeSorted ¶
MergeSorted appends the sorted union of two ascending slot lists to buf and returns it.
Types ¶
type DirectionE ¶
type DirectionE uint8
DirectionE selects which adjacency a sweep follows.
const ( // DirectionOut follows arcs s→d from the frontier to its out-neighbours. DirectionOut DirectionE = iota // DirectionIn follows arcs in reverse, from the frontier to its // in-neighbours. DirectionIn // DirectionBoth follows both, treating the graph as undirected. DirectionBoth )
type EdgeFuncs ¶
EdgeFuncs are the per-edge callbacks of an edge map, after Ligra.
Update is applied along an edge from frontier vertex s to d and reports whether d joins the next frontier. Under a pull sweep Update runs for every in-edge of d whose source is in the frontier, in ascending source order, and must write only state indexed by d; it may read source state that the sweep does not write. Under a push sweep it runs serially in ascending (s, d) order.
Cond reports whether d is still a candidate; nil means every vertex is. A pull sweep stops walking d's in-edges as soon as Cond(d) turns false after an Update, which is what makes a pull BFS linear in the visited edges.
type EdgeMapOptions ¶
type EdgeMapOptions struct {
// Direction selects the adjacency followed. Zero is DirectionOut.
Direction DirectionE
// DenseFraction overrides [DefaultDenseFraction]; a value of 1 or more
// never pulls, a value of 0 always pulls.
DenseFraction float64
// ForcePull or ForcePush fix the sweep regardless of the frontier size.
ForcePull bool
ForcePush bool
}
EdgeMapOptions tunes one edge map.
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine carries the worker count and reusable scratch for the sweeps.
func New ¶
New returns an engine with the given worker count; zero or less means runtime.GOMAXPROCS(0).
func (*Engine) ChunkedJobs ¶
ChunkedJobs runs body over jobs [0, n) in fixed chunks of size chunk that workers claim in any order; the caller folds per-chunk outputs in chunk index order to keep the result independent of scheduling. body receives the chunk index and its job range.
func (*Engine) EdgeMap ¶
func (e *Engine) EdgeMap(g *csr.Graph, frontier *Subset, f EdgeFuncs, opts EdgeMapOptions) (next *Subset, sweep SweepE)
EdgeMap applies f along the edges leaving frontier under opts and returns the next frontier and which sweep ran. The next frontier is a fresh Subset; the input is untouched.
func (*Engine) Fanout ¶
Fanout runs body once per index in [0, k), each on its own goroutine when the engine has more than one worker; body(w) must write only to state it owns for index w.
func (*Engine) ParallelFor ¶
ParallelFor runs body over [0, n) split into up to Workers contiguous chunks, one goroutine each; below MinParallel or with one worker it runs inline. body must write only to state indexed within [lo, hi).
func (*Engine) ReduceFloat64 ¶
ReduceFloat64 sums body over [0, n) with a summation order independent of the worker count: body is evaluated per ReduceChunk-sized chunk, chunk partials are stored at their index, and the partials are folded in chunk order. body(lo, hi) must return the chunk's own sum, folded in index order.
type Subset ¶
type Subset struct {
// contains filtered or unexported fields
}
Subset is a vertex subset — Ligra's vertexSubset — held either as a sorted slot list (sparse) or as a membership bitmap (dense). Both forms may be present; conversions are cached until the next mutation.
func FromSlots ¶
FromSlots returns a subset holding the given slots; they are sorted and deduplicated so the representation is canonical.
func (*Subset) Dense ¶
Dense returns the membership bitmap, converting from the list if needed. Shared; do not modify.
Source Files