Callbacks

Callbacks observe message passing. The engine reports an event before and after every rule call, every product of messages, every form constraint, every marginal of a random variable, every joint marginal a node computes, and every term of the free energy. A callback handler reacts to the events it is interested in. Callbacks serve debugging, tracing and monitoring, and they change nothing the engine computes. To transform the streams themselves, use stream postprocessors.

Attaching callbacks

You give callbacks in two places:

to a factor node, as the activation option callbacks of ReactiveMP.FactorNodeActivationOptions. The node reports its message rule calls, ReactiveMP.BeforeMessageRuleCallEvent and ReactiveMP.AfterMessageRuleCallEvent, its marginal rule calls, which compute the joint marginals of its clusters, and its term of the free energy, ReactiveMP.AfterFactorBoundFreeEnergyEvent.
to a variable, as the callbacks of the ReactiveMP.MessageProductContexts in its RandomVariableActivationOptions. The variable reports the product and form constraint events, and the context of its marginal also reports the marginal events and the variable's term of the free energy, ReactiveMP.AfterVariableBoundEntropyEvent.

An event after a rule call names the rule that ran, its rule: the RuleSpec, which shows where the rule is defined, the node's rule fallback where no rule matched, or nothing where a missing input skipped the rule. The engine builds an event only for a handler that listens to it, so a graph activated without callbacks, or with callbacks for other events, runs as fast as one that reports nothing.

The examples use the Gaussian node of The example node. The simplest handler is a NamedTuple keyed by event names, each entry a function of the event. Mind the trailing comma of a one-entry NamedTuple, (name = f,).

x, y = randomvar(label = :x), datavar(label = :y)
prior = factornode(Gaussian, [(:out, x), (:μ, constvar(0.0)), (:v, constvar(10.0))])
likelihood = factornode(Gaussian, [(:out, y), (:μ, x), (:v, constvar(1.0))])

callbacks = (
    after_message_rule_call = (event) -> println("rule towards ", event.mapping.target, ": ", event.result),
    after_marginal_computation = (event) -> println("marginal: ", getdata(event.result)),
)

marginal_context = MessageProductContext(; callbacks)
activate!(x, RandomVariableActivationOptions(nothing, MessageProductContext(), marginal_context))
activate!(y, DataVariableActivationOptions())
foreach(node -> activate!(node, FactorNodeActivationOptions(; callbacks)), (prior, likelihood))

subscription = subscribe!(get_stream_of_marginals(x), (q) -> nothing)
new_observation!(y, 2.0)
unsubscribe!(subscription)

rule towards :out: ExponentialFamily.NormalMeanVariance{Float64}(μ=0.0, v=10.0)
rule towards :μ: ExponentialFamily.NormalMeanVariance{Float64}(μ=2.0, v=1.0)
marginal: ExponentialFamily.NormalWeightedMeanPrecision{Float64}(xi=2.0, w=1.1)

When the observation arrives, x reads the messages of both nodes, which runs their rules, and forms its marginal from them.

A handler of a type of its own implements ReactiveMP.handle_event for each event it reacts to. The engine builds an event only for a handler that ReactiveMP.listens to its type, so a handler that reacts to a few events declares which, and the others cost it nothing. The handler below counts the rule calls of each target:

struct RuleCounter
    counts::Dict{Any, Int}
end

ReactiveMP.listens(::RuleCounter, ::Type{<:ReactiveMP.Event}) = false
ReactiveMP.listens(::RuleCounter, ::Type{<:ReactiveMP.AfterMessageRuleCallEvent}) = true

function ReactiveMP.handle_event(counter::RuleCounter, event::ReactiveMP.AfterMessageRuleCallEvent)
    target = event.mapping.target
    counter.counts[target] = get(counter.counts, target, 0) + 1
    return nothing
end

counter = RuleCounter(Dict{Any, Int}())
x, y = randomvar(label = :x), datavar(label = :y)
prior = factornode(Gaussian, [(:out, x), (:μ, constvar(0.0)), (:v, constvar(10.0))])
likelihood = factornode(Gaussian, [(:out, y), (:μ, x), (:v, constvar(1.0))])
activate!(x, RandomVariableActivationOptions())
activate!(y, DataVariableActivationOptions())
foreach(node -> activate!(node, FactorNodeActivationOptions(; callbacks = counter)), (prior, likelihood))

subscription = subscribe!(get_stream_of_marginals(x), (q) -> nothing)
foreach(value -> new_observation!(y, value), (1.0, 2.0, 3.0))
unsubscribe!(subscription)
counter.counts

Dict{Any, Int64} with 2 entries:
  :μ   => 3
  :out => 1

The prior's message towards out depends on constants alone, so it is computed once. The likelihood's message towards μ is computed once per observation. ReactiveMP.merge_callbacks combines several handlers into one.

ReactiveMP.Event — Type

ReactiveMP.Event{E}

The supertype of the events the engine reports to callbacks. E is the event's name, a Symbol such as :before_message_rule_call, and the event for the name :event_name is the struct EventNameEvent. An event carries what happened in its fields.

Symbol	Struct
`:before_message_rule_call`	`ReactiveMP.BeforeMessageRuleCallEvent`
`:after_message_rule_call`	`ReactiveMP.AfterMessageRuleCallEvent`
`:before_product_of_two_messages`	`ReactiveMP.BeforeProductOfTwoMessagesEvent`
`:after_product_of_two_messages`	`ReactiveMP.AfterProductOfTwoMessagesEvent`
`:before_product_of_messages`	`ReactiveMP.BeforeProductOfMessagesEvent`
`:after_product_of_messages`	`ReactiveMP.AfterProductOfMessagesEvent`
`:before_form_constraint_applied`	`ReactiveMP.BeforeFormConstraintAppliedEvent`
`:after_form_constraint_applied`	`ReactiveMP.AfterFormConstraintAppliedEvent`
`:before_marginal_computation`	`ReactiveMP.BeforeMarginalComputationEvent`
`:after_marginal_computation`	`ReactiveMP.AfterMarginalComputationEvent`
`:before_marginal_rule_call`	`ReactiveMP.BeforeMarginalRuleCallEvent`
`:after_marginal_rule_call`	`ReactiveMP.AfterMarginalRuleCallEvent`
`:before_factor_bound_free_energy`	`ReactiveMP.BeforeFactorBoundFreeEnergyEvent`
`:after_factor_bound_free_energy`	`ReactiveMP.AfterFactorBoundFreeEnergyEvent`
`:before_variable_bound_entropy`	`ReactiveMP.BeforeVariableBoundEntropyEvent`
`:after_variable_bound_entropy`	`ReactiveMP.AfterVariableBoundEntropyEvent`

Callbacks

Attaching callbacks

Event names

Event spans

All events