incremental

package module
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 26, 2026 License: Apache-2.0 Imports: 10 Imported by: 0

README

go-incremental

CI

A Go fork of Jane Street's Incremental - a library for self-adjusting computation. You build a DAG of nodes, Observe the ones you care about, and call Stabilize whenever inputs change; only the nodes actually affected by the change get recomputed.

Divergence from the OCaml original

Feature set is 1:1 parity: Var, Map/MapN, Bind, Join, IfThenElse, Freeze, Both, ArrayFold, UnorderedArrayFold, ReduceBalanced, cutoffs, At/AtIntervals/BeforeOrAfter, Snapshot, StepFunction, Expert nodes, and debug/introspection tooling (ForAnalyzer, dot-graph export).

The execution engine is not a port - it's a redesign. The OCaml original recomputes on a single thread via a global height-ordered heap. This port replaces that with a concurrent, non-blocking wavefront scheduler (see below). Node identity uses Go generics (Node[T]) instead of OCaml's existential wrapping.

Concurrency model

Every node's compute function is assumed pure: same inputs, same output, no side effects other than the returned value. This is a hard contract the engine relies on for correctness, not just performance.

Stabilization is seed-driven and costs O(affected subgraph), not O(graph):

  • every node carries an exact necessary refcount, maintained incrementally on observe/unobserve and edge changes (a node is necessary iff an observer transitively reads it);
  • between rounds the State accumulates dirty seeds - Var.Sets, clock-driven staleness, nodes that just became necessary - and a round's work set is the upward closure of those seeds through necessary parents;
  • a per-round coordinator goroutine owns all round bookkeeping and feeds ready nodes to workers over channels; a node recomputes only if one of its in-round children actually changed (cutoffs stop propagation for free);
  • a round is handed to the worker pool only when it is worth it: the engine measures what recomputing a node actually costs and dispatches only if spreading the round beats running it on one goroutine. Cheap graphs stay serial on their own, so NewState(0) is a safe default rather than a gamble; pass NewState(1) to rule out concurrency entirely.

Nodes travel to workers in batches, since a rendezvous each way costs far more than a cheap node's recompute. On an Apple M3 Pro that puts the crossover near 0.3 us of work per node: below it rounds stay serial, and a wide round of 12 us nodes runs 7.5x faster on twelve workers. Numbers and method: benchmarks/README.md.

Because compute functions are pure, a round's result is identical to what a serial, height-ordered recompute would have produced, regardless of how the scheduler interleaved work. Height is still tracked and used to correctly reshape the graph under Bind, it just no longer drives recompute order.

Concurrency follows an actor model rather than shared locks: one scheduler goroutine (the graph actor) owns roots, seeds, necessity cascades, and topology changes; round coordinators own round state; workers only compute and message. State.Stabilize() queues a round and returns a *Future immediately; State.StabilizeAndWait() is the blocking convenience wrapper. Var.Set never blocks: it hands the change to the actor and returns, with channel FIFO ordering guaranteeing a subsequent stabilization sees it.

Usage

s := incremental.NewState(0) // 0 -> GOMAXPROCS workers; 1 -> always serial
defer s.Close()

x := incremental.NewVar(s, 1)
y := incremental.NewVar(s, 2)
sum := incremental.Map2(x.Watch(), y.Watch(), func(a, b int) int { return a + b }, incremental.NeverCutoff[int]())

obs := incremental.Observe(s, sum)
defer obs.Unobserve()

if err := s.StabilizeAndWait(); err != nil {
    // handle err
}
v, _ := obs.Value() // 3

x.Set(10)
_ = s.StabilizeAndWait()
v, _ = obs.Value() // 12

Collections (IncrMap)

Scalar nodes are the wrong shape for relational data. The collection layer propagates per-key diffs between map-valued nodes - O(keys changed), never O(rows):

orders := incremental.NewMapVar[int, Order](s)
big := incremental.MapFilter(incremental.MapNode[int, Order](orders),
	func(_ int, o Order) bool { return o.Amount >= 100 })
byRegion := incremental.MapGroupAggregate(big,
	func(_ int, o Order) string { return o.Region },
	0.0,
	func(a float64, o Order) float64 { return a + o.Amount },
	func(a float64, o Order) float64 { return a - o.Amount },
)
obs := incremental.ObserveMap(s, byRegion)
obs.OnDiffs(func(diffs []incremental.MapDiff[string, float64]) { /* push to clients */ })

100k rows through filter + group-sum: one changed row stabilizes in ~7 us (8.7x faster than a full recompute).

The operator set covers MapFilter, MapValues, three flavours of grouped aggregate, and equi-joins:

operator aggregate shape cost per changed source entry
MapGroupAggregate invertible (sum, count) O(1)
MapGroupMinBy / MapGroupMaxBy extremes over an ordered projection O(1), except a removal of the current extreme (one pass over the group's distinct values)
MapGroupReduce any associative+commutative fold, no inverse O(1) insert; a removal or in-place change refolds that group once per round

Joins come in three shapes, all costing O(keys changed) rather than O(inputs), and all emitting a single diff for a key that moves on both sides in one round:

operator join keyed by
MapJoin inner equi-join on the key the shared key
MapLeftJoin left outer equi-join on the key the left key
MapLookupJoin foreign-key lookup (stream/table enrichment) the source key

MapLookupJoin is the stream/table shape: source entries carry a foreign key, one table entry may serve many of them. It keeps a reverse index from table key to referring source keys, so a table change revisits only the rows that actually reference it - O(source keys changed + rows referencing changed table keys).

Windowing

There is deliberately no window operator. Windowing is a retention policy plus time semantics - which time (event or processing), what watermark, what to do with late data, when to emit - and those are choices a general-purpose library should not make for you. Neither the OCaml original nor Salsa has them either.

It composes from what is here:

  • the event buffer is a MapVar keyed by event id;
  • eviction is Delete on entries that fell out of the window, driven either by your ingest loop or by an AtIntervals alarm on a Clock - and the clock is logical, advanced by AdvanceClock, so feed timestamps can drive it as a watermark and tests stay deterministic;
  • the aggregate is any of the grouped operators above, over that buffer;
  • tumbling or sliding buckets fall out of the group key: G is any comparable, so struct{Bucket int64; Key K} works.

Eviction is the operation a window does constantly, which is why MapGroupMinBy/MapGroupMaxBy treat it as the fast path: removing a member that is not the current extreme is O(1).

Benchmarks

benchmarks/ compares this library against naive full recompute, hand-rolled dirty-flag memoization, and RxGo v2 on shared, correctness-gated scenarios. Headline (portfolio pricing: instrument prices -> sector sums -> total, Apple M3 Pro, Go 1.26, median of 3):

scenario, 1 changed input incremental naive full recompute
10k instruments, 101 outputs observed 12.9 us 8.9 us
100k instruments, total observed 10.3 us 113 us

Incremental's cost is flat in graph size (the 100k-input graph stabilizes a sparse change faster than the 10k one); naive's is linear. With trivially cheap per-node compute the break-even sits around ten thousand inputs on this machine - real per-node work moves it sharply lower. When most inputs change per round, full recompute wins instead. Full tables (including the cases this library loses), methodology, and fairness notes: benchmarks/README.md.

Development

go test -race ./...
go vet ./...
gofmt -l .

License

Apache License 2.0 - see LICENSE. The design follows Jane Street's Incremental (see NOTICE); the implementation is original.

Documentation

Overview

Package incremental is a Go port of Jane Street's Incremental, a library for self-adjusting computation: build a DAG of nodes (Var, Map/MapN, Bind, Join, IfThenElse, Freeze, ArrayFold, UnorderedArrayFold, ReduceBalanced, At/AtIntervals, Snapshot, StepFunction, Expert...), Observe the nodes you care about, and call Stabilize to bring every observed node's value up to date with whatever Vars changed since the last round - recomputing only the nodes actually affected.

Purity and determinism contract

Every node's compute function is assumed pure: given the same inputs, it always produces the same output, with no observable side effects other than its return value. This is a hard contract, not a suggestion - the engine relies on it for correctness, not just performance.

Concurrency model

Unlike the OCaml original's single-threaded, height-ordered recompute heap, this port stabilizes with a wavefront dataflow scheduler: each dirty node tracks an atomic count of not-yet-settled recompute dependencies, and becomes eligible for a worker pool the moment that count reaches zero, so independent nodes recompute concurrently. Because compute functions are pure, the result of a round is identical to what a serial, height-ordered recompute would have produced, regardless of how the scheduler happened to interleave work. Dispatching costs more than a cheap node saves, so the engine measures per-node recompute cost and keeps such rounds serial by itself.

State.Stabilize is non-blocking: it queues a round and returns a *Future immediately, so a caller never blocks on recompute. State.StabilizeAndWait blocks until that round (and any rounds queued ahead of it) complete - the common choice for tests and simple call sites. All graph mutations - AddRoot/RemoveRoot, Var.Set, update listeners - are serialized through a single scheduler goroutine, so the graph is never observed in a half-mutated state, without requiring coarse locking around it.

Getting started 

s := incremental.NewState(0) // 0 -> serial; pass a count for concurrent recompute
defer s.Close()

x := incremental.NewVar(s, 1)
y := incremental.NewVar(s, 2)
sum := incremental.Map2(x.Watch(), y.Watch(), func(a, b int) int { return a + b }, incremental.NeverCutoff[int]())

obs := incremental.Observe(s, sum)
defer obs.Unobserve()

if err := s.StabilizeAndWait(); err != nil {
	// handle err
}
v, _ := obs.Value() // 3

x.Set(10)
_ = s.StabilizeAndWait()
v, _ = obs.Value() // 12
Example

Example shows the core loop: wrap inputs in Vars, derive with combinators, Observe what you care about, and Stabilize after changes - only the affected subgraph recomputes. 

package main

import (
	"fmt"

	incr "github.com/PrositAS/go-incremental"
)

func main() {
	s := incr.NewState(0) // 0 -> GOMAXPROCS workers
	defer s.Close()

	price := incr.NewVar(s, 100.0)
	qty := incr.NewVar(s, 3.0)
	total := incr.Map2(price.Watch(), qty.Watch(), func(p, q float64) float64 { return p * q })

	obs := incr.Observe(s, total)
	defer obs.Unobserve()

	if err := s.StabilizeAndWait(); err != nil {
		panic(err)
	}
	v, _ := obs.Value()
	fmt.Println(v)

	price.Set(110)
	if err := s.StabilizeAndWait(); err != nil {
		panic(err)
	}
	v, _ = obs.Value()
	fmt.Println(v)

}

Output:
300
330
Example (Collections)

Example_collections shows the collection layer: a MapVar of rows flowing through a filter and a group-aggregate, observed as a live view. Each row change propagates as per-key diffs - O(keys changed), never O(rows). 

package main

import (
	"fmt"
	"sort"

	incr "github.com/PrositAS/go-incremental"
)

// Example_collections shows the collection layer: a MapVar of rows flowing
// through a filter and a group-aggregate, observed as a live view. Each row
// change propagates as per-key diffs - O(keys changed), never O(rows).
func main() {
	s := incr.NewState(0)
	defer s.Close()

	type Order struct {
		Region string
		Amount float64
	}

	orders := incr.NewMapVar[int, Order](s)
	big := incr.MapFilter(orders.Watch(), func(_ int, o Order) bool { return o.Amount >= 100 })
	byRegion := incr.MapGroupAggregate(big,
		func(_ int, o Order) string { return o.Region },
		0.0,
		func(a float64, o Order) float64 { return a + o.Amount },
		func(a float64, o Order) float64 { return a - o.Amount },
	)

	view := incr.ObserveMap(s, byRegion)
	defer view.Unobserve()

	orders.Set(1, Order{"eu", 120})
	orders.Set(2, Order{"eu", 80}) // below threshold: filtered out
	orders.Set(3, Order{"us", 200})
	if err := s.StabilizeAndWait(); err != nil {
		panic(err)
	}
	printSorted(view.Snapshot())

	orders.Set(2, Order{"eu", 150}) // enters the filter
	orders.Delete(3)
	if err := s.StabilizeAndWait(); err != nil {
		panic(err)
	}
	printSorted(view.Snapshot())

}

func printSorted(m map[string]float64) {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for i, k := range keys {
		if i > 0 {
			fmt.Print(" ")
		}
		fmt.Printf("%s=%g", k, m[k])
	}
	fmt.Println()
}

Output:
eu=120 us=200
eu=270

Index

Examples

Constants

View Source 
const (
	// Necessary is delivered the first time a handler observes a node's value: when the
	// handler is registered, or when the node becomes necessary again after being
	// Unnecessary.
	Necessary = core.Necessary
	// Changed is delivered whenever the node's value actually changes.
	Changed = core.Changed
	// Invalidated is delivered once, when the observed node becomes invalid.
	Invalidated = core.Invalidated
	// Unnecessary is delivered once, when the observed node stops being necessary.
	Unnecessary = core.Unnecessary
)

Variables

View Source 
var ErrObserverNotStabilized = errors.New("incremental: observer has no value yet; stabilize first")

ErrObserverNotStabilized is returned by Observer.Value before the observed node has ever been computed by a stabilization round.

View Source 
var ErrObserverUnlinked = errors.New("incremental: observer use disallowed after Unobserve")

ErrObserverUnlinked is returned by Observer.Value and Observer.OnUpdate after Unobserve.

Functions

func AddDependency 

func AddDependency[T, U any](n *ExpertNode[T], dep *Dependency[U])

AddDependency attaches dep as a new child of n and forces n stale so it recomputes with the new dependency in scope - the Go counterpart of Incremental's Expert.Node.add_dependency. If dep's child already has a value and n isn't about to re-fire every callback anyway (see SetObservable), dep.OnChange fires immediately with that value.

func All 

func All[T any](inputs []core.ValueNode[T]) core.ValueNode[[]T]

All creates a node whose value is the slice of every input's current value, in order - the Go counterpart of Incremental's all.

func AppendUserInfo 

func AppendUserInfo(node core.AnyNode, info *DotUserInfo)

AppendUserInfo appends info to node's existing user info, or sets it if node has none yet - the Go counterpart of Incremental's append_user_info_graphviz.

func ArrayFold 

func ArrayFold[A, Acc any](children []core.ValueNode[A], init Acc, f func(Acc, A) Acc, opts ...NodeOpt[Acc]) core.ValueNode[Acc]

ArrayFold creates a node whose value folds f over the current values of children in order, starting from init - the Go counterpart of Incremental's array_fold. If children is empty, the result is a Const holding init, mirroring the OCaml original.

func AtLeastKOf 

func AtLeastKOf(inputs []core.ValueNode[bool], k int) core.ValueNode[bool]

AtLeastKOf creates a node reporting whether at least k of inputs currently hold true - the Go counterpart of Incremental's at_least_k_of. It recomputes in O(1) amortized time per change via UnorderedArrayFold's inverse-update path.

func Bind 

func Bind[A, B any](lhs core.ValueNode[A], f func(A) core.ValueNode[B]) core.ValueNode[B]

Bind creates a node whose right-hand side is rebuilt by calling f every time lhs's value changes - the Go counterpart of Incremental's bind. Each time lhs changes, f runs again against its new value to build a new right-hand side node, which is detached from the graph and replaced by the new one, spliced in in its place.

Unlike the OCaml original, nodes f constructs are not implicitly scoped to the bind's right-hand side: OCaml tracks them via a mutable "current scope" pushed for the duration of f, which is unsafe to share across the concurrently-recomputing goroutines of this engine. Consequently, when a right-hand side is superseded it is only detached - its edge to the bind's main node is removed - and never marked Invalid. This diverges from OCaml, which invalidates and tears down the whole bind scope, and it carries a known cost. A detached right-hand side is unreachable from every root, so it is never recomputed again, but it is NOT necessarily garbage-collected: edges are two-way, so any still-live node it depends on keeps it in that node's parent list. An abandoned right-hand side that references a longer-lived input - the common case, e.g. f building Map(sharedVar, ...) - therefore stays pinned to that input, so a graph that refires binds accumulates one phantom parent per superseded right-hand side on the shared input: an unbounded leak for long-running graphs. We accept this because the safe alternative is worse without scopes: invalidating the old right-hand side (as an earlier version did) permanently breaks the reuse pattern where f returns a pre-existing node - e.g. one of two long-lived nodes chosen by a boolean - since nothing here re-validates a node, and without scopes we cannot tell a fresh intermediate apart from a reused one. Proper teardown needs the bind-scope machinery in internal/core/scope.go, currently unwired because an ambient current-scope is unsafe under this engine's concurrent recompute.

func Both 

func Both[A, B any](a core.ValueNode[A], b core.ValueNode[B]) core.ValueNode[Pair[A, B]]

Both creates a node whose value pairs a and b's current values - the Go counterpart of Incremental's both. If a and b are both Const nodes, the result is itself a Const, mirroring the OCaml original's optimization to avoid an unnecessary Map2 node.

func Const 

func Const[T any](value T) *core.Node[T]

Const creates a node whose value is fixed at creation and never recomputed - the Go counterpart of Incremental's const. It is never Recomputable: like a Var, it settles immediately whenever the scheduler reaches it, since core.Node's generic IsStale treats "never recomputed" (StabilizationNum none) as the only staleness condition once a value has been assigned this way.

func Exists 

func Exists(inputs []core.ValueNode[bool]) core.ValueNode[bool]

Exists creates a node reporting whether any of inputs currently holds true - the Go counterpart of Incremental's exists.

func ForAll 

func ForAll(inputs []core.ValueNode[bool]) core.ValueNode[bool]

ForAll creates a node reporting whether every one of inputs currently holds true - the Go counterpart of Incremental's for_all.

func Freeze 

func Freeze[T any](child core.ValueNode[T], onlyFreezeWhen func(T) bool) core.ValueNode[T]

Freeze creates a node that copies child's value on every recompute until onlyFreezeWhen returns true for that value; from then on it detaches from child and keeps that value forever, so it never recomputes again - the Go counterpart of Incremental's freeze. onlyFreezeWhen defaults to "always true" (freeze on the first stabilization) when nil.

func IfThenElse 

func IfThenElse[T any](test core.ValueNode[bool], thenBranch, elseBranch core.ValueNode[T]) core.ValueNode[T]

IfThenElse creates a node whose value tracks thenBranch's or elseBranch's current value, following test - the Go counterpart of Incremental's if_then_else. It switches which branch it depends on (rather than reading both every recompute) whenever test's value changes, so the branch not taken doesn't need to be kept up to date.

func IncrementalStepFunction 

func IncrementalStepFunction[T any](clock *Clock, child core.ValueNode[StepFunctionValue[T]]) core.ValueNode[T]

IncrementalStepFunction creates a node whose value follows child's step function, advancing at each step's time as the clock passes it, and re-extracting the step function whenever child changes - the Go counterpart of Incremental's Clock.incremental_step_function.

func Join 

func Join[T any](input core.ValueNode[core.ValueNode[T]]) core.ValueNode[T]

Join creates a node whose value tracks input's current value, unwrapping one level of dynamism - the Go counterpart of Incremental's join. Where Bind rebuilds its right-hand side by running a function whenever the left-hand side changes, Join's right-hand side is simply whatever node input currently points to: input.Value() is itself a node, and Join's result mirrors that node's value, updating whenever input points somewhere new.

func Map 

func Map[T, R any](input core.ValueNode[T], f func(T) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map creates a node whose value is f applied to input's current value - the Go counterpart of Incremental's map. cutoff decides whether a recompute that produces an unchanged-per-cutoff value should stop propagating to the result's parents.

func Map2 

func Map2[A1, A2, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], f func(A1, A2) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map2 creates a node whose value is f applied to the current values of in1, in2 - the Go counterpart of Incremental's map2.

func Map3 

func Map3[A1, A2, A3, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], f func(A1, A2, A3) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map3 creates a node whose value is f applied to the current values of in1, in2, in3 - the Go counterpart of Incremental's map3.

func Map4 

func Map4[A1, A2, A3, A4, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], f func(A1, A2, A3, A4) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map4 creates a node whose value is f applied to the current values of in1, in2, in3, in4 - the Go counterpart of Incremental's map4.

func Map5 

func Map5[A1, A2, A3, A4, A5, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], f func(A1, A2, A3, A4, A5) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map5 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5 - the Go counterpart of Incremental's map5.

func Map6 

func Map6[A1, A2, A3, A4, A5, A6, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], f func(A1, A2, A3, A4, A5, A6) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map6 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6 - the Go counterpart of Incremental's map6.

func Map7 

func Map7[A1, A2, A3, A4, A5, A6, A7, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], in7 core.ValueNode[A7], f func(A1, A2, A3, A4, A5, A6, A7) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map7 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6, in7 - the Go counterpart of Incremental's map7.

func Map8 

func Map8[A1, A2, A3, A4, A5, A6, A7, A8, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], in7 core.ValueNode[A7], in8 core.ValueNode[A8], f func(A1, A2, A3, A4, A5, A6, A7, A8) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map8 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6, in7, in8 - the Go counterpart of Incremental's map8.

func Map9 

func Map9[A1, A2, A3, A4, A5, A6, A7, A8, A9, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], in7 core.ValueNode[A7], in8 core.ValueNode[A8], in9 core.ValueNode[A9], f func(A1, A2, A3, A4, A5, A6, A7, A8, A9) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map9 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6, in7, in8, in9 - the Go counterpart of Incremental's map9.

func Map10 

func Map10[A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], in7 core.ValueNode[A7], in8 core.ValueNode[A8], in9 core.ValueNode[A9], in10 core.ValueNode[A10], f func(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map10 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6, in7, in8, in9, in10 - the Go counterpart of Incremental's map10.

func Map11 

func Map11[A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], in7 core.ValueNode[A7], in8 core.ValueNode[A8], in9 core.ValueNode[A9], in10 core.ValueNode[A10], in11 core.ValueNode[A11], f func(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map11 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6, in7, in8, in9, in10, in11 - the Go counterpart of Incremental's map11.

func Map12 

func Map12[A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], in7 core.ValueNode[A7], in8 core.ValueNode[A8], in9 core.ValueNode[A9], in10 core.ValueNode[A10], in11 core.ValueNode[A11], in12 core.ValueNode[A12], f func(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map12 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6, in7, in8, in9, in10, in11, in12 - the Go counterpart of Incremental's map12.

func Map13 

func Map13[A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], in7 core.ValueNode[A7], in8 core.ValueNode[A8], in9 core.ValueNode[A9], in10 core.ValueNode[A10], in11 core.ValueNode[A11], in12 core.ValueNode[A12], in13 core.ValueNode[A13], f func(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map13 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6, in7, in8, in9, in10, in11, in12, in13 - the Go counterpart of Incremental's map13.

func Map14 

func Map14[A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], in7 core.ValueNode[A7], in8 core.ValueNode[A8], in9 core.ValueNode[A9], in10 core.ValueNode[A10], in11 core.ValueNode[A11], in12 core.ValueNode[A12], in13 core.ValueNode[A13], in14 core.ValueNode[A14], f func(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map14 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6, in7, in8, in9, in10, in11, in12, in13, in14 - the Go counterpart of Incremental's map14.

func Map15 

func Map15[A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15, R any](in1 core.ValueNode[A1], in2 core.ValueNode[A2], in3 core.ValueNode[A3], in4 core.ValueNode[A4], in5 core.ValueNode[A5], in6 core.ValueNode[A6], in7 core.ValueNode[A7], in8 core.ValueNode[A8], in9 core.ValueNode[A9], in10 core.ValueNode[A10], in11 core.ValueNode[A11], in12 core.ValueNode[A12], in13 core.ValueNode[A13], in14 core.ValueNode[A14], in15 core.ValueNode[A15], f func(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15) R, opts ...NodeOpt[R]) core.ValueNode[R]

Map15 creates a node whose value is f applied to the current values of in1, in2, in3, in4, in5, in6, in7, in8, in9, in10, in11, in12, in13, in14, in15 - the Go counterpart of Incremental's map15.

func ReduceBalanced 

func ReduceBalanced[A, Acc any](children []core.ValueNode[A], f func(A) Acc, reduce func(Acc, Acc) Acc, opts ...NodeOpt[Acc]) (result core.ValueNode[Acc], ok bool)

ReduceBalanced creates a node whose value combines f applied to every child's value via a balanced binary reduction tree - the Go counterpart of Incremental's reduce_balanced. f transforms each leaf; reduce combines two subtree results. Unlike ArrayFold, a change to one child only dirties the O(log n) nodes on its path to the root rather than the whole fold, at the cost of requiring reduce to be associative. ok is false (and the returned node nil) when children is empty, mirroring the OCaml original's `t option`.

func RemoveDependency 

func RemoveDependency[T, U any](n *ExpertNode[T], dep *Dependency[U])

RemoveDependency detaches dep from n and forces n stale so it recomputes without it - the Go counterpart of Incremental's Expert.Node.remove_dependency.

func SaveDot 

func SaveDot(roots []core.AnyNode, emitBindEdges bool) string

SaveDot renders the graph reachable from roots (by following child/recompute-dependency edges) as Graphviz "dot" source - the Go counterpart of Incremental's Node_to_dot.save_dot.

emitBindEdges is accepted for API parity with the OCaml original, which additionally draws a dashed edge from a bind's dynamically-created right-hand-side nodes back to the bind. This port's Bind (see bind.go) does not retain that set - only the node f last returned is tracked - so there is nothing to draw and this parameter is currently a no-op.

func Snapshot 

func Snapshot[T any](clock *Clock, valueAt core.ValueNode[T], at time.Time, before T) (core.ValueNode[T], error)

Snapshot creates a node that holds "before" until the clock reaches "at", at which point it takes on valueAt's value at that moment and keeps it forever - the Go counterpart of Incremental's Clock.snapshot. It errors if "at" is strictly before the clock's current time; if "at" already equals the current time, it degenerates to Freeze(valueAt, always).

func StepFunction 

func StepFunction[T any](clock *Clock, init T, steps []Step[T]) core.ValueNode[T]

StepFunction creates a node whose value starts at init and follows steps as the clock advances - the Go counterpart of Incremental's Clock.step_function. It is a convenience over IncrementalStepFunction for a step function fixed at creation.

func Sum 

func Sum[T any](inputs []core.ValueNode[T], zero T, add func(T, T) T, sub func(T, T) T, fullComputeEveryNChanges int) core.ValueNode[T]

Sum creates a node folding add over the current values of inputs, starting from zero, updating incrementally via sub (add's inverse) on each change - the Go counterpart of Incremental's sum. fullComputeEveryNChanges <= 0 means "never force a periodic full recompute beyond the first"; pass a positive value to bound floating-point drift.

func UnorderedArrayFold 

func UnorderedArrayFold[A, Acc any](
	children []core.ValueNode[A],
	init Acc,
	f func(Acc, A) Acc,
	update UnorderedArrayFoldUpdate[A, Acc],
	fullComputeEveryNChanges int,
	opts ...NodeOpt[Acc],
) core.ValueNode[Acc]

UnorderedArrayFold creates a node that folds f over the current values of children, order unspecified - the Go counterpart of Incremental's unordered_array_fold. Instead of a full recompute on every change, it applies update incrementally to just the children that changed since the last recompute, falling back to a full recompute every fullComputeEveryNChanges incremental updates (and on the first compute) to bound floating-point/accumulated error. fullComputeEveryNChanges <= 0 means "never force a periodic full recompute beyond the first". If children is empty, the result is a Const holding init.

Types

type BeforeOrAfter 

type BeforeOrAfter int

BeforeOrAfter is the value of an At/After incremental: Before until the clock reaches the target time, After from then on - the Go counterpart of Incremental's Before_or_after.t. 

const (
	Before BeforeOrAfter = iota
	After
)

func (BeforeOrAfter) String 

func (b BeforeOrAfter) String() string

String renders b as "Before" or "After".

type Clock 

type Clock struct {
	// contains filtered or unexported fields
}

Clock drives every time-based incremental (At, AtIntervals, Snapshot, StepFunction/IncrementalStepFunction): a logical clock that only moves when AdvanceClock/AdvanceClockBy is called, so stabilization stays deterministic and test-controllable instead of racing wall-clock time. It is the Go counterpart of Incremental's Clock.

func NewClock 

func NewClock(state *engine.State, start time.Time) *Clock

NewClock creates a Clock owned by state, starting at start.

func (*Clock) AdvanceClock 

func (c *Clock) AdvanceClock(to time.Time) error

AdvanceClock moves the clock forward to "to" and fires every alarm scheduled at or before "to", in ascending time order. Firing an alarm may itself schedule another one still at or before "to" (e.g. AtIntervals catching up several elapsed intervals in one big jump); AdvanceClock keeps draining until no pending alarm is due anymore, mirroring Incremental's advance_clock followed by fire_past_alarms. It is a no-op if "to" is not after the clock's current time. Like Incremental's advance_clock, it refuses to run while a stabilization round is in flight (ErrStabilizing) unless called reentrantly from an on-update handler.

func (*Clock) AdvanceClockBy 

func (c *Clock) AdvanceClockBy(span time.Duration) error

AdvanceClockBy moves the clock forward by span; see AdvanceClock.

func (*Clock) After 

func (c *Clock) After(span time.Duration) core.ValueNode[BeforeOrAfter]

After creates a node whose value is Before until span has elapsed on the clock, and After from then on - the Go counterpart of Incremental's Clock.after.

func (*Clock) At 

func (c *Clock) At(at time.Time) core.ValueNode[BeforeOrAfter]

At creates a node whose value is Before until the clock reaches "at", and After from then on - the Go counterpart of Incremental's Clock.at. If "at" is already at or before the clock's current time, the result is a plain Const(After).

func (*Clock) AtIntervals 

func (c *Clock) AtIntervals(interval time.Duration) core.ValueNode[struct{}]

AtIntervals creates a node whose value changes, without ever cutting off, every interval starting from the clock's current time at creation - the Go counterpart of Incremental's Clock.at_intervals. It panics if interval is not positive.

func (*Clock) Now 

func (c *Clock) Now() time.Time

Now returns the clock's current logical time.

func (*Clock) WatchNow 

func (c *Clock) WatchNow() core.ValueNode[time.Time]

WatchNow returns the node that reflects the clock's current time once stabilization has processed an AdvanceClock call.

type Cutoff 

type Cutoff[T any] struct {
	// contains filtered or unexported fields
}

Cutoff decides whether propagation of changes should stop at a node, based on the node's old and new value.

func AlwaysCutoff 

func AlwaysCutoff[T any]() Cutoff[T]

AlwaysCutoff always cuts off propagation.

func CompareCutoff 

func CompareCutoff[T any](compare func(oldValue, newValue T) int) Cutoff[T]

CompareCutoff cuts off propagation when compare returns 0.

func EqualCutoff 

func EqualCutoff[T any](equal func(oldValue, newValue T) bool) Cutoff[T]

EqualCutoff cuts off propagation when equal returns true.

func NeverCutoff 

func NeverCutoff[T any]() Cutoff[T]

NeverCutoff never cuts off propagation; this is also the zero value of Cutoff.

func NewCutoff 

func NewCutoff[T any](f func(oldValue, newValue T) bool) Cutoff[T]

NewCutoff creates a Cutoff from an arbitrary predicate.

func PhysEqualCutoff 

func PhysEqualCutoff[T comparable]() Cutoff[T]

PhysEqualCutoff cuts off propagation when the old and new value compare == equal. This mirrors Incremental's default cutoff (physical equality in OCaml).

func (Cutoff[T]) ShouldCutoff 

func (c Cutoff[T]) ShouldCutoff(oldValue, newValue T) bool

ShouldCutoff reports whether propagation should stop given a node's old and new value.

type Dependency 

type Dependency[T any] struct {
	// contains filtered or unexported fields
}

Dependency is a dynamically-managed edge from an ExpertNode to one of its children - the Go counterpart of Incremental's Expert.Dependency. OnChange, if non-nil, fires with the child's current value both immediately upon being attached (via AddDependency, if the child already has a value) and every time the child's value subsequently changes while attached.

func NewDependency 

func NewDependency[T any](child core.ValueNode[T], onChange func(T)) *Dependency[T]

NewDependency creates a Dependency watching child - the Go counterpart of Incremental's Expert.Dependency.create.

func (*Dependency[T]) Value 

func (d *Dependency[T]) Value() T

Value returns the dependency's child's current value. It is meant to be called only from within the owning ExpertNode's compute function or the dependency's own OnChange callback, where the child is guaranteed to already have a value.

type DotUserInfo 

type DotUserInfo struct {
	// contains filtered or unexported fields
}

DotUserInfo is user-controlled metadata optionally attached to a node, used only by SaveDot for debugging/inspection - the Go counterpart of Incremental's Dot_user_info. Core never reads it. The zero value is not valid; construct one with NewDotUserInfo.

func Append 

func Append(a, b *DotUserInfo) *DotUserInfo

Append combines a and b into a new DotUserInfo: their label rows form a set union, and their attributes are merged with b's values winning on key collision - the Go counterpart of Incremental's Dot_user_info.append.

func NewDotUserInfo 

func NewDotUserInfo(label []string, attributes map[string]string) *DotUserInfo

NewDotUserInfo creates a DotUserInfo with one label row (rendered as a graphviz record whose columns are label's entries) and a set of dot attributes.

func (*DotUserInfo) String 

func (info *DotUserInfo) String(name string) string

String renders info as a graphviz node statement named name, using an "Mrecord" shape - the Go counterpart of Incremental's Dot_user_info.to_string.

type ExpertNode 

type ExpertNode[T any] struct {
	*core.Node[T]
	// contains filtered or unexported fields
}

ExpertNode is a node whose set of children (dependencies) and value can both be updated incrementally by user code, rather than fixed at construction - the Go counterpart of Incremental's Expert/Expert1. f computes the node's value from whatever dependencies are currently attached (typically by reading state that each dependency's OnChange callback keeps up to date, rather than reading the dependencies directly); AddDependency and RemoveDependency may be called at any time, including from within f itself.

func NewExpertNode 

func NewExpertNode[T any](f func() T, onObservabilityChange func(isNowObservable bool)) *ExpertNode[T]

NewExpertNode creates an ExpertNode with no dependencies yet. onObservabilityChange, if non-nil, is called by SetObservable.

func (*ExpertNode[T]) MakeStale 

func (n *ExpertNode[T]) MakeStale()

MakeStale forces n to recompute on the next stabilization even though none of its dependencies changed - the Go counterpart of Incremental's Expert.Node.make_stale.

func (*ExpertNode[T]) OnChildChanged 

func (n *ExpertNode[T]) OnChildChanged(child core.AnyNode)

OnChildChanged implements core.ChildValueObserver: it re-fires the OnChange callback of every attached dependency on child, unless a full re-fire is already pending (see SetObservable) - the Go counterpart of Incremental's Expert.run_edge_callback, called from state.ml's maybe_change_value for every changed child.

func (*ExpertNode[T]) Recompute 

func (n *ExpertNode[T]) Recompute(now core.StabilizationNum, _ core.RoundCtx)

Recompute implements core.Recomputable. If any attached dependency's child is invalid, n invalidates itself instead of computing - the Go counterpart of Incremental's Expert.before_main_computation returning `Invalid. Otherwise, if a full re-fire is pending, every dependency's OnChange fires with its current value before f runs - the Go counterpart of before_main_computation's `Ok branch running every will_fire_all_callbacks callback - and then f computes n's new value.

func (*ExpertNode[T]) SetObservable 

func (n *ExpertNode[T]) SetObservable(isNowObservable bool)

SetObservable notifies n that its observability changed, i.e. whether it currently has a necessary path to some observer - the Go counterpart of Incremental's Expert.observability_change. This engine does not call it automatically; it is exposed for callers that track their own observability. Becoming unobservable arranges for every currently-attached dependency's OnChange to be re-fired in full the next time n recomputes, so callers don't miss changes that happened while n wasn't necessary.

func (*ExpertNode[T]) Watch 

func (n *ExpertNode[T]) Watch() core.ValueNode[T]

Watch returns n as a core.ValueNode, for use as another node's input - the Go counterpart of Incremental's Expert.Node.watch.

type Future 

type Future = engine.Future

Future represents the eventual completion of an asynchronous Stabilize call. See engine.Future: Stabilize never blocks its caller, and Future.Wait is how a caller opts into waiting for a stabilization round to finish.

type Kind 

type Kind int

Kind identifies which combinator produced a node. It exists for introspection and debug/dot output; unlike the OCaml GADT it carries no data itself, since node-specific state lives on the node. 

const (
	KindInvalid Kind = iota
	KindUninitialized
	KindConst
	KindVar
	KindMap
	KindMap2
	KindMap3
	KindMap4
	KindMap5
	KindMap6
	KindMap7
	KindMap8
	KindMap9
	KindMap10
	KindMap11
	KindMap12
	KindMap13
	KindMap14
	KindMap15
	KindBindLHSChange
	KindBindMain
	KindJoinLHSChange
	KindJoinMain
	KindIfTestChange
	KindIfThenElse
	KindFreeze
	KindArrayFold
	KindUnorderedArrayFold
	KindAt
	KindAtIntervals
	KindSnapshot
	KindStepFunction
	KindExpert
	KindMapVar
	KindMapFilter
	KindMapValues
	KindMapGroupAggregate
	KindMapGroupReduce
	KindMapGroupMinBy
	KindMapGroupMaxBy
	KindMapJoin
	KindMapLeftJoin
	KindMapLookupJoin
)

func (Kind) String 

func (k Kind) String() string

String returns the node kind's name, matching the OCaml original's Kind.name.

type MapDiff 

type MapDiff[K comparable, V any] struct {
	Key    K
	Old    V
	New    V
	HasOld bool
	HasNew bool
}

MapDiff is one key's change within a stabilization round. HasOld/HasNew distinguish insert (false/true), update (true/true), delete (true/false).

type MapNode 

type MapNode[K comparable, V any] interface {
	core.ValueNode[uint64] // engine plumbing; the value is a version counter

	Len() int
	Get(k K) (V, bool)
	// ForEach calls fn for every entry until fn returns false. Iteration
	// order is unspecified.
	ForEach(fn func(K, V) bool)
	// contains filtered or unexported methods
}

MapNode is the collection counterpart of a ValueNode: its contents are a map, and what propagates between rounds is the per-key diff list - O(keys changed), never O(map). Reads (Len, Get, ForEach) are safe from any goroutine; like Observer.Value they observe the last completed round.

func MapFilter 

func MapFilter[K comparable, V any](src MapNode[K, V], pred func(K, V) bool) MapNode[K, V]

MapFilter creates a collection node holding the entries of src whose (key, value) satisfy pred.

func MapGroupAggregate 

func MapGroupAggregate[K comparable, V any, G comparable, A any](
	src MapNode[K, V],
	group func(K, V) G,
	init A,
	add func(A, V) A,
	sub func(A, V) A,
) MapNode[G, A]

MapGroupAggregate creates a collection node keyed by group, holding the aggregate of src's entries in that group. add folds a value in; sub must be its inverse (sum, count, ...) - that inverse is what makes each source change O(1) instead of O(group). Groups with no members are removed. For aggregates with no inverse use MapGroupReduce, or MapGroupMinBy/MapGroupMaxBy for extremes over ordered values.

func MapGroupMaxBy 

func MapGroupMaxBy[K comparable, V any, G comparable, W cmp.Ordered](
	src MapNode[K, V],
	group func(K, V) G,
	by func(K, V) W,
) MapNode[G, W]

MapGroupMaxBy is MapGroupMinBy for the largest by(key, value) per group.

func MapGroupMinBy 

func MapGroupMinBy[K comparable, V any, G comparable, W cmp.Ordered](
	src MapNode[K, V],
	group func(K, V) G,
	by func(K, V) W,
) MapNode[G, W]

MapGroupMinBy creates a collection node keyed by group, holding the smallest by(key, value) among that group's entries in src. Groups with no members are removed.

Unlike MapGroupReduce, removals stay cheap: each group keeps a multiset of its projected values, so dropping a value that is not the current extreme is O(1), and dropping the extreme itself costs one pass over the group's distinct values. by and group must be pure - the node recomputes them on the old value to undo an entry, rather than storing a copy per key.

func MapGroupReduce 

func MapGroupReduce[K comparable, V any, G comparable, A any](
	src MapNode[K, V],
	group func(K, V) G,
	init A,
	combine func(A, V) A,
) MapNode[G, A]

MapGroupReduce is MapGroupAggregate for folds that have no inverse (min, max, set union, "any member satisfies p", ...). combine must be associative and commutative: a group's members are folded in unspecified order.

Insertions stay O(1) - combine folds the new value into the group's running aggregate. Removals and in-place value changes cannot be undone without an inverse, so a group that loses or rewrites a member is refolded from its members: O(group size), once per round per touched group however many of its members changed. Prefer MapGroupAggregate when an inverse exists, and MapGroupMinBy/MapGroupMaxBy for extremes over ordered values - both keep removals cheap.

func MapJoin 

func MapJoin[K comparable, L, R, O any](
	left MapNode[K, L],
	right MapNode[K, R],
	combine func(K, L, R) O,
) MapNode[K, O]

MapJoin creates a collection node holding combine(key, l, r) for every key present in both left and right - an incremental inner equi-join on the key.

Cost is O(keys changed on either side), not O(left + right): a round only revisits the keys whose left or right entry moved, reading the other side's settled contents for each. Keys that leave either side are removed from the output.

func MapLeftJoin 

func MapLeftJoin[K comparable, L, R, O any](
	left MapNode[K, L],
	right MapNode[K, R],
	combine func(k K, l L, r R, hasRight bool) O,
) MapNode[K, O]

MapLeftJoin keeps every key of left, passing combine the right entry and whether it was present - the incremental left outer equi-join. When a key arrives on or leaves the right side, only that key's output is recombined.

func MapLookupJoin 

func MapLookupJoin[K comparable, V any, FK comparable, T, O any](
	src MapNode[K, V],
	table MapNode[FK, T],
	on func(K, V) FK,
	combine func(k K, v V, t T, hasRow bool) O,
) MapNode[K, O]

MapLookupJoin enriches every entry of src with the table entry its foreign key points at - the incremental form of a stream/table join, where one table entry may serve many src entries. combine receives the table value and whether it was present, so a src entry with no matching row still produces output (drop those in a downstream MapFilter if an inner join is wanted).

The node keeps a reverse index from table key to the src keys referencing it, so a table change only revisits rows that actually reference it. Cost per round is O(src keys changed + src rows referencing changed table keys): a table entry with many referrers costs proportionally, which is inherent to one-to-many, not an artefact. Output is keyed by src's key space.

on must be pure - the node recomputes it against the old value to un-index an entry, rather than storing a copy of every key's foreign key.

func MapValues 

func MapValues[K comparable, V, W any](src MapNode[K, V], f func(K, V) W) MapNode[K, W]

MapValues creates a collection node mapping every entry of src through f, keeping keys.

type MapObserver 

type MapObserver[K comparable, V any] struct {
	// contains filtered or unexported fields
}

MapObserver keeps a MapNode necessary and provides snapshot and per-round diff access - the collection counterpart of Observer.

func ObserveMap 

func ObserveMap[K comparable, V any](state *State, node MapNode[K, V]) *MapObserver[K, V]

ObserveMap roots node so stabilization keeps it up to date.

func (*MapObserver[K, V]) Get 

func (o *MapObserver[K, V]) Get(k K) (V, bool)

func (*MapObserver[K, V]) Len 

func (o *MapObserver[K, V]) Len() int

Len and Get proxy the observed node.

func (*MapObserver[K, V]) Node 

func (o *MapObserver[K, V]) Node() MapNode[K, V]

func (*MapObserver[K, V]) OnDiffs 

func (o *MapObserver[K, V]) OnDiffs(fn func([]MapDiff[K, V]))

OnDiffs registers fn to run at the end of every round in which the observed map changed, with that round's diffs. fn runs on the scheduler goroutine and must not retain the slice - it is reused next round.

func (*MapObserver[K, V]) Snapshot 

func (o *MapObserver[K, V]) Snapshot() map[K]V

Snapshot returns a copy of the observed map's current contents.

func (*MapObserver[K, V]) Unobserve 

func (o *MapObserver[K, V]) Unobserve()

Unobserve releases the root and stops diff delivery.

type MapVar 

type MapVar[K comparable, V any] struct {
	// contains filtered or unexported fields
}

MapVar is the collection leaf: a map whose entries are set directly. Like Var, mutations never block and take effect at the next stabilization.

func NewMapVar 

func NewMapVar[K comparable, V any](state *State) *MapVar[K, V]

NewMapVar creates an empty MapVar in state's graph.

func (*MapVar[K, V]) Delete 

func (m *MapVar[K, V]) Delete(k K)

Delete schedules k's removal as of the next stabilization.

func (MapVar) ForEach 

func (c MapVar) ForEach(fn func(K, V) bool)

func (MapVar) Get 

func (c MapVar) Get(k K) (V, bool)

func (MapVar) Len 

func (c MapVar) Len() int

func (*MapVar[K, V]) Set 

func (m *MapVar[K, V]) Set(k K, v V)

Set schedules k to hold v as of the next stabilization.

func (*MapVar[K, V]) Watch 

func (m *MapVar[K, V]) Watch() MapNode[K, V]

Watch returns m as a MapNode - the collection counterpart of Var.Watch. Operators with a function argument infer K and V from it and accept m directly; Watch is for call sites with nothing else to infer from (ObserveMap in particular).

type NodeOpt 

type NodeOpt[T any] func(*nodeOptions[T])

NodeOpt configures a combinator's result node. Combinators take options variadically so the common case stays terse - Map(x, f) - while cutoffs remain available: Map(x, f, WithCutoff(EqualCutoff(eq))).

func WithCutoff 

func WithCutoff[T any](c Cutoff[T]) NodeOpt[T]

WithCutoff sets the result node's cutoff: recomputes whose new value the cutoff reports as unchanged do not propagate to parents. The default is NeverCutoff (every recompute counts as a change).

type NodeUpdate 

type NodeUpdate[T any] = core.NodeUpdate[T]

NodeUpdate describes a single on-update-handler notification, delivered at the end of a stabilization round. OldValue is only meaningful when Kind is Changed; NewValue is only meaningful when Kind is Necessary or Changed.

type NodeUpdateKind 

type NodeUpdateKind = core.NodeUpdateKind

NodeUpdateKind classifies what happened to an observed node during a stabilization round.

type Observer 

type Observer[T any] struct {
	// contains filtered or unexported fields
}

Observer is a root of the incremental DAG: for as long as it is linked, the node it observes (and everything that node depends on) is necessary, so Stabilize keeps it up to date. Create one with Observe; call Unobserve when done so the node can become unnecessary again.

func Observe 

func Observe[T any](state *engine.State, node core.ValueNode[T]) *Observer[T]

Observe creates an Observer rooted at node. node becomes (and stays, until Unobserve) necessary.

func (*Observer[T]) Observing 

func (o *Observer[T]) Observing() core.ValueNode[T]

Observing returns the node this observer watches.

func (*Observer[T]) OnUpdate 

func (o *Observer[T]) OnUpdate(fn func(NodeUpdate[T])) error

OnUpdate registers fn to run, in registration order, at the end of every stabilization round for as long as the observer stays linked; a newly registered fn still gets a Necessary delivery for the current value at the next round, even if nothing changes. fn must not call Set or Stabilize synchronously: the scheduler goroutine calls fn while completing a round, so doing so would deadlock, except Set on a Var, which is safe to call reentrantly from an on-update handler.

func (*Observer[T]) Unobserve 

func (o *Observer[T]) Unobserve()

Unobserve disallows future use of o and lets its node become unnecessary again (if no other observer or parent keeps it necessary).

func (*Observer[T]) Value 

func (o *Observer[T]) Value() (T, error)

Value returns the observed node's current value. It errors if the node has not yet been computed by a stabilization round, or if Unobserve has been called.

type Pair 

type Pair[A, B any] struct {
	First  A
	Second B
}

Pair holds the paired result of Both.

type State 

type State = engine.State

State owns an incremental graph's scheduling: registered roots, the worker pool, and the single goroutine that serializes graph mutations. Create one with NewState; call Close when done with it. See the package doc for the concurrency model Stabilize and StabilizeAndWait give you.

func NewState 

func NewState(workers int) *State

NewState creates a State with a worker pool sized to workers, defaulting to runtime.GOMAXPROCS(0) when workers <= 0; pass 1 to rule out concurrent recompute. Rounds only reach the pool when their measured per-node cost justifies it, so a graph of cheap nodes stays serial without being asked to. The returned State owns a goroutine; call Close when done with it.

type Step 

type Step[T any] struct {
	At    time.Time
	Value T
}

Step is one transition of a StepFunctionValue: its value becomes Value starting at At.

type StepFunctionValue 

type StepFunctionValue[T any] struct {
	Init  T
	Steps []Step[T]
}

StepFunctionValue is a function from time.Time to T with a finite number of steps, in nondecreasing time order - the Go counterpart of Incremental's Step_function.t.

func NewStepFunctionValue 

func NewStepFunctionValue[T any](init T, steps []Step[T]) StepFunctionValue[T]

NewStepFunctionValue creates a StepFunctionValue with value init before the first step, stepping to each step's Value at its At time in turn - the Go counterpart of Incremental's Step_function.create_exn. It panics if steps isn't in nondecreasing time order.

func (StepFunctionValue[T]) ValueAt 

func (sf StepFunctionValue[T]) ValueAt(at time.Time) T

ValueAt returns the step function's value at the given time - the Go counterpart of Incremental's Step_function.value.

type UnorderedArrayFoldUpdate 

type UnorderedArrayFoldUpdate[A, Acc any] struct {
	// FInverse removes a child's prior contribution, then f re-adds its new one:
	// update(acc, old, new) = f(FInverse(acc, old), new). Set this when f has a natural
	// inverse (e.g. sum/add and subtract).
	FInverse func(acc Acc, oldValue A) Acc
	// Update computes the new accumulator directly from the old and new child value, and
	// takes precedence over FInverse when both are set.
	Update func(acc Acc, oldValue, newValue A) Acc
}

UnorderedArrayFoldUpdate tells UnorderedArrayFold how to fold a single child's change into the accumulator without a full recompute over every child - the Go counterpart of Incremental's Unordered_array_fold.Update.

type ValueNode 

type ValueNode[T any] = core.ValueNode[T]

ValueNode is the read side of a node in the DAG: anything that produces a value of type T during stabilization. All combinators accept and return ValueNodes; consumers should use this alias rather than reaching into internal packages.

type Var 

type Var[T any] struct {
	// contains filtered or unexported fields
}

Var is a leaf of the incremental DAG: its value is supplied directly by calling Set, rather than derived from other nodes.

func NewVar 

func NewVar[T any](state *engine.State, initial T) *Var[T]

NewVar creates a Var holding initial, watched by a fresh node in state's graph.

func (*Var[T]) Set 

func (v *Var[T]) Set(value T)

Set updates v's value. If called while a stabilization round is in flight, the change is applied once that round completes, so it only becomes visible to the graph starting with the next round - Set never blocks and never races an in-flight round's recompute.

func (*Var[T]) Value 

func (v *Var[T]) Value() T

Value returns the latest value passed to Set (or the initial value), regardless of whether a stabilization round has processed it yet.

func (*Var[T]) Watch 

func (v *Var[T]) Watch() *core.Node[T]

Watch returns the node that reflects v's value once stabilization has processed it. If Set is called while a round is in flight, this node's value lags Value() by one round.

Source Files

View all Source files

Directories

Path Synopsis
internal
core
Package core holds internal data structures shared by the incremental engine that are not part of the public API.
 Package core holds internal data structures shared by the incremental engine that are not part of the public API.
engine
Package engine implements the non-blocking concurrent stabilization engine: a wavefront dataflow scheduler that replaces Incremental's single-threaded, height-ordered recompute heap.
 Package engine implements the non-blocking concurrent stabilization engine: a wavefront dataflow scheduler that replaces Incremental's single-threaded, height-ordered recompute heap.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.