ReactiveDAG 1.1.1

ReactiveDAG

A reactive DAG engine for .NET 8. You define inputs and computations, wire them together, and changes propagate through the graph automatically.

Good for simulations, multi-step calculations where intermediate results get reused, build/task orchestration, or anywhere you'd reach for a spreadsheet-like dependency model.

Architecture

There are two ways to use this:

DagPipelineBuilder — fluent API for building pipelines. You chain .AddInput().AddFunction() and it tracks the wiring for you. Best for straightforward linear/fan-in pipelines.

DagEngine — the underlying engine. Use it directly when you need explicit dependency arrays, node removal, streaming, or want to inspect the graph at runtime.

Under the hood, Cell<T> holds a value and DagNode<T> wraps it with computation logic. When an input changes, the engine walks dependents and recomputes what's needed.

Components

• DagEngine — manages nodes, propagates updates, detects cycles, handles concurrency
• DagPipelineBuilder — fluent builder on top of DagEngine
• Cell<T> — input cells hold values directly; function cells get their values from a compute function
• DagNode<T> — wraps a cell, tracks dependencies, handles lazy/on-demand computation

API Reference

Creating Inputs

// Via builder
builder.AddInput(42, out Cell<int> cell);

// Via engine directly
Cell<int> cell = engine.AddInput(42);

Adding Functions

// All inputs share a type
builder.AddFunction<int, int>(async inputs => inputs.Sum(), out var sum);

// Mixed types — pass explicit BaseCell[] dependencies
engine.AddFunction<string>(
    new BaseCell[] { intCell, dateCell },
    async inputs => $"{inputs[0]} on {inputs[1]}");

// Explicit typed dependencies
engine.AddFunction<int, int>(new[] { a, b }, async inputs => inputs[0] + inputs[1]);

Updating Inputs

await engine.UpdateInput(cell, newValue);

This triggers recomputation of everything downstream.

Getting Results

var value = await engine.GetResult(cell);

Streaming

Subscribe to a cell's value over time:

await foreach (var value in engine.StreamResults(cell, cancellationToken))
{
    Console.WriteLine(value);
}

Yields the current value first, then emits whenever the cell recomputes.

Inspecting the Graph

int count = engine.NodeCount;

foreach (var node in engine.GetAllNodes())
{
    var cell = node.GetCell();
    var deps = node.GetDependencies();
    var value = await node.EvaluateAsync();
}

bool changed = engine.HasChanged(myCell);

Removing Nodes

engine.RemoveNode(cell);

Combining Mixed-Type Cells

Cell<object[]> combined = builder.CombineCells(intCell, stringCell, boolCell);

Execution Model

• All compute functions are async (Func<T[], Task<TResult>>).
• UpdateInput walks dependents in topological order and recomputes them.
• Multiple inputs to a function node resolve concurrently (Task.WhenAll).
• If a node recomputes to the same value, propagation stops there (uses EqualityComparer<T>.Default).
• AddFunction checks for cycles immediately and rolls back if one would form.
• Thread-safe: atomic index generation, per-node compute locks, global semaphore for propagation.

Use Cases

• Backend orchestration: API request pipelines, service dependencies, event-driven workflows
• Financial calculations: risk analysis, transaction chains, dynamic pricing
• Simulations with inputs that change over time

How it Works

1. Create input cells with initial values.
2. Add function cells that depend on those inputs (or on other function cells).
3. Call UpdateInput when something changes.
4. The engine recomputes everything downstream automatically.

Examples

Please go to the Git repo for examples.