User guide

This page describes the execution model in detail. It assumes you have read the quick start example on the home page.

Anatomy of a node

A node has two halves: a declaration that never changes and a state that the engine mutates.

The spec is the single source of truth. setdata! refuses labels that are not output ports, getinputdata refuses labels that are not input ports, and connection validation refuses ports that do not exist at all.

Ports are declared with PortSpec:

PortSpec("Input table", DataFrame, :table)                       # required SinglePort
PortSpec("Extra", Int, :extra, SinglePort(); required = false)   # optional
PortSpec("Values", Int, :values, MultiPort())                    # many-to-one

The datatype is used for connection type checking, label is the symbol you use everywhere in code, and required controls whether the engine refuses to run a node whose buffer is still empty.

Introspection helpers: haveinputs, getportnumber, getporttype, getportspec, isportexist, ismultiport, nodetypestr.

Building a graph

w  = Workflow(1)                      # Workflow{DAW}
id = add_node!(w, DataNode(MyNode, spec))
cid = add_connection!(w, parent_id, :out, child_id, :in)

add_connection! validates through check_connection_validity and raises unless:

• both nodes exist,
• both ports exist in their specs,
• the target input port is free, or declared as a MultiPort,
• the output datatype is a subtype of the input datatype.

A rejected connection consumes nothing. Identifiers come from monotonically increasing counters, so ids of deleted nodes and connections are never reused - after delete_node!(w, 2) the next node is 3.

Two ordering rules are worth remembering:

• add_node! resets a node not in :idle or :clean state.

Configure with setsettings! after adding.

• if the parent is already :clean when the connection is created, its output is copied into the child buffer immediately - no recomputation needed.

Topology queries: getnode, getconnection, isnodeexist, find_connections, getportconnections, get_parents, get_children.

Writing node behaviour

Behaviour is attached to a singleton AbstractNodeType subtype through execute_unsafe!:

struct Doubler <: AbstractNodeType end

function MetidaFlows.execute_unsafe!(node::DataNode{Doubler})
    x = getinputdata(node, :in)
    setdata!(node, :out, 2x)
    return [:out]          # ports produced by this call
end

The returned vector is the contract with the engine: only connections leaving these ports are refreshed by push_buffer!, and the vector is stored as the ready_ports state. A node may legitimately return a subset of its output ports.

Four optional hooks specialise validation; all default to true:

Because validate_result runs before push_buffer!, a node that fails result validation never publishes anything downstream.

Statuses and the execution lifecycle

execute! walks these stages:

1. reset the per-run node log if the run changed;
2. detect re-entry (check_cyclic) and warn instead of recursing forever;
3. return immediately if the node is already :clean;
4. mark :executing;
5. execute parents recursively (execute_upstream);
6. validate structure, required inputs (check_input_buffer) and settings;
7. call execute_unsafe!;
8. validate the result;
9. store ready_ports and push outputs into the child buffers;
10. invalidate children (invalidate_downstream);
11. mark :clean.

The four flags in parentheses come from ExecuteSettings:

execute!(w, id)                                                   # all flags on
execute!(w, id; settings = ExecuteSettings(; execute_upstream = false))
execute!(w, id; settings = ExecuteSettings(false))                # all flags off

execute! returns the vector of ports produced, or an empty vector when the node was rejected or failed - so an empty result is a signal to inspect getstatus.

Invalidation and caching

Caching is what makes incremental recomputation work: a :clean node returns its stored ready_ports without calling execute_unsafe! again. Everything that could make a cached result wrong therefore has to mark nodes :dirty.

invalidate_downstream! is the propagation primitive. Starting at a node it marks it :dirty, drops its cached outputs, deletes the child buffer entry belonging to each outgoing connection and recurses. Traversal stops at nodes that are already :dirty, since their subtree was invalidated earlier.

It is triggered by setsettings!, add_connection!, delete_connection! and by execute! itself.

Three reset levels are available:

Low-level buffer access - setinputbuffer!, invalidate_buffer!, push_buffer! - is available when you drive execution manually, but note that writing a buffer by hand bypasses invalidation.

Schedulers

scheduler!(w)                    # DAW
scheduler!(w; maxiter = 5000)    # ABW

Both return true when the traversal finishes. That is not a success flag: individual nodes may still end up :failed or :invalid_*, so check getstatus when correctness matters.

In ABW, a node popped while a parent is still :dirty is dropped and re-queued later by that parent; isready implements the check. A node that has input ports but no incoming connection never becomes ready and stays :dirty.

Many-to-one inputs

A MultiPort input accepts any number of connections. Values are keyed by connection id, so parents never overwrite each other:

spec = NodeSpec("Collect",
    [PortSpec("values", Int, :ins, MultiPort())],
    [PortSpec("sum", Int, :out)])

function MetidaFlows.execute_unsafe!(node::DataNode{Collect})
    buffer = getinputdata(node, :ins)          # Dict{connection id, value}
    setdata!(node, :out, sum(values(buffer); init = 0))
    return [:out]
end

For a SinglePort, getinputdata returns the single value, or nothing if the buffer is empty. A specific connection can always be read with getinputdata(node, label, connection_id).

Serialization and schemas

workflow_to_dict produces a JSON-ready snapshot of the structure: nodes, connections and the incoming/outgoing indices. Note that "nodes" and "connections" are keyed by stringified ids while "incoming" and "outgoing" keep integer ids, and that execution state (run_id, logs, cached data) is not included.

Building blocks: node_to_dict, node_properties_to_dict, spec_to_dict, portspec_to_dict, connection_to_dict.

Schemas describe a node to a UI. Extend them per node type:

function MetidaFlows.settings_schema_usermod!(d, node::DataNode{MyNode})
    d["schema"] = Dict("threshold" => Dict(
        "type" => Float64, "default" => 0.5, "required" => true,
        "validator" => x -> 0 <= x <= 1))
    return d
end

function MetidaFlows.node_schema_usermod!(d, node::DataNode{MyNode})
    d["section"] = "Statistics"
    d["color"]   = "#8b5cf6"
    return d
end

settings_schema and node_schema call these hooks. Keep in mind that the "settings" key of node_to_dict holds the settings schema, not the current values.

A larger example

Loading a CSV file, converting it and summarising it:

using MetidaFlows, CSV, DataFrames
using MetidaFlows: Workflow, NodeSpec, PortSpec, DataNode, AbstractNodeType,
                   add_node!, add_connection!, setsettings!, scheduler!,
                   getdata, setdata!, getinputdata

struct LoadCSV     <: AbstractNodeType end
struct ToDataFrame <: AbstractNodeType end
struct Summarise   <: AbstractNodeType end

csv_spec = NodeSpec("Load CSV", PortSpec[],
    [PortSpec("CSV File", CSV.File, :csv)], [:file])

df_spec = NodeSpec("DataFrame",
    [PortSpec("CSV File", CSV.File, :csv)],
    [PortSpec("DataFrame", DataFrame, :dataframe)])

sum_spec = NodeSpec("Summary",
    [PortSpec("DataFrame", DataFrame, :dataframe)],
    [PortSpec("Row count", Int, :nrows)])

function MetidaFlows.execute_unsafe!(node::DataNode{LoadCSV})
    setdata!(node, :csv, CSV.File(node.settings[:file]))
    return [:csv]
end

function MetidaFlows.execute_unsafe!(node::DataNode{ToDataFrame})
    setdata!(node, :dataframe, DataFrame(getinputdata(node, :csv)))
    return [:dataframe]
end

function MetidaFlows.execute_unsafe!(node::DataNode{Summarise})
    setdata!(node, :nrows, size(getinputdata(node, :dataframe), 1))
    return [:nrows]
end

w   = Workflow(0)
id1 = add_node!(w, DataNode(LoadCSV, csv_spec))
id2 = add_node!(w, DataNode(ToDataFrame, df_spec))
id3 = add_node!(w, DataNode(Summarise, sum_spec))

add_connection!(w, id1, :csv, id2, :csv)
add_connection!(w, id2, :dataframe, id3, :dataframe)

setsettings!(w, id1, Dict(:file => "data.csv"))

scheduler!(w)
getdata(w, id3, :nrows)

Changing the file path invalidates the whole chain; executing id3 again recomputes all three nodes. Changing nothing and executing again recomputes nothing.

Things to keep in mind

• Settings are merged, not replaced: setsettings! only overwrites the keys you pass.
• Manual execute! with execute_upstream = false will not refresh stale parents; the node is simply rejected with :invalid_node if a required input is missing.