sim

package
v0.0.0-...-741d8d9 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 16, 2026 License: LGPL-3.0 Imports: 43 Imported by: 0

README

Simulation package

Network simulation infrastructure for testing erasure-coded broadcast strategies against a gossipsub baseline. Supports configurable topologies derived from real Ethereum node distribution, two execution drivers (Shadow for scale, simnet for fast iteration), and a Python CLI for orchestration.

Prerequisites

  • Python 3.12+ with uv for the simctl CLI
  • Go 1.25+ for building the simulation node
  • Shadow for shadow mode simulations (see shadow-rs.github.io)

Quick start

# Install CLI (from sim/cli directory)
cd sim/cli && uv sync

# Generate default config
simctl init -o config.yaml

# Run simulation (driver is set in config: simulation.driver)
simctl run config.yaml

# Compare strategies with an experiment
simctl experiment init -o experiment.yaml
# Edit experiment.yaml to configure strategies...
simctl experiment run experiment.yaml --output-dir=results/

Configuration

A single YAML schema drives both Go binaries and the Python CLI. The config has four sections: simulation, strategy, workload, and topology.

Run config

See cli/configs/small_rs.yaml for a minimal example.

simulation:
  driver: shadow               # shadow or simnet
  log_level: info              # debug, info, warn, error
  bandwidth_log_frequency_ms: 100

strategy:
  name: RS                     # RS, RS-ChunkLen, RLNC, RLNC-ChunkLen, gossipsub
  data_shards: 16
  parity_shards: 16
  enable_bitmaps: true
  bitmap_threshold: 100

workload:
  num_messages: 10
  message_size: 100000         # bytes
  publish_wait_seconds: 10.0
  stop_time_minutes: 30.0

topology:
  generate:                    # Python generates topology before running Go
    num_nodes: 100
    degree: 8
    type: random               # random or ring
    seed: 42
    super_node_fraction: 0.0

The topology section has two mutually exclusive modes:

  • topology.generate: Python generates a topology JSON file from these parameters, saves it alongside the run config, and rewrites the config with topology.file pointing to the generated file before invoking Go.
  • topology.file: Points directly to a pre-existing topology JSON file. Go reads this; it never sees generate parameters.

If Go encounters topology.generate without topology.file, it errors with a message directing you to run via simctl.

Strategy parameters

Each strategy reads only its own fields; unknown fields are ignored.

Strategy Fields
RS data_shards, parity_shards, enable_bitmaps, bitmap_threshold
RS-ChunkLen chunk_len, enable_bitmaps, bitmap_threshold
RLNC num_chunks, enable_bitmaps
RLNC-ChunkLen num_chunks_per_generation, target_chunk_size, origin_redundancy, enable_bitmaps
gossipsub (none)
Experiment config

Experiments share topology and workload across multiple strategy variations. See cli/configs/example_experiment.yaml.

name: strategy-comparison
description: "Compare RS, RLNC, and gossipsub"

simulation:
  driver: shadow
  log_level: info

topology:
  generate:
    num_nodes: 100
    degree: 8
    type: random
    seed: 42

workload:
  num_messages: 10
  message_size: 100000
  publish_wait_seconds: 10.0
  stop_time_minutes: 30.0

strategies:
  - name: RS
    data_shards: 16
    parity_shards: 16

  - name: RLNC
    num_chunks: 16

  - name: gossipsub
    publish_wait_seconds: 30.0

The topology is generated once and shared by all strategy runs within the experiment.

Topology generation

Topology generation is handled by Python (sim/cli/simctl/topology.py) using real-world data from sim/cli/data/:

data/country_weights.json contains the relative frequency of Ethereum nodes per country, derived from network crawl data. Examples: US (5031), Germany (2241), France (1032), Finland (962). When generating a topology, each node is assigned a country via weighted random selection, so a 100-node topology reflects the actual geographic distribution of the Ethereum network.

data/country_latencies.json is a country-to-country latency matrix (measured round-trip times). After nodes are placed and connected into a graph, each edge's latency is looked up from this matrix based on the countries of the source and target nodes.

Bandwidth is assigned by role:

  • Node 0 (block builder): 50 Mbps up / 100 Mbps down
  • Super nodes (controlled by super_node_fraction): 1024 / 1024 Mbps
  • Regular nodes: 50 / 50 Mbps

The generation algorithm:

  1. Assigns countries and bandwidth to all nodes
  2. Builds a spanning tree for connectivity (every node reachable)
  3. Adds random edges until each node reaches the target degree
  4. Looks up edge latencies from the country matrix

The output is a topology.json with nodes (num, upload/download bandwidth, country) and edges (source, target, latency). Go reads this JSON directly via LoadTopology() and ignores the country field.

CLI reference

simctl
├── init               # Generate default run config
├── topology           # Generate network topology standalone
├── run                # Run single simulation
├── analyze            # Analyze results
├── experiment
│   ├── init           # Generate experiment config
│   └── run            # Run multiple strategies
└── remote
    ├── run            # Run on remote host
    └── experiment     # Run experiment on remote host
simctl run
simctl run config.yaml
simctl run config.yaml --output-dir=results/

The driver (shadow or simnet) is read from simulation.driver in the config.

simctl experiment run
simctl experiment run experiment.yaml --output-dir=results/
simctl remote

Run simctl commands on a remote host. Syncs the local codebase via rsync, executes the command, monitors for completion, and tars the output on finish.

simctl remote run config.yaml --host=user@server
simctl remote experiment experiment.yaml --host=user@server
simctl remote experiment experiment.yaml --host=user@server --dry-run

Execution modes

Shadow mode uses the Shadow discrete-event network simulator for realistic large-scale testing (100+ nodes). Each node runs as a separate process with simulated network I/O. The Python CLI generates a Shadow YAML config with GML network graph, builds the simnode binary, and invokes Shadow.

Simnet mode runs all nodes in-process using Go's testing/synctest for deterministic execution. No Shadow installation required. Reliable up to ~16 nodes.

Strategies

RS and RLNC nodes are created via ECStrategy, which wraps the broadcast engine's Scheme interface. Gossipsub is a standalone libp2p implementation used as a baseline.

Strategy Description
RS Reed-Solomon erasure coding
RS-ChunkLen RS with fixed chunk size instead of fixed shard count
RLNC Random linear network coding
RLNC-ChunkLen RLNC with generations and target chunk size
gossipsub libp2p gossipsub baseline (no erasure coding)

Running tests

# Unit tests (fast, in-process simnet)
GOEXPERIMENT=synctest go test ./sim/... -v -short -count=1

# Large network tests (32 nodes, realistic topologies)
GOEXPERIMENT=synctest go test ./sim/... -v -run TestLargeNetwork -timeout=30m

Adding a new strategy

For erasure-coded strategies, implement a Scheme (see broadcast/rs, broadcast/rlnc) and wire it via ECStrategy in config.go. For non-EC strategies, implement the Node interface directly (see strategy_gossipsub.go). In both cases, add the strategy name to StrategyConfig.UnmarshalYAML and strategyFunc() in config.go, and to the Python config in cli/simctl/config.py.

Package layout

sim/
├── config.go              # RunConfig, StrategyConfig, NewNodeFunc
├── scenario.go            # Scenario orchestration, RunSimnetScenario
├── node.go                # Node interface
├── host.go                # QUIC host setup
├── driver.go              # Driver interface
├── driver_simnet.go       # In-process simnet driver
├── driver_shadow.go       # Shadow driver with serialized UDP writes
├── observer.go            # Observer for tracking bytes and chunk verdicts
├── collector.go           # Metrics collection
├── trace_writer.go        # Trace event output
├── trace_observer.go      # Trace-level observation
├── strategy_gossipsub.go  # Gossipsub node implementation
├── cmd/
│   └── shadow/            # Shadow binary (--config, --node-num)
└── cli/
    ├── simctl/            # Python CLI (config, runner, topology, experiment)
    └── data/
        ├── country_weights.json     # Ethereum node geographic distribution
        └── country_latencies.json   # Country-to-country RTT matrix

Known limitations

  • Tests using testing/synctest require GOEXPERIMENT=synctest
  • Shadow simulations require Shadow to be installed separately
  • Simnet is reliable up to ~16 nodes; beyond that, tests can stall

Documentation

Overview

package sim is a simple implementation of a host using quic for transport. It's provided here to explain the transport capabilities required to drive `broadcast.Broadcaster` and must not be used in production.

Index

Constants

View Source 
const DefaultListenPort = 8000

Variables

This section is empty.

Functions

This section is empty.

Types

type BandwidthEvent 

type BandwidthEvent struct {
	NodeNum            int
	SentBps            int
	ReceivedBps        int
	SentBytesTotal     int
	ReceivedBytesTotal int
	At                 time.Time
}

BandwidthEvent represents a periodic bandwidth usage sample from a node.

type BroadcastNode 

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

BroadcastNode wraps broadcast.Engine and a generic Channel to implement the Node interface. The generic Channel boundary is captured at construction via closures (publishFn, stopFn) and a receive channel.

func (*BroadcastNode) Addr 

func (n *BroadcastNode) Addr() net.Addr

func (*BroadcastNode) BandwidthStats 

func (n *BroadcastNode) BandwidthStats() (bytesSent, bytesReceived int)

func (*BroadcastNode) Close 

func (n *BroadcastNode) Close() error

func (*BroadcastNode) DialPeer 

func (n *BroadcastNode) DialPeer(ctx context.Context, p int, addr net.Addr) error

func (*BroadcastNode) NodeNum 

func (n *BroadcastNode) NodeNum() int

func (*BroadcastNode) Publish 

func (n *BroadcastNode) Publish(messageID string, data []byte)

func (*BroadcastNode) Receive 

func (n *BroadcastNode) Receive(ctx context.Context) (string, []byte, error)

func (*BroadcastNode) ResetBandwidthStats 

func (n *BroadcastNode) ResetBandwidthStats() (bytesSent, bytesReceived int)

ResetBandwidthStats returns bytes sent/received since the last reset and advances the baseline.

func (*BroadcastNode) Start 

func (n *BroadcastNode) Start(ctx context.Context)

type ChunkStats 

type ChunkStats struct {
	Accepted  int
	Redundant int
	Decoding  int
	Surplus   int
}

ChunkStats holds per-node chunk reception verdict counts.

type Driver 

type Driver interface {
	NewNode(nodeNum int, logger *slog.Logger) (Node, error)
	NodeAddr(nodeNum int) net.Addr
	Start()
	Close() error
}

Driver abstracts node creation and network management across execution backends (Shadow, simnet).

type EdgeSpec 

type EdgeSpec struct {
	Source    int `json:"source" yaml:"source"`
	Target    int `json:"target" yaml:"target"`
	LatencyMs int `json:"latency_ms" yaml:"latency_ms"`
}

EdgeSpec describes a directed edge in the topology.

type GossipsubNode 

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

GossipsubNode implements the Node interface using libp2p's gossipsub protocol.

func (*GossipsubNode) Addr 

func (g *GossipsubNode) Addr() net.Addr

func (*GossipsubNode) BandwidthStats 

func (g *GossipsubNode) BandwidthStats() (bytesSent, bytesReceived int)

func (*GossipsubNode) Close 

func (g *GossipsubNode) Close() error

func (*GossipsubNode) DialPeer 

func (g *GossipsubNode) DialPeer(ctx context.Context, nodeNum int, addr net.Addr) error

func (*GossipsubNode) NodeNum 

func (g *GossipsubNode) NodeNum() int

func (*GossipsubNode) Publish 

func (g *GossipsubNode) Publish(messageID string, data []byte)

func (*GossipsubNode) Receive 

func (g *GossipsubNode) Receive(ctx context.Context) (string, []byte, error)

func (*GossipsubNode) ResetBandwidthStats 

func (g *GossipsubNode) ResetBandwidthStats() (bytesSent, bytesReceived int)

func (*GossipsubNode) Start 

func (g *GossipsubNode) Start(ctx context.Context)

type LogCollector 

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

LogCollector reads Published, Received, and Bandwidth events from a Scenario and logs them. Call Run in a goroutine before starting the node.

func (*LogCollector) Run 

func (c *LogCollector) Run(ctx context.Context)

type Node 

type Node interface {
	Start(ctx context.Context)
	Publish(messageID string, data []byte)
	Receive(ctx context.Context) (messageID string, data []byte, err error)
	DialPeer(ctx context.Context, nodeNum int, addr net.Addr) error
	BandwidthStats() (sent, received int)
	ResetBandwidthStats() (sent, received int)
	Addr() net.Addr
	NodeNum() int
	Close() error
}

Node defines the interface for network simulation nodes.

func NewGossipsubNode 

func NewGossipsubNode(conn net.PacketConn, nodeNum int, logger *slog.Logger, tw *TraceWriter) (Node, error)

NewGossipsubNode creates a new GossipsubNode with QUIC transport.

type NodeEvent 

type NodeEvent struct {
	MessageID string
	Data      []byte
	NodeNum   int
	At        time.Time
}

NodeEvent represents a publish or receive event from a node.

type NodeSpec 

type NodeSpec struct {
	Num            int `json:"num" yaml:"num"`
	UploadBWMbps   int `json:"upload_bw_mbps" yaml:"upload_bw_mbps"`
	DownloadBWMbps int `json:"download_bw_mbps" yaml:"download_bw_mbps"`
}

NodeSpec describes a node in the simulation topology.

type Observer 

type Observer struct {
	broadcast.NoOpObserver
	// contains filtered or unexported fields
}

Observer tracks bytes sent bucketed by session role (origin vs relay) and chunk reception verdicts. Embed NoOpObserver for methods we don't override.

func NewObserver 

func NewObserver() *Observer

func (*Observer) Chunks 

func (o *Observer) Chunks() ChunkStats

Chunks returns chunk reception verdict counts.

func (*Observer) OnChunkRcvd 

func (o *Observer) OnChunkRcvd(_ broadcast.PeerID, _ broadcast.ChannelID, _ broadcast.MessageID, verdict broadcast.Verdict)

func (*Observer) OnChunkSent 

func (o *Observer) OnChunkSent(_ broadcast.PeerID, channelID broadcast.ChannelID, messageID broadcast.MessageID, bytesSent int)

func (*Observer) OnSessionStarted 

func (o *Observer) OnSessionStarted(channelID broadcast.ChannelID, messageID broadcast.MessageID, role broadcast.SessionRole)

func (*Observer) Reset 

func (o *Observer) Reset() ObserverSnapshot

Reset snapshots all counters and zeros them.

func (*Observer) Stats 

func (o *Observer) Stats() (originSent, relaySent int)

Stats returns total bytes sent as origin and as relay.

type ObserverSnapshot 

type ObserverSnapshot struct {
	Chunks     ChunkStats
	OriginSent int
	RelaySent  int
}

ObserverSnapshot holds a point-in-time snapshot of all observer counters.

type QUICHost 

type QUICHost struct {
	Transport *quic.Transport
	Listener  *quic.Listener
	UDPAddr   *net.UDPAddr
	Config    *quic.Config
}

QUICHost is QUIC transport and listener for ec broadcast. The host is provided for illustrating the transport requirements of an application that uses `broadcast.Broadcaster`.

func NewQUICHost 

func NewQUICHost(conn net.PacketConn) (QUICHost, error)

NewQUICHost creates a new `QUICHost`. The returned host listens for new connections on conn.

func (*QUICHost) Accept 

func (q *QUICHost) Accept(ctx context.Context) (*quic.Conn, error)

Accept accepts a new connection.

func (*QUICHost) Close 

func (q *QUICHost) Close() error

func (*QUICHost) Dial 

func (q *QUICHost) Dial(ctx context.Context, addr net.Addr, conf *quic.Config) (*quic.Conn, error)

Dial dials the host at `addr`.

type RSStrategyConfig 

type RSStrategyConfig struct {
	DataShards        int  `yaml:"data_shards"`
	ParityShards      int  `yaml:"parity_shards"`
	ChunkLen          int  `yaml:"chunk_len"`
	ForwardMultiplier int  `yaml:"forward_multiplier"`
	EnableBitmaps     bool `yaml:"enable_bitmaps"`
	BitmapThreshold   int  `yaml:"bitmap_threshold"`
}

RSStrategyConfig holds RS-specific parameters.

type RunConfig 

type RunConfig struct {
	Simulation SimulationConfig `yaml:"simulation"`
	Strategy   StrategyConfig   `yaml:"strategy"`
	Workload   WorkloadConfig   `yaml:"workload"`
	Topology   TopologyConfig   `yaml:"topology"`
}

RunConfig is the unified YAML configuration for a simulation run.

func LoadRunConfig 

func LoadRunConfig(path string) (*RunConfig, error)

LoadRunConfig reads and parses a YAML run config from path.

func (*RunConfig) BuildTraceHeaderOptions 

func (rc *RunConfig) BuildTraceHeaderOptions(topo Topology) (TraceHeaderOptions, error)

func (*RunConfig) LoadTopology 

func (rc *RunConfig) LoadTopology() (Topology, error)

LoadTopology reads the topology JSON file referenced by Topology.File. Returns an error if File is empty (topology not yet generated).

func (*RunConfig) NewScenario 

func (rc *RunConfig) NewScenario(driver string, logger *slog.Logger) (*Scenario, error)

NewScenario builds a Scenario from this config, selecting the environment based on the driver ("shadow" or "simnet").

type Scenario 

type Scenario struct {
	NumMessages           int
	MessageSize           int
	Driver                Driver
	BandwidthLogFrequency time.Duration
	Logger                *slog.Logger
	// contains filtered or unexported fields
}

Scenario orchestrates simulation test scenarios.

func (*Scenario) Close 

func (s *Scenario) Close()

Close shuts down the scenario.

func (*Scenario) NewLogCollector 

func (s *Scenario) NewLogCollector() *LogCollector

func (*Scenario) NewNode 

func (s *Scenario) NewNode(ctx context.Context, nodeNum int) (Node, error)

NewNode creates a node for the given node number.

func (*Scenario) NewStatsCollector 

func (s *Scenario) NewStatsCollector() *StatsCollector

func (*Scenario) PushEventsTo 

func (s *Scenario) PushEventsTo(published, received chan NodeEvent, bandwidth chan BandwidthEvent)

PushEventsTo registers channels to receive scenario events. Nil channels are ignored. Each call adds new channels; multiple consumers each get a copy of every event.

func (*Scenario) RunNode 

func (s *Scenario) RunNode(ctx context.Context, node Node, peers []int, publishWait time.Duration)

RunNode starts a node and handles publishing/receiving based on node number.

func (*Scenario) Start 

func (s *Scenario) Start()

Start begins the simulation.

type ScenarioStats 

type ScenarioStats struct {
	PublishedMessages map[broadcast.MessageID][]byte
	PublishedAt       map[broadcast.MessageID]time.Time

	ReceivedMessages map[int]map[broadcast.MessageID][]byte        // nodeNum → messageID → data
	ReceivedAt       map[int]map[broadcast.MessageID]time.Time     // nodeNum → messageID → time
	ReceivedLatency  map[int]map[broadcast.MessageID]time.Duration // nodeNum → messageID → latency from PublishedAt

	OriginBytesSent map[int]int // nodeNum → total bytes sent as origin
	RelayBytesSent  map[int]int // nodeNum → total bytes sent as relay

	TransportBytesSent     map[int]int // nodeNum → total bytes sent (transport level)
	TransportBytesReceived map[int]int // nodeNum → total bytes received (transport level)

	ChunksPerNode map[int]ChunkStats // nodeNum → chunk reception verdicts
}

ScenarioStats holds collected publish/receive data from a simulation run.

func RunSimnetScenario 

func RunSimnetScenario(ctx context.Context, s *Scenario, publishWait time.Duration) (ScenarioStats, error)

RunSimnetScenario runs a complete simnet scenario: creates nodes, wires them according to the topology, and returns collected stats.

type ShadowDriver 

type ShadowDriver struct {
	Strategy    StrategyFunc
	TraceWriter *TraceWriter
	// contains filtered or unexported fields
}

ShadowDriver implements Driver for Shadow simulation.

func (*ShadowDriver) Close 

func (s *ShadowDriver) Close() error

Close is a no-op for Shadow.

func (*ShadowDriver) NewNode 

func (s *ShadowDriver) NewNode(nodeNum int, logger *slog.Logger) (Node, error)

NewNode creates a node for Shadow simulation.

func (*ShadowDriver) NodeAddr 

func (s *ShadowDriver) NodeAddr(nodeNum int) net.Addr

NodeAddr resolves the address for a node in Shadow via DNS.

func (*ShadowDriver) Observer 

func (s *ShadowDriver) Observer() *Observer

Observer returns the observer for the node created by this driver.

func (*ShadowDriver) Start 

func (s *ShadowDriver) Start()

Start is a no-op for Shadow.

type SimnetDriver 

type SimnetDriver struct {
	Strategy    StrategyFunc
	Topology    Topology
	TraceWriter *TraceWriter
	// contains filtered or unexported fields
}

SimnetDriver implements Driver for simnet-based testing.

func (*SimnetDriver) BandwidthByRole 

func (s *SimnetDriver) BandwidthByRole() (origin, relay map[int]int)

BandwidthByRole returns per-node origin and relay byte counts from the Observers injected into each node.

func (*SimnetDriver) ChunkStatsByNode 

func (s *SimnetDriver) ChunkStatsByNode() map[int]ChunkStats

ChunkStatsByNode returns per-node chunk reception verdicts.

func (*SimnetDriver) Close 

func (s *SimnetDriver) Close() error

Close shuts down the simnet simulation.

func (*SimnetDriver) NewNode 

func (s *SimnetDriver) NewNode(nodeNum int, logger *slog.Logger) (Node, error)

NewNode creates a node using simnet for network simulation.

func (*SimnetDriver) NodeAddr 

func (s *SimnetDriver) NodeAddr(nodeNum int) net.Addr

NodeAddr returns the network address for a node.

func (*SimnetDriver) Start 

func (s *SimnetDriver) Start()

Start initializes the simnet simulation.

func (*SimnetDriver) TransportStats 

func (s *SimnetDriver) TransportStats() (sent, received map[int]int)

TransportStats returns per-node bytes sent and received at the transport (QUIC) level. Must be called before nodes are closed.

type SimulationConfig 

type SimulationConfig struct {
	Driver                  string `yaml:"driver"`
	LogLevel                string `yaml:"log_level"`
	LogFile                 string `yaml:"log_file,omitempty"`
	TraceFile               string `yaml:"trace_file,omitempty"`
	BandwidthLogFrequencyMs int    `yaml:"bandwidth_log_frequency_ms"`
}

SimulationConfig holds runtime/infrastructure settings.

func (*SimulationConfig) BandwidthLogFrequency 

func (c *SimulationConfig) BandwidthLogFrequency() time.Duration

type StatsCollector 

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

StatsCollector collects Published and Received events from a Scenario into a ScenarioStats. Call Run in a goroutine; it blocks until the context is cancelled and then returns the collected stats.

func (*StatsCollector) Run 

func (c *StatsCollector) Run(ctx context.Context) ScenarioStats

type StrategyConfig 

type StrategyConfig struct {
	Name string
	RS   *RSStrategyConfig
}

StrategyConfig uses custom UnmarshalYAML to dispatch by name into the correct typed config. Only the matching strategy pointer is non-nil.

func (*StrategyConfig) UnmarshalYAML 

func (sc *StrategyConfig) UnmarshalYAML(value *yaml.Node) error

type StrategyFunc 

type StrategyFunc func(nodeNum int, conn net.PacketConn, logger *slog.Logger, obs broadcast.Observer, tw *TraceWriter) (Node, error)

StrategyFunc creates a simulation node.

func ECStrategy 

func ECStrategy[CI broadcast.ChunkIdent, R broadcast.Wire, P broadcast.Wire](scheme broadcast.Scheme[CI, R, P]) StrategyFunc

ECStrategy returns a StrategyFunc that creates broadcast nodes using the given erasure coding scheme. Engine, channel, and subscription setup are handled here; the scheme is the only varying part.

func GossipsubStrategy 

func GossipsubStrategy() StrategyFunc

GossipsubStrategy returns a StrategyFunc that creates gossipsub nodes.

type Topology 

type Topology struct {
	Nodes []NodeSpec `json:"nodes" yaml:"nodes"`
	Edges []EdgeSpec `json:"edges" yaml:"edges"`
}

Topology describes the network topology for simulation.

func (Topology) EdgesMap 

func (t Topology) EdgesMap() map[int][]int

EdgesMap returns a map from node number to its outgoing peer node numbers.

type TopologyConfig 

type TopologyConfig struct {
	File     string            `yaml:"file,omitempty"`
	Generate *TopologyGenerate `yaml:"generate,omitempty"`
}

TopologyConfig supports two mutually exclusive modes. Go only uses File; Generate is for the Python CLI.

type TopologyGenerate 

type TopologyGenerate struct {
	NumNodes          int     `yaml:"num_nodes"`
	Degree            int     `yaml:"degree"`
	Type              string  `yaml:"type"`
	Seed              int     `yaml:"seed"`
	SuperNodeFraction float64 `yaml:"super_node_fraction"`
}

TopologyGenerate holds parameters for Python-side topology generation.

type TraceHeaderOptions 

type TraceHeaderOptions struct {
	DecoderName string
	PeerIDs     []string
}

type TraceWriter 

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

TraceWriter writes broadcast trace events as compact NDJSON. Safe for concurrent use from multiple goroutines.

func NewTraceWriter 

func NewTraceWriter(w io.Writer, t0 time.Time, nodes []string, topology Topology, config json.RawMessage) (*TraceWriter, error)

NewTraceWriter creates a TraceWriter and writes the header line.

func NewTraceWriterWithOptions 

func NewTraceWriterWithOptions(w io.Writer, t0 time.Time, nodes []string, topology Topology, config json.RawMessage, opts TraceHeaderOptions) (*TraceWriter, error)

NewTraceWriterWithOptions creates a TraceWriter and writes the header line.

func (*TraceWriter) Close 

func (tw *TraceWriter) Close() error

Close writes the footer and flushes.

func (*TraceWriter) WriteEvent 

func (tw *TraceWriter) WriteEvent(ts time.Time, node int, ev string, args ...any)

WriteEvent writes a single event as a JSON array tuple.

type TracingObserver 

type TracingObserver struct {
	*Observer
	// contains filtered or unexported fields
}

TracingObserver implements broadcast.Observer by writing compact event tuples to a shared TraceWriter while also tracking stats via an embedded Observer.

func NewTracingObserver 

func NewTracingObserver(nodeIdx int, tw *TraceWriter) *TracingObserver

func (*TracingObserver) OnChunkError 

func (o *TracingObserver) OnChunkError(err broadcast.ChunkProcessError)

func (*TracingObserver) OnChunkRcvd 

func (o *TracingObserver) OnChunkRcvd(peer broadcast.PeerID, channelID broadcast.ChannelID, messageID broadcast.MessageID, verdict broadcast.Verdict)

func (*TracingObserver) OnChunkSent 

func (o *TracingObserver) OnChunkSent(peer broadcast.PeerID, channelID broadcast.ChannelID, messageID broadcast.MessageID, bytesSent int)

func (*TracingObserver) OnPeerGone 

func (o *TracingObserver) OnPeerGone(peer broadcast.PeerID)

func (*TracingObserver) OnPeerHandshook 

func (o *TracingObserver) OnPeerHandshook(peer broadcast.PeerID, version broadcast.ProtocolVersion, channels []broadcast.ChannelID)

func (*TracingObserver) OnPeerSubscribed 

func (o *TracingObserver) OnPeerSubscribed(peer broadcast.PeerID, channelID broadcast.ChannelID)

func (*TracingObserver) OnPeerUnsubscribed 

func (o *TracingObserver) OnPeerUnsubscribed(peer broadcast.PeerID, channelID broadcast.ChannelID)

func (*TracingObserver) OnPreambleOpened 

func (o *TracingObserver) OnPreambleOpened(peer broadcast.PeerID, channelID broadcast.ChannelID, messageID broadcast.MessageID)

func (*TracingObserver) OnRoutingUpdate 

func (o *TracingObserver) OnRoutingUpdate(peer broadcast.PeerID, channelID broadcast.ChannelID, messageID broadcast.MessageID)

func (*TracingObserver) OnSessionDecoded 

func (o *TracingObserver) OnSessionDecoded(channelID broadcast.ChannelID, messageID broadcast.MessageID, latency time.Duration)

func (*TracingObserver) OnSessionDisposed 

func (o *TracingObserver) OnSessionDisposed(channelID broadcast.ChannelID, messageID broadcast.MessageID, reason string)

func (*TracingObserver) OnSessionStarted 

func (o *TracingObserver) OnSessionStarted(channelID broadcast.ChannelID, messageID broadcast.MessageID, role broadcast.SessionRole)

func (*TracingObserver) OnStrategyProgress 

func (o *TracingObserver) OnStrategyProgress(channelID broadcast.ChannelID, messageID broadcast.MessageID, chunksHave, chunksNeed int)

type WorkloadConfig 

type WorkloadConfig struct {
	NumMessages        int     `yaml:"num_messages"`
	MessageSize        int     `yaml:"message_size"`
	PublishWaitSeconds float64 `yaml:"publish_wait_seconds"`
	StopTimeMinutes    float64 `yaml:"stop_time_minutes"`
}

WorkloadConfig describes what to publish.

func (*WorkloadConfig) PublishWait 

func (c *WorkloadConfig) PublishWait() time.Duration

func (*WorkloadConfig) StopTime 

func (c *WorkloadConfig) StopTime() time.Duration

Source Files

View all Source files

Directories

Path Synopsis
cmd
shadow command

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.