-
-
Notifications
You must be signed in to change notification settings - Fork 1
602. Patterns and recipes
The previous pages explain the mechanics of F-Mesh — this one collects patterns: proven ways to structure meshes. Runnable programs demonstrating most of them live in the fmesh-examples repository.
For a chain of uniform stages, don't hand-wire every pipe — write a builder that accepts components, gives each one standard in/out ports, tags it with a stage label, and pipes stage N to stage N+1:
func buildPipeline(name string, components ...*component.Component) (*fmesh.FMesh, error) {
for i, c := range components {
c.AddLabel("stage", strconv.Itoa(i+1))
if err := c.AddInputs("in"); err != nil { return nil, err }
if err := c.AddOutputs("out"); err != nil { return nil, err }
if i > 0 {
prev := components[i-1]
if err := prev.OutputByName("out").PipeTo(c.InputByName("in")); err != nil {
return nil, err
}
}
}
// ... fmesh.New + AddComponents ...
}The stage label then lets you find the entry and exit points by role instead of by name:
first := fm.Components().FindAny(func(c *component.Component) bool {
return c.Labels().ValueIs("stage", "1")
})
_ = first.InputByName("in").PutSignals(signal.New("start"))A star topology where every node talks to every other node through a central bus component. Each node is wired bidirectionally (node.tx → bus.rx, bus.tx → node.rx), and the bus itself is a one-liner:
component.WithActivationFunc(func(this *component.Component) error {
return port.ForwardSignals(this.InputByName("rx"), this.OutputByName("tx"))
})Every node receives everything and filters for itself — e.g. by an address kept in its state or by signal labels. This mirrors real shared-medium protocols, and adding a node never changes existing wiring.
When many producers pipe into one input port, that port accumulates one signal per producer per cycle. The consumer reads the whole group and reduces it:
var readings []float64
_ = this.InputByName("in").Signals().ForEach(func(sig *signal.Signal) error {
readings = append(readings, sig.PayloadOrDefault(0.0).(float64))
return nil
})
combined := slices.Min(readings) // or sum, average, majority vote…Because all producers of the same cycle land together, this is a natural way to model shared-medium effects (arbitration, interference, voting). If your component instead expects exactly one signal, assert it — Signals().Len() > 1 on such a port usually means a wiring bug.
Give the router one output port per destination and decide per signal. Labels are the natural routing key:
component.WithOutputs("passed", "dropped"),
component.WithActivationFunc(func(this *component.Component) error {
return this.InputByName("in").Signals().ForEach(func(sig *signal.Signal) error {
if sig.Labels().ValueIs("genre", "pop") {
return this.OutputByName("dropped").PutSignals(sig)
}
return this.OutputByName("passed").PutSignals(sig)
})
})ForEachIf(predicate, action) and port.ForwardWithFilter(src, dst, predicate) express the same idea more compactly — see Conditional forwarding.
Indexed ports give you a numbered port family for a variable number of workers; component state holds the round-robin cursor:
lb, _ := component.New("lb",
component.WithInputs("in"), component.WithOutputs("out"),
component.WithIndexedOutputs("downstream", 0, numWorkers-1), // to workers
component.WithIndexedInputs("upstream", 0, numWorkers-1), // from workers
component.WithInitialState(func(s component.State) { s.Set("next", 0) }),
component.WithActivationFunc(func(this *component.Component) error {
next := this.State().GetOrDefault("next", 0).(int)
_ = this.InputByName("in").Signals().ForEach(func(sig *signal.Signal) error {
portName := fmt.Sprintf("downstream%d", next%numWorkers)
next++
return this.OutputByName(portName).PutSignals(sig)
})
this.State().Set("next", next)
// collect worker responses back into the single egress port
for i := 0; i < numWorkers; i++ {
_ = port.ForwardSignals(this.InputByName(fmt.Sprintf("upstream%d", i)),
this.OutputByName("out"))
}
return nil
}))Because the cursor lives in state, distribution stays fair even across multiple Run() calls on the same mesh.
A single component whose outputs feed back into its own inputs becomes an iterative process — the mesh's cycles drive the iteration, and the loop state travels through the ports:
_ = gen.LoopbackPipe("o_prev", "i_prev")
_ = gen.LoopbackPipe("o_cur", "i_cur")
// bootstrap the loop before Run()
_ = gen.InputByName("i_prev").PutSignals(signal.New(0))
_ = gen.InputByName("i_cur").PutSignals(signal.New(1))A cyclic mesh keeps running only while signals keep arriving. There is no explicit "stop" call — a component ends the loop by not emitting into it:
if next >= limit {
return nil // no output ⇒ no next activation ⇒ the mesh stops naturally
}
return this.OutputByName("o_cur").PutSignals(signal.New(next))This is the mental model for every cyclic topology. If a loop component unconditionally emits, the loop runs until CyclesLimit — which also means the default limits (see Scheduling rules) act as a useful safety net while you develop.
When a feedback loop needs a one-time trigger to begin, give one component a dedicated bootstrap input, seed it before Run(), and branch on it inside the activation:
if this.InputByName("bootstrap").HasSignals() {
// first activation: emit the initial signal into the loop, skip normal logic
return this.OutputByName("request").PutSignals(signal.New(initialValue))
}
// steady state: react to the loop input, re-emit (or stay silent to stop)A component with a loopback pipe that re-emits a signal each activation stays alive every cycle — useful for watchdogs, timers, and startup sequences:
_ = watchdog.LoopbackPipe("keepalive", "keepalive")
// inside the activation func:
if !meshIsIdle {
_ = this.OutputByName("keepalive").PutSignals(signal.New(true))
}
// stop emitting ⇒ the whole mesh quiescesA variant carries a countdown payload: seed N, re-emit N-1 while positive, and you get a component that does something for exactly N cycles.
If component A consumes B's output and B consumes A's, neither can be first within a single tick. The fix: the stateful side publishes its current state early (from its own state, before consuming this tick's inputs) and folds the new inputs into state for the next tick — accepting one tick of latency:
// Phase A — a tick arrived: publish the *stored* value immediately,
// so downstream consumers can proceed within this same run.
if this.InputByName("time").HasSignals() {
level := this.State().GetOrDefault("level", 0.0).(float64)
_ = this.OutputByName("level").PutSignals(signal.New(level))
}
// Phase B — inputs from peers arrived: update state for the next tick.
if this.InputByName("updates").HasSignals() {
// ...fold updates into this.State()...
}Run() is a one-shot drain: it cycles until no component activates, then returns. But the mesh object survives — component state persists, input signals persist until consumed, and output ports are cleared at the start of each run. So a long-lived program can loop: inject inputs, Run(), read outputs, repeat.
for job := range jobs {
_ = worker.InputByName("in").PutSignals(signal.New(job))
if _, err := fm.Run(); err != nil { break }
results, _ := worker.OutputByName("out").Signals().AllPayloads()
worker.OutputByName("out").Clear() // don't let stale results accumulate
resultsChan <- results
}If you read outputs between runs (rather than after the final one), Clear() them once consumed. Use fmesh.WithUnlimitedTime() / WithUnlimitedCycles() when a single run may legitimately be slow (e.g. real network I/O inside an activation function).
For discrete-time simulations, make the tick a signal. A mesh-level BeforeRun hook injects it, a clock component timestamps it, and pipes deliver it to every component that needs time:
fm.SetupHooks(func(hooks *fmesh.Hooks) {
hooks.BeforeRun(func(mesh *fmesh.FMesh) error {
return mesh.ComponentByName("clock").InputByName("ctl").PutSignals(signal.New("tick"))
})
})Pack the tick's fields (sequence number, simulated duration, Δt) as scalars on one signal, so components that integrate over time get the Δt they need. One Run() = one tick; call Run() in a loop to advance the clock.
A mesh can be stepped interactively — from a REPL, a UI, or a network endpoint. Two rules make this safe and pleasant:
- Single-goroutine ownership. Only one goroutine ever touches the mesh (inject, run, read). F-Mesh objects are not synchronized for external concurrent access, so other goroutines communicate with the owner via a channel — a channel of closures works well:
commands := make(chan func(fm *fmesh.FMesh))
// the owning goroutine:
for cmd := range commands {
cmd(fm) // e.g. inject a signal, inspect state
_, _ = fm.Run() // advance one step
}
// any other goroutine:
commands <- func(fm *fmesh.FMesh) {
_ = fm.ComponentByName("source").InputByName("in").PutSignals(signal.New("hello"))
}- Detect idle steps so you can pause instead of spinning, using the run report:
runInfo, _ := fm.Run()
if runInfo.Cycles.Count(func(c *cycle.Cycle) bool { return c.HasActivatedComponents() }) == 0 {
// nothing happened — wait for the next command instead of running again
}Give each component its own constructor returning (*component.Component, error), with errors wrapped at every step. Meshes are then assembled from factories, keeping main free of wiring detail. For families of similar components, parameterize the factory (newWorker(i int)) and add the results in bulk: fm.AddComponents(workers...).
Group related components in a plain Go struct that knows how to enumerate and connect them:
type Node struct {
Controller, Transmitter, Receiver *component.Component
}
func (n *Node) AllComponents() []*component.Component { /* ... */ }
func (n *Node) ConnectTo(bus *Bus) error { /* internal + bus wiring */ }main stays declarative: build nodes, fm.AddComponents(node.AllComponents()...), node.ConnectTo(bus).
A whole mesh can hide behind one component whose activation function runs it — see Nested meshes for the mechanics. At scale, structure the wrapper's activation as sequential phases:
-
validate — return
component.ErrWaitingForInputsKeepuntil all required inputs are present; -
sense — forward the wrapper's inputs into the inner mesh's input ports (
port.ForwardSignals); -
act —
innerMesh.Run(); - feedback — forward inner output ports back to the wrapper's outputs.
Build the inner mesh once (in the wrapper's constructor, closed over by the activation function), not on every activation — component state inside the inner mesh then persists across ticks.
With many components, explicit pipe lists get tedious. Adopt a port-naming convention and wire by lookup — e.g. at assembly time, pipe the clock's output to every component that has an input named time, or match inputs named <subsystem>_<port> to the corresponding subsystem output. Keep port names as shared constants in one package — pipes should never depend on magic strings.
When many components share behavior that differs only in data, generate the activation function from a declarative table instead of writing near-identical closures:
func activationFromTable(handlers map[string]func(*component.Component, *signal.Signal) error) component.ActivationFunc {
return func(this *component.Component) error {
return this.InputByName("in").Signals().ForEach(func(sig *signal.Signal) error {
if handler, ok := handlers[sig.Labels().ValueOrDefault("cmd", "")]; ok {
return handler(this, sig)
}
return nil // unknown commands are ignored
})
}
}Each component instance is then just a table; the uniform dispatch lives in one place.
These build on Metadata:
-
Labels as routing keys, not just annotations — filters route on
sig.Labels().ValueIs(k, v); a relay forwards only signals labeled for it. -
Find components by role, not name —
fm.Components().FindAny(...)over labels likerole=aggregatororstage=1decouples orchestration code from component names. -
Pack multi-field values as scalars on one signal when the fields are numeric and you want them queryable (
signal.New("tick").WithScalar("seq", n).WithScalar("delta_t_ms", dt)); use a plain struct payload when the value is opaque to the mesh. Pair eachpackXhelper with anunpackXand unit-test the round trip. -
Tag provenance when aggregating —
port.ForwardWithMap(src, dst, func(s *signal.Signal) *signal.Signal { return s.WithLabel("from", name) })lets one collector port carry signals from many sources without losing origin. -
Bulk-label a group on injection —
signal.NewGroup(payloads...).Map(func(s *signal.Signal) *signal.Signal { return s.WithLabel("to", "usb") })thenPutSignalGroups(...).
Rather than having observers reach into many components, add one aggregator component that pipes in every observable port and re-emits everything on a single output — a queryable single source of truth. Its ports can even be created dynamically from a list of "component::port" paths at mesh-build time. Tests and UIs then read one component.
External consumers (TUIs, dashboards) should never touch the mesh. Publish from an AfterRun hook to a sink, and let the consumer read the stream:
fm.SetupHooks(func(hooks *fmesh.Hooks) {
hooks.AfterRun(func(mesh *fmesh.FMesh) error {
return mesh.ComponentByName("publisher").OutputByName("stream").Signals().
ForEach(func(line *signal.Signal) error {
return sink.Publish(line.PayloadOrDefault("").(string))
})
})
})Put the sink behind a small interface (Publish(line string) error) so stdout, a socket, or a no-op are interchangeable, and make it non-blocking and lossy (buffered channel, drop-and-count on overflow) so a slow observer can never back-pressure the mesh.
Give fallible components a dedicated errors output port piped to a logger/collector component, and continue past per-item failures instead of returning an error (which, under the default strategy, stops the whole mesh). Reserve the returned error for genuinely fatal conditions.
MaybeActivate covers single components; for whole-mesh behavior, use hooks as probes:
-
Collect a time series with a component-level
AfterActivationhook (or a mesh-levelAfterRunhook reading the aggregator), appending each observed value to a slice while the mesh runs. - Assert properties, not exact values: monotonically increasing clock, oscillation (sign changes, min–max range), values staying within bounds and not pinned to a clamp boundary — robust against timing noise in a way exact-value asserts are not.
-
Bound simulated time, not wall time: a hook on the clock component watches simulated duration and triggers shutdown when the target is reached (guard the trigger with
sync.Once— firing a shutdown command twice into a closed channel is a classic leak). - Unit-test your signal-packing helpers (pack/unpack round trips, invariants like "shares sum to 100%") separately from the mesh.
-
A multi-input component activates when any input has signals. If an activation phase must run exactly once per tick, gate it on the specific port:
if !this.InputByName("time").HasSignals() { return nil }. -
return nilvs waiting. Returningnilon partial input silently consumes what arrived;component.ErrWaitingForInputsKeepre-queues the component keeping its buffered inputs until the rest arrive. Choose deliberately — see Waiting for inputs. - Don't depend on activation order within a cycle. Components activate concurrently; if a value must be seen "before" another, carry it in state across ticks (see the feedback-loop deadlock pattern above) instead of relying on ordering.
-
Type-assert payloads with the
, okform at trust boundaries and treat failures as data errors (route to an error port), not panics — payloads areany. -
Signals fanned out through pipes are shared pointers. Treat payloads as immutable; transform with
MapPayload/With*copies (see Signals). -
Outputs are cleared at the start of each
Run(); inputs persist. Read (and if neededClear()) outputs between runs; seed state through input ports or component state, never by pre-loading outputs.