GitHub

Internals

The helpers on this page are internal: the engine's own machinery, documented for contributors. They are not part of the public API, and they may change without notice.

Node inputs and dependencies

At activation, a node finds what each outbound message is computed from. These inputs are the dependencies its algorithm declares, or else the engine's default scheme. ReactiveMP.default_dependencies computes the default scheme for one interface. Here it runs for the interface μ (position 2) of the Gaussian node of the example node, under three factorisations:

x, y = randomvar(label = :x), datavar(label = :y)
interfaces = [(:out, y), (:μ, x), (:v, constvar(1.0))]

map([((:out, :μ, :v),), ((:out,), (:μ,), (:v,)), ((:out, :μ), (:v,))]) do factorisation
    messages, marginals = ReactiveMP.default_dependencies(factornode(Gaussian, interfaces, factorisation), 2)
    (messages = first(messages), marginals = first(marginals))
end
3-element Vector{NamedTuple{(:messages, :marginals)}}:
 (messages = (:out, :v), marginals = ())
 (messages = (), marginals = (:out, :v))
 (messages = (:out,), marginals = (:v,))

With one cluster, the rule towards μ reads the messages on out and v. With one cluster per interface, it reads their marginals. With the clusters (out, μ) and (v), it reads the message on out and the marginal of v.

A rule runs once every input has a new value since its last run, so each of its inputs refreshes once per run, which in variational message passing is one coordinate update. Its marginal inputs make two exceptions, ReactiveMP.MarginalRelaxation, after a run that consumed initial marginals alone: a single refreshed input runs it once more, and a single value computed from data runs it every time one arrives. That starts factorisations whose clusters depend on each other, which would otherwise wait on each other forever, without letting rules re-fire each other on initial values before any data arrives.

Each input carries a label for the rule: its interface name, a cluster's key, or a group member. ReactiveMP.input_names folds a group's members into one tuple, and the labels become the names a ReactiveMP.MessageMapping carries.

ReactiveMP.default_dependencies — Function
ReactiveMP.default_dependencies(factornode, iindex::Int) -> ((labels, interfaces), (labels, sources))

The engine's default scheme, a regular variational message passing scheme driven by the factorisation: the message out of interface iindex is computed from the inbound messages on the other interfaces of its cluster, and the marginals of every other cluster. For a deterministic node it is the inbound messages on every other interface, and no marginal.

Returns the labelled message dependencies and marginal dependencies, each as (labels, sources): labels from ReactiveMP.input_label, or a cluster's key, and the interfaces or local marginals they come from.

source
ReactiveMP.declared_dependencies — Function
ReactiveMP.declared_dependencies(factornode, spec::DependenciesSpec, interface) -> ((labels, interfaces), (labels, sources))

What the message out of interface is computed from under a declared DependenciesSpec, in the order the declaration lists them: m[:x] is an interface's inbound message, q[:x] its variable's marginal, and q[:a, :b] the local marginal of that cluster. A group dependency selects members relative to the target's own index. A target declared with default is ReactiveMP.extended_default_dependencies.

Throws

  • ArgumentError when the declaration has no dependencies for the target, asks for the message of a cluster, or names a cluster the factorisation does not have.
source
ReactiveMP.extended_default_dependencies — Function
ReactiveMP.extended_default_dependencies(factornode, interface, inputs)

The dependencies of a target declared with default, :a => (default, q[:a]): the default scheme's inputs, and each of inputs, a single interface's message or marginal, placed in interface order. A marginal added this way is the variable's own, and an input the default scheme has already is not added twice.

source
ReactiveMP.input_names — Function
ReactiveMP.input_names(labels) -> Val

The names of a rule's inputs, as the Val a mapping carries. Each label is an interface name, a cluster's member tuple, a ReactiveMP.GroupMember or a ReactiveMP.EmptyGroup; the consecutive members of a group become one ReactiveMP.GroupInputs.

Throws

  • ArgumentError when a group's members are not in member order, or when a name appears more than once, such as a group whose members are not consecutive.
source
ReactiveMP.GroupInputs — Type
ReactiveMP.GroupInputs{G, N, K}()

Group G among the names of a rule's inputs: its members K, a tuple of indices in order, out of the N the group has, which are consecutive among the inputs. The rule receives the group as one tuple of length N in member order, with nothing where a member is not an input.

source
ReactiveMP.EmptyGroup — Type
ReactiveMP.EmptyGroup(group::Symbol, length::Int)

The label of a group from which a dependency selects no member, such as m[:in][!k] for a group of one member: it takes no input, and the rule receives a tuple of length nothings.

source
ReactiveMP.FactorNodeLocalClusters — Type
ReactiveMP.FactorNodeLocalClusters(interfaces, factorization::Tuple; lone_group_joint = true)

The clusters of a factor node: the factorisation, as tuples of positions into interfaces, and one ReactiveMP.FactorNodeLocalMarginal per cluster, keyed with ReactiveMP.clusterkey. lone_group_joint says whether a cluster that is a whole group of one member, and nothing else, is the joint over the group, as a deterministic node's inputs are; factornode passes false for a stochastic node, where such a cluster is the member's own marginal.

source
ReactiveMP.clusterkey — Function
ReactiveMP.clusterkey(cluster::Tuple{Vararg{Int}}, interfaces; lone_group_joint = true)

The key a cluster of interface positions is read by:

  • the name of its only interface, :μ, for a cluster of one interface, which shares its variable's marginal;
  • otherwise the tuple of its members' names, a joint, in which a group whose members are all in the cluster appears once, by its name: (:in,) is the joint over the group in. A joint holding only some of a group's members names each as (group, index), (:out, (:T, 1)).

A cluster that is a whole group of one member, and nothing else, is the joint over the group, (:in,), as a deterministic node's inputs are, with lone_group_joint = true; with false, as in a stochastic node, it is a cluster of one interface, :in.

source
ReactiveMP.MarginalRelaxation — Type
ReactiveMP.MarginalRelaxation()

When a rule's marginal inputs fire it: the callback of their combination, and the state its ReactiveMP.RelaxedInputs share. The combination fires once every input has a new value since the last firing, with two exceptions, both after a firing that consumed initial marginals alone, those ReactiveMP.set_initial_marginal! set or computed from them:

  • the next update of any single input fires it again, once for each subscription;
  • an update that is not initial, a value computed from data, fires it again, every time.

Strict refreshing would otherwise deadlock structured or mean-field factorisations whose clusters wait on each other: each waits for the others to refresh, and an initial value consumed by a firing is never refreshed on its own (RxInfer#344). Real values reach a rule far from the data only after several rounds, so a rule may need restarting more than once, whenever one arrives. Initial values alone restart it at most once: restarting on every one let the rules re-fire each other on initial values before any data arrived, tens of thousands of rule calls for a mixture of two components on five points, where these two exceptions make a few dozen.

It applies to the marginal inputs only: a rule's message inputs keep strict refreshing. It keeps the subscription it belongs to, so a combination subscribed again relaxes once more.

source

Rule arguments

A mapping hands a rule its inputs through ReactiveMP.rule_arguments. It builds them from the latest messages and marginals with the helpers below, and it collects their annotations the same way.

ReactiveMP.rule_messages — Function
ReactiveMP.rule_messages(f, names, messages) -> Messages

The Messages a rule reads, f of each message keyed by its interface name: getdata for the rule's arguments, getannotations for its annotations. A group is one tuple under its name (see ReactiveMP.GroupInputs). With names and messages both nothing, it is empty.

source
ReactiveMP.rule_marginals — Function
ReactiveMP.rule_marginals(f, names, marginals) -> Marginals

The Marginals a rule reads, f of each marginal. A marginal keyed by a symbol is the marginal of one interface, one keyed by a tuple of names the joint of a cluster, and a group is one tuple under its name (see ReactiveMP.GroupInputs).

A joint whose value is a FactorizedCluster reaches the rule as its blocks instead: a block of one interface as that interface's marginal, q[:out], and a larger one as a joint, q[:out, :μ]. A block of one member of a group stays a joint of that member, q[((:T, 1),)], since the group's length, which q[:T] needs, is not known from the cluster. The blocks are in the cluster's type, so this is decided at compile time.

Throws

  • ArgumentError when a FactorizedCluster's blocks do not partition the cluster's members, each block in the cluster's order.
source

Scratch

A rule may declare a scratch, working memory that it writes before it reads. Each message and marginal mapping keeps the scratch of the rule it runs. The mapping builds the scratch at the rule's first call and reuses it while the same rule runs on the stream. The scratch never leaves the rule and is never shared with another.

ReactiveMP.scratch_for! — Function
ReactiveMP.scratch_for!(slot::ScratchSlot, spec, algorithm, ctx, args, target, checked::Bool = false)

The scratch to run the rule spec with: the one slot holds, if it was built for this rule, or a new one from rule_scratch, which slot then keeps. nothing for a rule that declares no scratch.

With checked, the checked_buffers diagnostic, a reused scratch is poisoned first (see ReactiveMP.poison!), so a rule that reads its scratch before writing it computes NaN.

source
ReactiveMP.poison! — Function
ReactiveMP.poison!(memory)

Fill every floating-point array in memory with NaN, looking inside arrays, tuples and named tuples, and return memory. Anything else is left as it is. The checked_buffers diagnostic applies it to a rule's scratch before each reuse.

source

Static inputs

A node with static inputs reads some of its interfaces as values, not as messages: a constant, or the last observation of a data variable. ReactiveMP.StaticFold passes those values to the node's function, and ReactiveMP.with_statics holds a node's streams back until every static input has a value (see Static inputs).

ReactiveMP.StaticFold — Type
ReactiveMP.StaticFold(f, statics::Tuple)

The function of a node that folds its static inputs (static_inputs = :fold), which factornode builds from its nodefn. Called with the free inputs, it calls f with every input in its place, each static one at its position k taking the latest value of its variable: the constant, or the last observation. statics holds (k, variable) pairs.

source
ReactiveMP.with_statics — Function
ReactiveMP.with_statics(factornode, stream)

stream, gated on the node's static inputs: it emits only once each of them has a value, and again when one of them updates. Without static inputs it is stream itself.

source

The equality chain

A random variable with more than one connection computes its outbound messages along an equality chain. The chain has one equality node per connection, and the nodes cache the partial products of the variable's inbound messages.

ReactiveMP.EqualityChain — Type
ReactiveMP.EqualityChain(inputmsgs::Vector{MessageObservable{AbstractMessage}}, postprocessor, variable::AbstractVariable, context::MessageProductContext)

How a random variable with more than one connection computes its outbound messages. The message to connection i is the product of the inbound messages on every other connection, which the chain computes from two partial products, of the messages before i and of those after it. The partial products are cached in the ReactiveMP.EqualityNodes and shared by the outbound messages, which matters for a variable of high degree; a new inbound message on connection i invalidates only the partial products that contain it.

A partial product is a part of a message, not a message: it is multiplied by context's fold strategy alone, pairwise with ReactiveMP.compute_product_of_two_messages, which applies the form constraint under FormConstraintCheckEach. The product of the two parts is the message, and is computed with ReactiveMP.compute_product_of_messages: under FormConstraintCheckLast the form constraint applies to it once, and the callbacks see one whole product per outbound message.

Arguments

  • inputmsgs: the variable's inbound message streams, one per connection;
  • postprocessor: the stream postprocessor applied to the chain's streams, or nothing;
  • variable: the variable the products are for, which the callbacks receive;
  • context: how the messages are multiplied; a random variable gives its prod_context_for_message_computation.
source
ReactiveMP.EqualityNode — Type
ReactiveMP.EqualityNode()

One link of an ReactiveMP.EqualityChain, for one connection of a random variable: the partial products of the inbound messages on either side of it, kept so that they are reused.

 <- left    -------     right ->
     ----- |   =   | ------
            -------
               |
               | outbound

Fields

  • left: the stream that says the product of this and every later inbound message can be computed; it emits missing, a notification only;
  • right: the stream that says the product of this and every earlier inbound message can be computed; it emits missing too;
  • cache_left: the cached product towards the left; the chain tracks whether it is valid;
  • cache_right: the cached product towards the right; the chain tracks whether it is valid.
source

Callbacks

ReactiveMP.merge_callbacks returns a ReactiveMP.MergedCallbacks, and the engine reports its events through ReactiveMP.@invoke_callback, which builds an event only for a handler that listens to it (see Callbacks).

ReactiveMP.@invoke_callback — Macro
ReactiveMP.@invoke_callback(callbacks, EventType(args...))

invoke_callback(callbacks, EventType(args...)), with the event built only when ReactiveMP.listens(callbacks, EventType): an event nobody listens to costs nothing.

source

Helpers

@proxy_methods defines the statistics, such as mean and var, that Message and Marginal forward to their data.

ReactiveMP.MacroHelpers.@proxy_methods — Macro
ReactiveMP.MacroHelpers.@proxy_methods(proxy_type, proxy_getter, proxy_methods)

Define each function of proxy_methods, a vector literal, on proxy_type as that function of proxy_getter of the value: how Message and Marginal forward mean, var and the other statistics to their data.

Examples

@proxy_methods Message getdata [Distributions.mean, Distributions.var]

defines

Distributions.mean(proxy::Message) = Distributions.mean(getdata(proxy))
Distributions.var(proxy::Message) = Distributions.var(getdata(proxy))

Throws

  • ErrorException when proxy_methods is not a vector literal.
source