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 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 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 drives a PartitionCoordinator’s Run loop.

A nil ExecutionStrategy on a coordinator (or on Implementations) selects the default spawn-per-step two-phase execution, which is exactly equivalent to repeatedly calling Step until termination. Strategies only ever change how Run advances the simulation; single Step semantics are never affected, so callers that drive a coordinator one Step at a time (including the test harnesses) always observe the default behaviour.

Every strategy must produce byte-identical output to the default for the same Settings and Implementations: a strategy is purely an execution-policy choice, not a semantic one. This invariant is enforced by the cross-strategy equivalence tests.

type ExecutionStrategy interface {
    // Run advances the coordinator from its current state until its
    // TerminationCondition is met.
    Run(c *PartitionCoordinator)
}

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 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. Run 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) Run

func (e *InlineExecution) Run(c *PartitionCoordinator)

Run advances the coordinator to termination inline on the calling goroutine.

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
}

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)
}

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          Iteration                      `yaml:"-"`
    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
for !coordinator.ReadyToTerminate() {
    var wg sync.WaitGroup
    coordinator.Step(&wg)
}

Performance:

Thread Safety:

type PartitionCoordinator struct {
    Iterators            []*StateIterator
    Shared               *IteratorInputMessage
    TimestepFunction     TimestepFunction
    TerminationCondition TerminationCondition
    RunStrategy          ExecutionStrategy
    // 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) 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 by repeatedly calling Step until termination.

When RunStrategy is non-nil it owns the whole run loop; otherwise Run uses the default spawn-per-step two-phase execution that is equivalent to repeatedly calling Step.

func (*PartitionCoordinator) Step

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

Step performs one simulation tick: compute dt, request iterations, then apply state/time updates.

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.

This strategy is stateless and safe to share across coordinators; all per-run state is created inside Run.

type PersistentWorkerExecution struct{}

func (*PersistentWorkerExecution) Run

func (e *PersistentWorkerExecution) Run(c *PartitionCoordinator)

Run spins up one worker goroutine per partition, advances the simulation through the two-phase barrier each step, then tears the workers down.

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, referring to implementations by type names for templating.

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

func LoadSimulationConfigStringsFromYaml

func LoadSimulationConfigStringsFromYaml(path string) *SimulationConfigStrings

LoadSimulationConfigStringsFromYaml loads SimulationConfigStrings from YAML.

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 Run uses when no strategy is configured.

This strategy is stateless and safe to share across coordinators.

type SpawnPerStepExecution struct{}

func (*SpawnPerStepExecution) Run

func (e *SpawnPerStepExecution) Run(c *PartitionCoordinator)

Run advances the coordinator until termination using the default spawn-per-step two-phase execution.

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 (*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 TerminationCondition

TerminationCondition decides when the simulation should end.

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

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
}

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