simulator

import "github.com/umbralcalc/stochadex/pkg/simulator"

Package simulator provides the core simulation engine and infrastructure for stochadex simulations. It includes the main simulation loop, state management, partition coordination, and execution control mechanisms.

Key Features:

Architecture: The simulator uses a partition-based approach where simulations are divided into independent partitions that can be executed concurrently. Each partition maintains its own state history and can communicate with other partitions through defined interfaces.

Usage Patterns:

Index

Variables

var File_cmd_messages_partition_state_proto protoreflect.FileDescriptor

func DeriveSeed

func DeriveSeed(base uint64, stream int) uint64

DeriveSeed mixes a base seed with a stream index, so the partitions of one re-entrant run get distinct but jointly-determined seeds. Splitting this way means two runs sharing a base seed reproduce each other exactly, while two partitions within a run do not share a random stream.

func RegisterComponent

func RegisterComponent(family, typeName string, build func(ComponentSpec) (interface{}, error))

RegisterComponent registers a data-spec builder for a component that lives downstream of simulator. family is one of “output_condition”, “output_function”, “termination_condition”, “timestep_function”. Call it from an init(); it panics on a duplicate so two packages cannot silently claim one name.

func ReseedIterations

func ReseedIterations(settings *Settings, implementations *Implementations, base uint64)

ReseedIterations re-Configures every iteration with a seed derived from base, restoring them to a state that depends only on base. After this call the next run reproduces any earlier run made with the same base and starting state.

This relies on the framework rule that Configure re-initialises all mutable state. An iteration that stashes state outside Configure’s reach breaks the guarantee — which is the same defect RunWithHarnesses already fails on.

func RunWithHarnesses

func RunWithHarnesses(settings *Settings, implementations *Implementations) error

RunWithHarnesses runs all iterations, each wrapped in a test harness and returns any errors if found. The simulation is also run twice to check for statefulness residues.

It uses the default spawn-per-step execution. To exercise a specific ExecutionStrategy under the same checks, use RunWithHarnessesUsing.

func RunWithHarnessesUsing

func RunWithHarnessesUsing(settings *Settings, implementations *Implementations, strategy ExecutionStrategy) error

RunWithHarnessesUsing is like RunWithHarnesses but advances the simulation with the given ExecutionStrategy (nil selects the default spawn-per-step execution and uses the manual Step loop so behaviour is unchanged). It applies every per-step correctness check (params mutation, NaN, state width, history integrity) and the twice-run statefulness-residue check, so each strategy is validated against the same rigour as the default.

type ComponentSpec

ComponentSpec is the value for a framework component in a config: a data spec ({type: every_step, …}) resolved at load time by the registry below, needing no Go toolchain. A partition’s bespoke maths goes through expressions: instead; the framework’s own catalogue is named here by type.

type ComponentSpec struct {
    // Type is the data-spec discriminator (its "type" key).
    Type string
    // Fields holds the remaining data-spec keys (everything but "type").
    Fields map[string]interface{}
}

func (ComponentSpec) IsData

func (c ComponentSpec) IsData() bool

IsData reports whether the spec was populated (a {type: …} data spec).

func (ComponentSpec) IsZero

func (c ComponentSpec) IsZero() bool

IsZero reports whether the spec was never populated (the YAML omitted it).

func (*ComponentSpec) UnmarshalYAML

func (c *ComponentSpec) UnmarshalYAML(unmarshal func(interface{}) error) error

UnmarshalYAML accepts a mapping with a non-empty string “type” key.

type ConfigGenerator

ConfigGenerator builds Settings and Implementations programmatically and can generate runnable configs on demand.

type ConfigGenerator struct {
    // contains filtered or unexported fields
}

func NewConfigGenerator

func NewConfigGenerator() *ConfigGenerator

NewConfigGenerator creates a new ConfigGenerator with empty ordering.

func (*ConfigGenerator) GenerateConfigs

func (c *ConfigGenerator) GenerateConfigs() (*Settings, *Implementations)

GenerateConfigs constructs Settings and Implementations ready to run. It computes state widths, converts named references, and configures iterations with their partition indices.

func (*ConfigGenerator) GetGlobalSeed

func (c *ConfigGenerator) GetGlobalSeed() uint64

GetGlobalSeed returns the current global seed.

func (*ConfigGenerator) GetPartition

func (c *ConfigGenerator) GetPartition(name string) *PartitionConfig

GetPartition retrieves a partition config by name.

func (*ConfigGenerator) GetSimulation

func (c *ConfigGenerator) GetSimulation() *SimulationConfig

GetSimulation returns the current simulation config.

func (*ConfigGenerator) PartitionNames

func (c *ConfigGenerator) PartitionNames() []string

PartitionNames returns the partition names in the order they were added, which is the same order used to assign partition indices in GenerateConfigs. It returns a copy so callers cannot mutate the internal ordering.

func (*ConfigGenerator) ResetPartition

func (c *ConfigGenerator) ResetPartition(name string, config *PartitionConfig)

ResetPartition replaces the config for a partition by name.

func (*ConfigGenerator) SetGlobalSeed

func (c *ConfigGenerator) SetGlobalSeed(seed uint64)

SetGlobalSeed assigns a random seed to each partition derived from the provided global seed.

func (*ConfigGenerator) SetPartition

func (c *ConfigGenerator) SetPartition(config *PartitionConfig)

SetPartition adds a new partition config. Names must be unique.

func (*ConfigGenerator) SetSimulation

func (c *ConfigGenerator) SetSimulation(config *SimulationConfig)

SetSimulation sets the current simulation config.

type ConstantTimestepFunction

ConstantTimestepFunction uses a fixed stepsize.

type ConstantTimestepFunction struct {
    Stepsize float64
}

func (*ConstantTimestepFunction) NextIncrement

func (t *ConstantTimestepFunction) NextIncrement(timestepsHistory *CumulativeTimestepsHistory) float64

type CumulativeTimestepsHistory

CumulativeTimestepsHistory is a rolling window of cumulative timesteps with NextIncrement and CurrentStepNumber.

type CumulativeTimestepsHistory struct {
    NextIncrement     float64
    Values            *mat.VecDense
    CurrentStepNumber int
    StateHistoryDepth int
}

type DownstreamStateValues

DownstreamStateValues contains information to broadcast state values to downstream iterators via channel.

type DownstreamStateValues struct {
    Channel chan []float64
    Copies  int
}

type EnsembleRun

EnsembleRun pairs the seed used for a single ensemble member with the data recorded from running it.

type EnsembleRun struct {
    Seed    uint64
    Storage *StateTimeStorage
}

func RunSeededEnsemble

func RunSeededEnsemble(build func() *ConfigGenerator, seeds []uint64, maxConcurrency int) []EnsembleRun

RunSeededEnsemble launches one independent PartitionCoordinator per seed and runs them concurrently, varying the global seed applied to each via the ConfigGenerator. It returns one EnsembleRun per seed, index-aligned to the seeds slice.

The build closure MUST construct a fresh ConfigGenerator (and therefore fresh Iteration instances) on every call. This is load-bearing: ConfigGenerator.GenerateConfigs hands back the same Iteration pointers it was given and reconfigures them in place, so two members sharing one generator would share mutable iteration state (RNGs, buffers) and race. Building anew per member guarantees isolation.

Each member’s OutputFunction is replaced with a fresh StateTimeStorage sink so its trajectory is captured into the returned EnsembleRun; the member’s OutputCondition (and every other part of its SimulationConfig, including any ExecutionStrategy such as PersistentWorkerExecution) is respected.

maxConcurrency bounds how many members run at once; values <= 0 default to runtime.GOMAXPROCS(0). Results are deterministic: re-running with the same seeds yields identical per-member output regardless of maxConcurrency.

type EveryNStepsOutputCondition

EveryNStepsOutputCondition emits output once every N steps.

type EveryNStepsOutputCondition struct {
    N int
}

func (*EveryNStepsOutputCondition) IsOutputStep

func (c *EveryNStepsOutputCondition) IsOutputStep(partitionName string, state []float64, timestepsHistory *CumulativeTimestepsHistory) bool

type EveryStepOutputCondition

EveryStepOutputCondition calls the OutputFunction at every step.

type EveryStepOutputCondition struct{}

func (*EveryStepOutputCondition) IsOutputStep

func (c *EveryStepOutputCondition) IsOutputStep(partitionName string, state []float64, timestepsHistory *CumulativeTimestepsHistory) bool

type ExecutionStrategy

ExecutionStrategy chooses how a PartitionCoordinator advances — the policy for turning per-partition work into completed steps (spawn a goroutine per partition per step, keep one persistent worker per partition, run everything inline on the calling goroutine, …).

A nil ExecutionStrategy on a coordinator (or on Implementations) selects the default spawn-per-step two-phase execution. A strategy is purely an execution-policy choice, not a semantic one: every strategy must produce byte-identical output to the default for the same Settings and Implementations. This invariant is enforced by the cross-strategy equivalence tests.

A strategy’s single primitive is NewStepper, which builds a Stepper holding whatever per-run state the policy needs. Both batch execution (PartitionCoordinator.Run) and manual stepwise driving (PartitionCoordinator.NewStepper) are expressed in terms of it, so every strategy is steppable in exactly the same way the default two-phase algorithm is — there is no strategy that can only run to termination.

type ExecutionStrategy interface {
    // NewStepper returns a Stepper that advances c one step at a time under
    // this strategy's execution policy.
    NewStepper(c *PartitionCoordinator) Stepper
}

func ResolveExecutionStrategy

func ResolveExecutionStrategy(spec ComponentSpec) (ExecutionStrategy, error)

ResolveExecutionStrategy builds an ExecutionStrategy from a data spec. The three strategies are zero-field structs, so each is a nullary construction. An empty (omitted) spec resolves to nil, which selects the default spawn-per-step policy.

type ExponentialDistributionTimestepFunction

ExponentialDistributionTimestepFunction draws dt from an exponential distribution parameterised by Mean and Seed.

type ExponentialDistributionTimestepFunction struct {
    Mean float64
    Seed uint64
    // contains filtered or unexported fields
}

func NewExponentialDistributionTimestepFunction

func NewExponentialDistributionTimestepFunction(mean float64, seed uint64) *ExponentialDistributionTimestepFunction

NewExponentialDistributionTimestepFunction constructs an exponential-dt timestep function given mean and seed.

func (*ExponentialDistributionTimestepFunction) NextIncrement

func (t *ExponentialDistributionTimestepFunction) NextIncrement(timestepsHistory *CumulativeTimestepsHistory) float64

type FinalizingOutputFunction

FinalizingOutputFunction is the optional counterpart to OutputFunction for sinks that hold a resource which must be flushed, sealed or released once the run is over — a columnar buffer that only becomes a readable batch after the last row, a database handle that ingests in one shot, an open file.

PartitionCoordinator.Run calls Finalize exactly once, after the final step and before returning, on an OutputFunction that implements this. It is an OPTIONAL interface deliberately: OutputFunction stays two methods, every existing sink is unaffected, and a sink that needs no teardown simply does not implement it.

type FinalizingOutputFunction interface {
    OutputFunction
    Finalize()
}

type Implementations

Implementations provides concrete implementations for a simulation run.

type Implementations struct {
    Iterations           []Iteration
    OutputCondition      OutputCondition
    OutputFunction       OutputFunction
    TerminationCondition TerminationCondition
    TimestepFunction     TimestepFunction
    ExecutionStrategy    ExecutionStrategy
}

type InlineExecution

InlineExecution runs the simulation entirely on the calling goroutine: no worker goroutines, no channel handshakes and no WaitGroup barrier. Each step runs the iteration phase for every partition and then the update phase for every partition, in index order, by calling the iterators directly.

This is the only strategy that synchronises nothing per step, so it is the one that reaches serial speed when concurrency buys nothing — most obviously a single-partition run, where the default per-step goroutine spawn and channel round-trip are pure overhead.

Within-step params_from_upstream edges are supported, but because inline execution has no blocking channel handshake to wait on, every upstream producer must be ordered before its downstream consumers: the producer’s staged output is read directly, so it must already have run this step. NewStepper validates this up front and panics with a clear message if any consumer is ordered before (or at the same index as) one of its upstreams — which also catches cycles — rather than silently reading stale values. Reorder the partitions so upstreams precede consumers, or use a concurrent strategy. Partitions coupled only through state-history reads (which are lag-based and need no within-step handshake) are unaffected by the ordering rule.

Output is byte-identical to the default strategy: the two phases are still applied in order, so the iteration phase observes the previous step’s committed history exactly as the barrier guarantees, and upstream params carry the same current-step producer output the channel broadcast would.

This strategy is stateless and safe to share across coordinators.

type InlineExecution struct{}

func (*InlineExecution) NewStepper

func (e *InlineExecution) NewStepper(c *PartitionCoordinator) Stepper

NewStepper validates the partition ordering and returns a Stepper that advances the coordinator inline on the calling goroutine. It panics if any consumer is not ordered strictly after all of its upstreams (which also rejects cycles and self-edges).

type Iteration

Iteration defines the interface for per-partition state update functions in stochadex simulations.

The Iteration interface is the fundamental building block for defining how simulation state evolves over time. Each partition in a simulation uses an Iteration to compute its next state values based on the current state, parameters, and time information.

Design Philosophy: The Iteration interface emphasizes modularity and composability. By providing a simple, well-defined interface, it enables the creation of complex simulations through the combination of simple, focused iterations. This design supports both built-in iteration types and custom user-defined iterations.

Interface Methods:

Configuration Phase: Configure is called once per partition during simulation setup. It receives:

This phase is used for:

Iteration Phase: Iterate is called each simulation step to compute the next state values. It receives:

It must return:

Implementation Requirements:

Example Usage:

type MyIteration struct {
    // Internal state
}

func (m *MyIteration) Configure(partitionIndex int, settings *Settings) {
    // Initialize iteration
}

func (m *MyIteration) Iterate(params *Params, partitionIndex int,
                              stateHistories []*StateHistory,
                              timestepsHistory *CumulativeTimestepsHistory) []float64 {
    // Compute next state values
    return []float64{newValue1, newValue2, ...}
}

Common Iteration Types:

Performance Considerations:

Thread Safety:

type Iteration interface {
    Configure(partitionIndex int, settings *Settings)
    Iterate(
        params *Params,
        partitionIndex int,
        stateHistories []*StateHistory,
        timestepsHistory *CumulativeTimestepsHistory,
    ) []float64
}

type IterationSettings

IterationSettings is the YAML-loadable per-partition configuration.

Usage hints:

type IterationSettings struct {
    Name               string                    `yaml:"name"`
    Params             Params                    `yaml:"params"`
    ParamsFromUpstream map[string]UpstreamConfig `yaml:"params_from_upstream,omitempty"`
    InitStateValues    []float64                 `yaml:"init_state_values"`
    Seed               uint64                    `yaml:"seed"`
    StateWidth         int                       `yaml:"state_width"`
    StateHistoryDepth  int                       `yaml:"state_history_depth"`
}

type IterationTestHarness

IterationTestHarness wraps an iteration and performs checks on its behaviour while running.

type IterationTestHarness struct {
    Iteration Iteration
    Err       error
    // contains filtered or unexported fields
}

func (*IterationTestHarness) Configure

func (h *IterationTestHarness) Configure(partitionIndex int, settings *Settings)

func (*IterationTestHarness) Iterate

func (h *IterationTestHarness) Iterate(params *Params, partitionIndex int, stateHistories []*StateHistory, timestepsHistory *CumulativeTimestepsHistory) []float64

type IteratorInputMessage

IteratorInputMessage carries shared histories into iterator jobs.

type IteratorInputMessage struct {
    StateHistories   []*StateHistory
    TimestepsHistory *CumulativeTimestepsHistory
}

type JsonLogChannelOutputFunction

JsonLogChannelOutputFunction writes JSON log entries via a background goroutine using a channel for improved throughput.

type JsonLogChannelOutputFunction struct {
    // contains filtered or unexported fields
}

func NewJsonLogChannelOutputFunction

func NewJsonLogChannelOutputFunction(filePath string) *JsonLogChannelOutputFunction

NewJsonLogChannelOutputFunction creates a JsonLogChannelOutputFunction. Call Close (defer it) to ensure flushing at the end of a run.

func (*JsonLogChannelOutputFunction) Close

func (j *JsonLogChannelOutputFunction) Close()

Close flushes and stops the background writer. Defer it after construction. It blocks until the writer goroutine has drained the channel and flushed every buffered entry to the file, so callers may read the file once Close returns.

func (*JsonLogChannelOutputFunction) Configure

func (j *JsonLogChannelOutputFunction) Configure(*Settings)

func (*JsonLogChannelOutputFunction) Output

func (j *JsonLogChannelOutputFunction) Output(partitionName string, state []float64, cumulativeTimesteps float64)

type JsonLogEntry

JsonLogEntry is the serialised record format used by JSON log outputs.

type JsonLogEntry struct {
    PartitionName       string    `json:"partition_name"`
    State               []float64 `json:"state"`
    CumulativeTimesteps float64   `json:"time"`
}

type JsonLogOutputFunction

JsonLogOutputFunction writes newline-delimited JSON log entries.

type JsonLogOutputFunction struct {
    // contains filtered or unexported fields
}

func NewJsonLogOutputFunction

func NewJsonLogOutputFunction(filePath string) *JsonLogOutputFunction

NewJsonLogOutputFunction creates a new JsonLogOutputFunction.

func (*JsonLogOutputFunction) Configure

func (j *JsonLogOutputFunction) Configure(*Settings)

func (*JsonLogOutputFunction) Output

func (j *JsonLogOutputFunction) Output(partitionName string, state []float64, cumulativeTimesteps float64)

type NamedPartitionIndex

NamedPartitionIndex pairs the name of a partition with the partition index assigned to it by the PartitionCoordinator.

type NamedPartitionIndex struct {
    Name  string
    Index int
}

type NamedUpstreamConfig

NamedUpstreamConfig is like UpstreamConfig but refers to upstream by name.

type NamedUpstreamConfig struct {
    Upstream string `yaml:"upstream"`
    Indices  []int  `yaml:"indices,omitempty"`
}

type NilOutputCondition

NilOutputCondition never outputs.

type NilOutputCondition struct{}

func (*NilOutputCondition) IsOutputStep

func (c *NilOutputCondition) IsOutputStep(partitionName string, state []float64, timestepsHistory *CumulativeTimestepsHistory) bool

type NilOutputFunction

NilOutputFunction outputs nothing from the simulation.

type NilOutputFunction struct{}

func (*NilOutputFunction) Configure

func (f *NilOutputFunction) Configure(*Settings)

func (*NilOutputFunction) Output

func (f *NilOutputFunction) Output(partitionName string, state []float64, cumulativeTimesteps float64)

type NumberOfStepsTerminationCondition

NumberOfStepsTerminationCondition terminates after MaxNumberOfSteps.

type NumberOfStepsTerminationCondition struct {
    MaxNumberOfSteps int
}

func (*NumberOfStepsTerminationCondition) Terminate

func (t *NumberOfStepsTerminationCondition) Terminate(stateHistories []*StateHistory, timestepsHistory *CumulativeTimestepsHistory) bool

type OnlyGivenPartitionsOutputCondition

OnlyGivenPartitionsOutputCondition emits output only for listed partitions.

type OnlyGivenPartitionsOutputCondition struct {
    Partitions map[string]bool
}

func (*OnlyGivenPartitionsOutputCondition) IsOutputStep

func (o *OnlyGivenPartitionsOutputCondition) IsOutputStep(partitionName string, state []float64, timestepsHistory *CumulativeTimestepsHistory) bool

type OutputCondition

OutputCondition decides whether an output should be emitted this step.

type OutputCondition interface {
    IsOutputStep(partitionName string, state []float64, timestepsHistory *CumulativeTimestepsHistory) bool
}

func ResolveOutputCondition

func ResolveOutputCondition(spec ComponentSpec) (OutputCondition, error)

ResolveOutputCondition builds an OutputCondition from a data spec.

type OutputFunction

OutputFunction writes state/time to an output sink when the OutputCondition is met.

Configure is called once before parallel output begins (from NewPartitionCoordinator). Use it to pre-register partition names, cache indices, or open resources. Implementations that need no setup can leave it empty.

type OutputFunction interface {
    Configure(settings *Settings)
    Output(partitionName string, state []float64, cumulativeTimesteps float64)
}

func ResolveOutputFunction

func ResolveOutputFunction(spec ComponentSpec) (OutputFunction, error)

ResolveOutputFunction builds an OutputFunction from a data spec. Live-object sinks (state storage, channel, websocket) have no data form and are absent.

type Params

Params stores per-partition parameter values.

Usage hints:

type Params struct {
    Map map[string][]float64 `yaml:",inline"`
    // contains filtered or unexported fields
}

func NewParams

func NewParams(params map[string][]float64) Params

NewParams constructs a Params instance.

func (*Params) Get

func (p *Params) Get(name string) []float64

Get returns parameter values or panics with a helpful message.

func (*Params) GetCopy

func (p *Params) GetCopy(name string) []float64

GetCopy returns a copy of parameter values or panics with a helpful message.

func (*Params) GetCopyOk

func (p *Params) GetCopyOk(name string) ([]float64, bool)

GetCopyOk returns a copy of parameter values if present along with a flag.

func (*Params) GetIndex

func (p *Params) GetIndex(name string, index int) float64

GetIndex returns a single parameter value or panics.

func (*Params) GetOk

func (p *Params) GetOk(name string) ([]float64, bool)

GetOk returns parameter values if present along with a boolean flag.

func (*Params) Set

func (p *Params) Set(name string, values []float64)

Set creates or updates parameter values by name.

func (*Params) SetIndex

func (p *Params) SetIndex(name string, index int, value float64)

SetIndex updates a single parameter value or panics on invalid index.

func (*Params) SetPartitionName

func (p *Params) SetPartitionName(name string)

SetPartitionName attaches the owning partition name for better errors.

type PartitionConfig

PartitionConfig defines a partition to add to a simulation.

Usage hints:

type PartitionConfig struct {
    Name string `yaml:"name"`
    // Iteration is the resolved iteration. It is set programmatically or by
    // resolving IterationSpec's data form at load time.
    Iteration Iteration `yaml:"-"`
    // IterationSpec is the loaded `iteration:` value as a {type: ...} data spec,
    // resolved into Iteration at load time by the api iteration registry. It is
    // empty when the partition's iteration comes from an expression or an embedded
    // run instead.
    IterationSpec      ComponentSpec                  `yaml:"iteration,omitempty"`
    Params             Params                         `yaml:"params"`
    ParamsAsPartitions map[string][]string            `yaml:"params_as_partitions,omitempty"`
    ParamsFromUpstream map[string]NamedUpstreamConfig `yaml:"params_from_upstream,omitempty"`
    InitStateValues    []float64                      `yaml:"init_state_values"`
    StateHistoryDepth  int                            `yaml:"state_history_depth"`
    Seed               uint64                         `yaml:"seed"`
}

func LoadPartitionConfigFromYaml

func LoadPartitionConfigFromYaml(path string) *PartitionConfig

LoadPartitionConfigFromYaml loads PartitionConfig from a YAML file path.

Usage hints:

func (*PartitionConfig) Init

func (p *PartitionConfig) Init()

Init ensures params maps are initialised; call after unmarshalling YAML.

type PartitionConfigOrdering

PartitionConfigOrdering maintains the ordering and lookup for partitions. Can be updated dynamically via Append.

type PartitionConfigOrdering struct {
    Names        []string
    IndexByName  map[string]int
    ConfigByName map[string]*PartitionConfig
}

func (*PartitionConfigOrdering) Append

func (p *PartitionConfigOrdering) Append(config *PartitionConfig)

Append inserts another partition into the ordering and updates lookups.

type PartitionCoordinator

PartitionCoordinator orchestrates iteration work across partitions and applies state/time history updates in a coordinated manner.

The PartitionCoordinator is the central component that manages the execution of all partitions in a simulation. It coordinates the timing, communication, and state updates across all partitions, ensuring proper synchronization and maintaining simulation consistency.

Architecture: The coordinator uses a two-phase execution model:

  1. Iteration Phase: All partitions compute their next state values
  2. Update Phase: State and time histories are updated with new values

This design ensures that all partitions see consistent state information during each iteration, preventing race conditions and maintaining simulation determinism.

Concurrency Model:

Execution Flow:

  1. Compute next timestep increment using TimestepFunction
  2. Request iterations from all partitions (parallel execution)
  3. Wait for all iterations to complete
  4. Update state and time histories (parallel execution)
  5. Check termination condition
  6. Repeat until termination

Fields:

Example Usage:

coordinator := NewPartitionCoordinator(settings, implementations)

// Run simulation until termination
coordinator.Run()

// Or step-by-step control under the configured execution strategy
stepper := coordinator.NewStepper()
defer stepper.Close()
for !coordinator.ReadyToTerminate() {
    stepper.Step()
}

Performance:

Thread Safety:

type PartitionCoordinator struct {
    Iterators            []*StateIterator
    Shared               *IteratorInputMessage
    TimestepFunction     TimestepFunction
    TerminationCondition TerminationCondition
    RunStrategy          ExecutionStrategy
    // OutputFunction is retained solely so Run can Finalize a sink that implements
    // FinalizingOutputFunction; per-step output goes through the iterators.
    OutputFunction OutputFunction
    // contains filtered or unexported fields
}

func NewPartitionCoordinator

func NewPartitionCoordinator(settings *Settings, implementations *Implementations) *PartitionCoordinator

NewPartitionCoordinator wires Settings and Implementations into a runnable coordinator with initial state/time histories and channels.

func (*PartitionCoordinator) NewStepper

func (c *PartitionCoordinator) NewStepper() Stepper

NewStepper returns a Stepper that advances the coordinator one step at a time under its configured RunStrategy (a nil RunStrategy selects the default spawn-per-step execution). This is the strategy-aware counterpart to Step: it lets callers drive any execution strategy stepwise — inspecting or mutating state between steps — exactly as the default algorithm can be driven with Step, while keeping that strategy’s execution policy (persistent workers, inline execution, …).

The caller drives Step until ReadyToTerminate reports true and must call the stepper’s Close when done to release any resources it holds:

stepper := coordinator.NewStepper()
defer stepper.Close()
for !coordinator.ReadyToTerminate() {
    stepper.Step()
}

func (*PartitionCoordinator) ReadyToTerminate

func (c *PartitionCoordinator) ReadyToTerminate() bool

ReadyToTerminate returns whether the TerminationCondition is met.

func (*PartitionCoordinator) RequestMoreIterations

func (c *PartitionCoordinator) RequestMoreIterations(wg *sync.WaitGroup)

RequestMoreIterations spawns a goroutine per partition to run ReceiveAndIteratePending.

func (*PartitionCoordinator) Run

func (c *PartitionCoordinator) Run()

Run advances the coordinator to termination under its configured RunStrategy (a nil RunStrategy selects the default spawn-per-step two-phase execution). It is the canonical run loop shared by every strategy: build a Stepper, step until termination, then release the stepper.

func (*PartitionCoordinator) Step

func (c *PartitionCoordinator) Step(wg *sync.WaitGroup)

Step performs one simulation tick under the default spawn-per-step execution: compute dt, request iterations, then apply state/time updates. It is the single-step primitive the SpawnPerStepExecution stepper delegates to; other strategies advance a step through their own Stepper. Callers that want to drive a step under the coordinator’s configured strategy should use NewStepper instead.

func (*PartitionCoordinator) UpdateHistory

func (c *PartitionCoordinator) UpdateHistory(wg *sync.WaitGroup)

UpdateHistory spawns a goroutine per partition to run UpdateHistory and shifts time history forward, adding NextIncrement to t[0].

type PartitionState

PartitionState carries one partition’s state vector at a single cumulative-time point.

type PartitionState struct {

    // Simulation time (the cumulative sum of timestep increments) at this output.
    CumulativeTimesteps float64 `protobuf:"fixed64,1,opt,name=cumulative_timesteps,json=cumulativeTimesteps,proto3" json:"cumulative_timesteps,omitempty"`
    // Name of the partition that produced this state.
    PartitionName string `protobuf:"bytes,2,opt,name=partition_name,json=partitionName,proto3" json:"partition_name,omitempty"`
    // The partition's state vector for this step (length = the partition's state width).
    State []float64 `protobuf:"fixed64,3,rep,packed,name=state,proto3" json:"state,omitempty"`
    // contains filtered or unexported fields
}

func (*PartitionState) Descriptor

func (*PartitionState) Descriptor() ([]byte, []int)

Deprecated: Use PartitionState.ProtoReflect.Descriptor instead.

func (*PartitionState) GetCumulativeTimesteps

func (x *PartitionState) GetCumulativeTimesteps() float64

func (*PartitionState) GetPartitionName

func (x *PartitionState) GetPartitionName() string

func (*PartitionState) GetState

func (x *PartitionState) GetState() []float64

func (*PartitionState) ProtoMessage

func (*PartitionState) ProtoMessage()

func (*PartitionState) ProtoReflect

func (x *PartitionState) ProtoReflect() protoreflect.Message

func (*PartitionState) Reset

func (x *PartitionState) Reset()

func (*PartitionState) String

func (x *PartitionState) String() string

type PersistentWorkerExecution

PersistentWorkerExecution runs the simulation with one long-lived goroutine per partition rather than spawning a fresh goroutine per partition per phase per step. Each worker loops “wait-for-iterate -> iterate -> signal-done -> wait-for-update -> update -> signal-done”, which removes the per-step goroutine spawn/teardown cost.

The two-phase barrier is retained: workers are still woken and acknowledged once per phase so the update phase observes a consistent snapshot. This strategy therefore moves the per-step constant down (no spawn allocations) but keeps the per-step cross-goroutine synchronisation; it does not cross the serial floor for trivially small per-step work.

Output is byte-identical to the default strategy: the per-partition work and the barrier ordering are unchanged; only the goroutine lifetime differs. The workers are spawned by NewStepper and torn down by the returned Stepper’s Close, so a stepwise caller keeps the same persistent workers across every Step rather than paying setup per step.

This strategy is stateless and safe to share across coordinators; all per-run state lives on the Stepper.

type PersistentWorkerExecution struct{}

func (*PersistentWorkerExecution) NewStepper

func (e *PersistentWorkerExecution) NewStepper(c *PartitionCoordinator) Stepper

NewStepper spins up one long-lived worker goroutine per partition and returns a Stepper that drives them through the two-phase barrier each Step. The workers run until the Stepper’s Close is called.

type ReentrantRun

ReentrantRun describes one evaluation of a sub-simulation: where it starts, how its randomness is fixed, and how long it goes on for.

type ReentrantRun struct {
    // Rows sets each partition's initial row, in partition order. A nil entry,
    // or a short slice, leaves that partition's configured values alone.
    Rows [][]float64
    // Histories seeds whole state-history windows for the named partition
    // indices, for models that read further back than one step. It takes
    // precedence over Rows for those partitions.
    Histories map[int]*StateHistory
    // InitTimeValue overrides the sub-simulation's start time when non-nil.
    InitTimeValue *float64
    // Seed, when non-nil, reseeds every iteration before the run so the result
    // depends only on this run's inputs. Nil leaves the iterations' random
    // streams where the last run left them — the streaming behaviour that
    // general.EmbeddedSimulationRunIteration has by default.
    Seed *uint64
    // Steps is how many steps to run. Zero defers to the sub-simulation's own
    // termination condition instead.
    Steps int
    // AfterConfigure runs once the iterations have been (re)configured and
    // before the run starts. Reseeding calls Configure, which the framework
    // requires to re-initialise all mutable state — so anything injected into an
    // iteration from outside (see general.StateMemoryIteration) has to be
    // re-applied here, or a reseeded run would lose it.
    AfterConfigure func()
}

type ReentrantSimulation

ReentrantSimulation evaluates a sub-simulation as a pure function of a starting state, a seed, and a step count.

It owns the settings and implementations it is given: Advance mutates their initial values, params and seeds in place. Do not share one across goroutines, and do not hand the same Implementations to a coordinator running concurrently.

type ReentrantSimulation struct {
    // contains filtered or unexported fields
}

func NewReentrantSimulation

func NewReentrantSimulation(settings *Settings, implementations *Implementations) *ReentrantSimulation

NewReentrantSimulation wraps a configured sub-simulation for re-entrant use.

The caller usually wants implementations.ExecutionStrategy set to &InlineExecution{}: a re-entrant run is typically short and small, so the per-step goroutine round-trip of the default strategy dominates its cost.

func (*ReentrantSimulation) Advance

func (r *ReentrantSimulation) Advance(rows [][]float64, seed uint64, steps int) [][]float64

Advance is the pure, fixed-length form of Run: evaluate the sub-simulation for steps steps from rows, under seed. It is the shape a planner wants, where a transition has to be a function of (state, action).

func (*ReentrantSimulation) PartitionIndex

func (r *ReentrantSimulation) PartitionIndex(name string) (int, bool)

PartitionIndex returns the index of a named partition, and whether it exists.

func (*ReentrantSimulation) Run

func (r *ReentrantSimulation) Run(run ReentrantRun) [][]float64

Run evaluates the sub-simulation and returns the resulting rows, one per partition in partition order.

With Seed set the run is pure with respect to its inputs: repeating a call repeats its result, whatever ran in between. Rows are copied in and out, so the caller’s slices are never retained or mutated. Callers that only want the rows concatenated should prefer RunInto, which reuses a buffer instead of allocating one slice per partition.

func (*ReentrantSimulation) RunInto

func (r *ReentrantSimulation) RunInto(run ReentrantRun, dst []float64) []float64

RunInto is Run for callers that want every partition’s final row concatenated into one slice they own. dst is truncated and reused, so a caller holding one buffer across runs pays no per-partition allocation — the shape an iteration wants, since its own return value is a single row.

The returned slice aliases dst. Treat it as valid only until the next call.

func (*ReentrantSimulation) RunWindows

func (r *ReentrantSimulation) RunWindows(run ReentrantRun) [][]float64

RunWindows is Run for callers that need each partition’s whole state-history window back rather than just its latest row — models whose iterations read further back than one step, where the window IS part of the state.

Each returned slice is one partition’s window flattened row-major with the latest row first, so it round-trips through NewStateHistoryFromWindow.

func (*ReentrantSimulation) SetParam

func (r *ReentrantSimulation) SetParam(partition int, key string, values []float64)

SetParam sets a params key on a partition by index, which is how an input that is not part of the state — a control, an action, a parameter draw — enters a re-entrant run.

func (*ReentrantSimulation) StateWidths

func (r *ReentrantSimulation) StateWidths() []int

StateWidths returns each partition’s state width, in partition order.

type Settings

Settings is the YAML-loadable top-level simulation configuration.

type Settings struct {
    Iterations            []IterationSettings `yaml:"iterations"`
    InitTimeValue         float64             `yaml:"init_time_value"`
    TimestepsHistoryDepth int                 `yaml:"timesteps_history_depth"`
}

func LoadSettingsFromYaml

func LoadSettingsFromYaml(path string) *Settings

LoadSettingsFromYaml loads Settings from a YAML file path.

Usage hints:

func (*Settings) Init

func (s *Settings) Init()

Init fills in defaults and ensures maps are initialised. Call immediately after unmarshalling from YAML.

type SimulationConfig

SimulationConfig defines additional run-level configuration.

type SimulationConfig struct {
    OutputCondition      OutputCondition
    OutputFunction       OutputFunction
    TerminationCondition TerminationCondition
    TimestepFunction     TimestepFunction
    InitTimeValue        float64
    ExecutionStrategy    ExecutionStrategy
}

type SimulationConfigStrings

SimulationConfigStrings is the YAML-loadable version of SimulationConfig. Each component field is a ComponentSpec data spec ({type: …}) resolved at load time by the registry, needing no Go toolchain. ExecutionStrategy is optional and resolves to nil (the default spawn-per-step policy) when omitted.

type SimulationConfigStrings struct {
    OutputCondition      ComponentSpec `yaml:"output_condition"`
    OutputFunction       ComponentSpec `yaml:"output_function"`
    TerminationCondition ComponentSpec `yaml:"termination_condition"`
    TimestepFunction     ComponentSpec `yaml:"timestep_function"`
    InitTimeValue        float64       `yaml:"init_time_value"`
    ExecutionStrategy    ComponentSpec `yaml:"execution_strategy,omitempty"`
}

func LoadSimulationConfigStringsFromYaml

func LoadSimulationConfigStringsFromYaml(path string) *SimulationConfigStrings

LoadSimulationConfigStringsFromYaml loads SimulationConfigStrings from YAML.

func (*SimulationConfigStrings) ResolveDataComponents

func (s *SimulationConfigStrings) ResolveDataComponents() (*SimulationConfig, error)

ResolveDataComponents returns a SimulationConfig with every component data spec constructed and InitTimeValue copied across. An omitted component is left nil. ExecutionStrategy resolves to nil (the default spawn-per-step policy) when omitted. It errors if a data spec names an unknown type or a bad field.

type SpawnPerStepExecution

SpawnPerStepExecution is the default execution strategy: each step spawns one goroutine per partition for the iteration phase and again for the update phase, synchronised by a two-phase barrier. It is the named, explicitly selectable form of the behaviour used when no strategy is configured.

This strategy is stateless and safe to share across coordinators.

type SpawnPerStepExecution struct{}

func (*SpawnPerStepExecution) NewStepper

func (e *SpawnPerStepExecution) NewStepper(c *PartitionCoordinator) Stepper

NewStepper returns a Stepper that advances the coordinator using the default spawn-per-step two-phase execution (one goroutine per partition per phase).

type StateHistory

StateHistory is a rolling window of state vectors.

Usage hints:

type StateHistory struct {
    // each row is a different state in the history, by convention,
    // starting with the most recent at index = 0
    Values *mat.Dense
    // NextValues is a per-partition reusable scratch buffer of length =
    // StateWidth, pre-allocated when the history is constructed. An Iteration
    // may write its next state into this buffer and return it to avoid
    // allocating a fresh row every step (GetNextStateRowToUpdate hands it back
    // pre-filled with the current state).
    NextValues        []float64
    StateWidth        int
    StateHistoryDepth int
}

func NewStateHistoryFromWindow

func NewStateHistoryFromWindow(window []float64, width, depth int) *StateHistory

NewStateHistoryFromWindow builds a StateHistory from a flattened window, row-major with the latest row first — the shape ReentrantSimulation.RunWindows returns, so a caller can carry a model’s whole window through its own state encoding and hand it back to start the next run.

func (*StateHistory) CopyStateRow

func (s *StateHistory) CopyStateRow(index int) []float64

CopyStateRow copies a row from the state history given the index.

func (*StateHistory) GetNextStateRowToUpdate

func (s *StateHistory) GetNextStateRowToUpdate() []float64

GetNextStateRowToUpdate returns the partition’s reusable NextValues buffer pre-filled with a copy of the current state (row 0), ready for the iteration to mutate and return.

It always copies — never exposes a row of Values directly — so that mutating and returning the result cannot corrupt live history or any retained output.

type StateIterator

StateIterator runs an Iteration for a partition on a goroutine and manages reads/writes to history and output.

type StateIterator struct {
    Iteration       Iteration
    Params          Params
    Partition       NamedPartitionIndex
    ValueChannels   StateValueChannels
    OutputCondition OutputCondition
    OutputFunction  OutputFunction
}

func NewStateIterator

func NewStateIterator(iteration Iteration, params Params, partitionName string, partitionIndex int, valueChannels StateValueChannels, outputCondition OutputCondition, outputFunction OutputFunction, initState []float64, timestepsHistory *CumulativeTimestepsHistory) *StateIterator

NewStateIterator creates a StateIterator and may emit initial output if the condition is met by the initial state/time.

func (*StateIterator) ApplyHistoryUpdate

func (s *StateIterator) ApplyHistoryUpdate(inputMessage *IteratorInputMessage)

ApplyHistoryUpdate applies the pending state update to the partition history for the given input message. It is the work performed by the update phase, factored out from the channel receive so that long-lived workers can own the receive themselves.

func (*StateIterator) Iterate

func (s *StateIterator) Iterate(stateHistories []*StateHistory, timestepsHistory *CumulativeTimestepsHistory) []float64

Iterate runs the Iteration and optionally triggers output if the condition is met for the new state/time.

func (*StateIterator) IteratePending

func (s *StateIterator) IteratePending(inputMessage *IteratorInputMessage)

IteratePending updates upstream-driven params, runs Iterate, and stores a pending state update for the given input message. It is the work performed by the iteration phase, factored out from the channel receive so that long-lived workers can own the receive themselves.

func (*StateIterator) IteratePendingInline

func (s *StateIterator) IteratePendingInline(inputMessage *IteratorInputMessage)

IteratePendingInline runs the iteration phase for inline execution: it reads upstream-driven params directly from producers’ staged NextValues (no channels) and does not broadcast downstream. Callers must process partitions in an order where every upstream precedes its downstream consumers, which InlineExecution validates before running.

func (*StateIterator) ReceiveAndIteratePending

func (s *StateIterator) ReceiveAndIteratePending(inputChannel <-chan *IteratorInputMessage)

ReceiveAndIteratePending listens for an IteratorInputMessage, updates upstream-driven params, runs Iterate, and stores a pending state update.

func (*StateIterator) UpdateHistory

func (s *StateIterator) UpdateHistory(inputChannel <-chan *IteratorInputMessage)

UpdateHistory applies the pending state update to the partition history.

type StateTimeStorage

StateTimeStorage stores simulation time series data organised by partition name.

Two append paths serve different use cases:

GetValues, GetTimes, GetNames, SetValues, SetTimes and the registration methods (PreRegisterPartitions, GetIndex, IndexOf) are all intended for single-goroutine setup or post-simulation use.

The only internal synchronisation that remains is a mutex guarding the shared times slice, since N partition goroutines may all call appendTimeIfNew with the same timestamp; an atomic fast-path skips the mutex in the common case where the timestamp is already recorded.

type StateTimeStorage struct {
    // contains filtered or unexported fields
}

func NewStateTimeStorage

func NewStateTimeStorage() *StateTimeStorage

NewStateTimeStorage constructs a new StateTimeStorage.

func (*StateTimeStorage) Append

func (s *StateTimeStorage) Append(name string, time float64, values []float64)

Append appends values for name and records time. Not safe for concurrent use; intended for single-goroutine data loading.

func (*StateTimeStorage) AppendByIndex

func (s *StateTimeStorage) AppendByIndex(index int, time float64, values []float64)

AppendByIndex appends values for a pre-registered partition index and records time at most once per unique timestamp.

Lock-free for the store. Preconditions (all hold under normal coordinator use):

func (*StateTimeStorage) GetIndex

func (s *StateTimeStorage) GetIndex(name string) int

GetIndex returns or creates the index for name.

func (*StateTimeStorage) GetNames

func (s *StateTimeStorage) GetNames() []string

GetNames returns all registered partition names.

func (*StateTimeStorage) GetTimes

func (s *StateTimeStorage) GetTimes() []float64

GetTimes returns a snapshot of the time axis.

func (*StateTimeStorage) GetValues

func (s *StateTimeStorage) GetValues(name string) [][]float64

GetValues returns a snapshot of all time series rows for name, panicking if absent.

func (*StateTimeStorage) IndexOf

func (s *StateTimeStorage) IndexOf(name string) (int, bool)

IndexOf returns the index and true if name is registered, or 0 and false.

func (*StateTimeStorage) PreRegisterPartitions

func (s *StateTimeStorage) PreRegisterPartitions(names []string)

PreRegisterPartitions ensures each name has a stable index and an empty row buffer before AppendByIndex is called concurrently. Idempotent.

func (*StateTimeStorage) SetTimes

func (s *StateTimeStorage) SetTimes(times []float64)

SetTimes replaces the time axis.

func (*StateTimeStorage) SetValues

func (s *StateTimeStorage) SetValues(name string, values [][]float64)

SetValues replaces the entire series for name.

type StateTimeStorageOutputFunction

StateTimeStorageOutputFunction stores output into StateTimeStorage when the condition is met.

type StateTimeStorageOutputFunction struct {
    Store *StateTimeStorage
    // contains filtered or unexported fields
}

func (*StateTimeStorageOutputFunction) Configure

func (f *StateTimeStorageOutputFunction) Configure(settings *Settings)

Configure pre-registers all partition names on Store and caches their indices for lock-free lookup in Output. Safe to call multiple times.

func (*StateTimeStorageOutputFunction) Output

func (f *StateTimeStorageOutputFunction) Output(partitionName string, state []float64, cumulativeTimesteps float64)

type StateValueChannels

StateValueChannels provides upstream/downstream channels for inter-iterator communication.

type StateValueChannels struct {
    Upstreams  map[string]*UpstreamStateValues
    Downstream *DownstreamStateValues
}

func (*StateValueChannels) BroadcastDownstream

func (s *StateValueChannels) BroadcastDownstream(stateValues []float64)

BroadcastDownstream sends state values to all configured downstream copies. Each listener receives an independent copy so params wiring cannot mutate a slice shared with other partitions or with the producer’s state buffer.

func (*StateValueChannels) UpdateUpstreamParams

func (s *StateValueChannels) UpdateUpstreamParams(params *Params)

UpdateUpstreamParams updates Params with values received from upstream channels.

func (*StateValueChannels) UpdateUpstreamParamsInline

func (s *StateValueChannels) UpdateUpstreamParamsInline(params *Params, stateHistories []*StateHistory)

UpdateUpstreamParamsInline updates Params with values read directly from the producers’ staged NextValues, without any channel handshake. It is the inline-execution counterpart to UpdateUpstreamParams and requires that every upstream producer has already run its iteration phase this step (i.e. that partitions are processed in an order where upstreams precede consumers).

type StdoutOutputFunction

StdoutOutputFunction outputs the state to the terminal.

type StdoutOutputFunction struct{}

func (*StdoutOutputFunction) Configure

func (s *StdoutOutputFunction) Configure(*Settings)

func (*StdoutOutputFunction) Output

func (s *StdoutOutputFunction) Output(partitionName string, state []float64, cumulativeTimesteps float64)

type Stepper

Stepper advances a PartitionCoordinator one step at a time under a specific ExecutionStrategy’s policy. It owns the per-run state that policy needs — e.g. the persistent worker goroutines of PersistentWorkerExecution — which is why stepping is a distinct object rather than a bare coordinator method: that state lives across steps and must be released with Close when the run ends.

Typical stepwise use, which mirrors the default algorithm’s step-by-step control but keeps the chosen strategy’s execution policy:

stepper := coordinator.NewStepper()
defer stepper.Close()
for !coordinator.ReadyToTerminate() {
    stepper.Step()
    // inspect or mutate state between steps here (keyboard input,
    // external coupling, output throttling, per-step assertions, ...)
}

Step advances by exactly one simulation tick — compute the next timestep increment, run the iteration phase for every partition, then the update phase — leaving the coordinator in the same committed state the default algorithm reaches after one Step. Close releases resources and must be called exactly once; Step must not be called after Close.

type Stepper interface {
    Step()
    Close()
}

type TerminationCondition

TerminationCondition decides when the simulation should end.

type TerminationCondition interface {
    Terminate(
        stateHistories []*StateHistory,
        timestepsHistory *CumulativeTimestepsHistory,
    ) bool
}

func ResolveTerminationCondition

func ResolveTerminationCondition(spec ComponentSpec) (TerminationCondition, error)

ResolveTerminationCondition builds a TerminationCondition from a data spec.

type TimeElapsedTerminationCondition

TimeElapsedTerminationCondition terminates after MaxTimeElapsed.

type TimeElapsedTerminationCondition struct {
    MaxTimeElapsed float64
}

func (*TimeElapsedTerminationCondition) Terminate

func (t *TimeElapsedTerminationCondition) Terminate(stateHistories []*StateHistory, timestepsHistory *CumulativeTimestepsHistory) bool

type TimestepFunction

TimestepFunction computes the next time increment.

type TimestepFunction interface {
    NextIncrement(
        timestepsHistory *CumulativeTimestepsHistory,
    ) float64
}

func ResolveTimestepFunction

func ResolveTimestepFunction(spec ComponentSpec) (TimestepFunction, error)

ResolveTimestepFunction builds a TimestepFunction from a data spec.

type UpstreamConfig

UpstreamConfig is the YAML-loadable representation of a slice of data from the output of a partition which is computationally upstream.

type UpstreamConfig struct {
    Upstream int   `yaml:"upstream"`
    Indices  []int `yaml:"indices,omitempty"`
}

type UpstreamStateValues

UpstreamStateValues contains information to receive state values from an upstream iterator via channel.

Upstream is the partition index of the producer. It is unused by the channel-based strategies (which receive blockingly on Channel) but lets inline execution read the producer’s staged NextValues directly.

type UpstreamStateValues struct {
    Channel  chan []float64
    Indices  []int
    Upstream int
}

type WebsocketOutputFunction

WebsocketOutputFunction serialises and sends outputs via a websocket connection when the condition is met.

type WebsocketOutputFunction struct {
    // contains filtered or unexported fields
}

func NewWebsocketOutputFunction

func NewWebsocketOutputFunction(connection *websocket.Conn, mutex *sync.Mutex) *WebsocketOutputFunction

NewWebsocketOutputFunction constructs a WebsocketOutputFunction with a connection and a mutex for safe concurrent writes.

func (*WebsocketOutputFunction) Configure

func (w *WebsocketOutputFunction) Configure(*Settings)

func (*WebsocketOutputFunction) Output

func (w *WebsocketOutputFunction) Output(partitionName string, state []float64, cumulativeTimesteps float64)

Generated by gomarkdoc