rewrite of sei-load generator and sender.

README.md

@@ -30,10 +30,10 @@ Edit `my-config.json`:
     "newAccountRate": 0.1
   },
   "settings": {
-    "workers": 5,
     "tps": 100,
     "statsInterval": "10s",
     "bufferSize": 1000,
+    "connsPerEndpoint": 1,
     "trackUserLatency": true
   }
 }
@@ -50,10 +50,10 @@ Edit `my-config.json`:
 | Flag | Default | Description |
 |------|---------|-------------|
 | `--config, -c` | | Config file path (required) |
-| `--workers, -w` | 1 | Workers per endpoint |
 | `--tps, -t` | 0 | Transactions per second (0 = unlimited) |
 | `--stats-interval, -s` | 10s | Stats logging interval |
-| `--buffer-size, -b` | 1000 | Buffer size per worker |
+| `--buffer-size, -b` | 1000 | Sender queue size |
+| `--conns-per-endpoint` | 1 | RPC connections to open per endpoint |
 | `--dry-run` | false | Simulate without sending |
 | `--debug` | false | Log each transaction |
 | `--track-receipts` | false | Enable the block-indexed tx→inclusion tracker (stamps InclusionTime; reports included/expired/inflight-at-shutdown) |
@@ -66,12 +66,12 @@ Edit `my-config.json`:
 
 ### Basic Load Test
 ```bash
-./seiload --config my-config.json --workers 5 --tps 100
+./seiload --config my-config.json --tps 100
 ```
 
 ### High Throughput Test
 ```bash
-./seiload --config my-config.json --workers 20 --buffer-size 2000
+./seiload --config my-config.json --buffer-size 2000 --max-in-flight 20000 --conns-per-endpoint 4
 ```
 
 ### Debug Mode
@@ -122,21 +122,21 @@ Edit `my-config.json`:
 ### Settings
 ```json
 "settings": {
-  "workers": 5,
   "tps": 100,
   "statsInterval": "10s",
   "bufferSize": 1000,
+  "connsPerEndpoint": 1,
   "trackUserLatency": true
 }
 ```
 
 **Settings Precedence**: CLI flags > Config file settings > Default values
 
 Available settings:
-- `workers`: Number of workers per endpoint
 - `tps`: Transactions per second (0 = unlimited)
 - `statsInterval`: Stats logging interval (e.g., "10s", "5m")
-- `bufferSize`: Buffer size per worker
+- `bufferSize`: Sender queue size
+- `connsPerEndpoint`: RPC connections to open per endpoint
 - `dryRun`: Simulate without sending transactions
 - `debug`: Enable debug logging
 - `trackReceipts`: Track transaction receipts
@@ -225,10 +225,10 @@ make clean
 - Try `--dry-run` to test config
 
 ### Low Performance
-- Increase `--workers`
 - Increase `--buffer-size`
+- Increase `--max-in-flight` for open-loop runs
 
 ### Memory Issues
 - Reduce `--buffer-size`
-- Reduce worker count
+- Reduce `--max-in-flight` for open-loop runs
 - Disable receipt tracking

config/config.go

@@ -3,7 +3,6 @@ package config
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/sei-protocol/sei-load/utils"
 	"math/big"
 	"time"
 )
@@ -12,19 +11,12 @@ import (
 type LoadConfig struct {
 	ChainID int64 `json:"chainId,omitempty"`
 	// SeiChainID is the textual chain ID used for tagging metric collection.
-	SeiChainID string   `json:"seiChainID,omitempty"`
-	Endpoints  []string `json:"endpoints"`
-	// Number of shards to divide the senders into.
-	// Txs within each shard are sent sequentially.
-	// Defaults to Endpoints * Settings.TasksPerEndpoint.
-	// WARNING: this is unrelated to the server-side autobahn sharding
-	// (which assigns tx sender addrs to lanes). It is solely used to maximize
-	// txs/s throughput of the load generator.
-	NumShards  utils.Option[int] `json:"numShards,omitzero"`
-	Accounts   *AccountConfig    `json:"accounts,omitempty"`
-	Scenarios  []Scenario        `json:"scenarios,omitempty"`
-	MockDeploy bool              `json:"mockDeploy,omitempty"`
-	Settings   *Settings         `json:"settings,omitempty"`
+	SeiChainID string         `json:"seiChainID,omitempty"`
+	Endpoints  []string       `json:"endpoints"`
+	Accounts   *AccountConfig `json:"accounts,omitempty"`
+	Scenarios  []Scenario     `json:"scenarios,omitempty"`
+	MockDeploy bool           `json:"mockDeploy,omitempty"`
+	Settings   *Settings      `json:"settings,omitempty"`
 	// Funding, when set, funds the generated account pool from a root key at
 	// startup so the run works against a real chain. See funding.go.
 	Funding *FundingConfig `json:"funding,omitempty"`
@@ -33,18 +25,12 @@ type LoadConfig struct {
 	// Seed roots the deterministic PRNG sub-streams that drive the run. Same
 	// seed + config reproduces the per-stream draw multiset, so the workload
 	// (the distribution of keys, sizes, gas, and accounts) is statistically
-	// reproducible for fair A/B comparison. Per-tx emission ordering is
-	// reproducible only at a single worker; above one worker the multiset still
-	// matches but ordering does not, and on-chain arrival order is concurrent
+	// reproducible for fair A/B comparison. On-chain arrival order is concurrent
 	// regardless. A nil Seed means "unseeded": the generator resolves a random
 	// one and records it for after-the-fact replay.
 	Seed *uint64 `json:"seed,omitempty"`
 }
 
-func (c *LoadConfig) GetNumShards() int {
-	return c.NumShards.Or(len(c.Endpoints) * c.Settings.TasksPerEndpoint)
-}
-
 func (c *LoadConfig) TotalQueueSize() int {
 	// Backward compatible formula, consider making it a config value.
 	return len(c.Endpoints) * c.Settings.BufferSize

config/distribution.go

@@ -4,10 +4,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"math"
-	"math/rand/v2"
+	mrand "math/rand/v2"
 	"sync"
-
-	"github.com/sei-protocol/sei-load/utils/rng"
 )
 
 var (
@@ -18,7 +16,7 @@ var (
 
 // indexSampler draws an index in [0, n) from some keyspace distribution.
 type indexSampler interface {
-	SampleIndex(n uint64) (uint64, error)
+	SampleIndex(rng *mrand.Rand, n uint64) (uint64, error)
 }
 
 // Distribution is a tagged keyspace index sampler selected by a "Name"
@@ -31,25 +29,13 @@ type Distribution struct {
 
 func (d *Distribution) Name() string { return d.name }
 
-// SetStream binds the sampler to a deterministic sub-stream (nil = unseeded
-// global RNG); a zero-value Distribution draws nothing, so it no-ops. See
-// package doc for the reproducibility contract.
-func (d *Distribution) SetStream(s *rng.Stream) {
-	switch delegate := d.delegate.(type) {
-	case *UniformDistribution:
-		delegate.stream = s
-	case *ZipfianDistribution:
-		delegate.stream = s
-	}
-}
-
 // SampleIndex delegates to the selected sampler; a zero-value (no Name)
 // Distribution returns 0.
-func (d *Distribution) SampleIndex(n uint64) (uint64, error) {
+func (d *Distribution) SampleIndex(rng *mrand.Rand, n uint64) (uint64, error) {
 	if d.delegate == nil {
 		return 0, nil
 	}
-	return d.delegate.SampleIndex(n)
+	return d.delegate.SampleIndex(rng, n)
 }
 
 func (d *Distribution) UnmarshalJSON(data []byte) error {
@@ -64,7 +50,7 @@ func (d *Distribution) UnmarshalJSON(data []byte) error {
 	case "":
 		return nil
 	case "uniform":
-		// No JSON parameters; the stream is bound later via SetStream.
+		// No JSON parameters; the PRNG is supplied at draw time.
 		d.delegate = &UniformDistribution{}
 		return nil
 	case "zipfian":
@@ -83,20 +69,13 @@ func (d *Distribution) UnmarshalJSON(data []byte) error {
 }
 
 // UniformDistribution draws each index with equal probability.
-//
-// copy-safe: holds no mutex; the *rng.Stream pointer aliases on copy.
-type UniformDistribution struct {
-	stream *rng.Stream
-}
+type UniformDistribution struct{}
 
-func (u *UniformDistribution) SampleIndex(n uint64) (uint64, error) {
+func (u *UniformDistribution) SampleIndex(rng *mrand.Rand, n uint64) (uint64, error) {
 	if n == 0 {
 		return 0, fmt.Errorf("uniform sample: empty keyspace (n == 0)")
 	}
-	if u.stream != nil {
-		return u.stream.Uint64N(n), nil
-	}
-	return rand.Uint64N(n), nil
+	return rng.Uint64N(n), nil
 }
 
 // ZipfianDistribution is the YCSB precomputed-zeta generator: zeta(n, theta) is
@@ -107,8 +86,6 @@ func (u *UniformDistribution) SampleIndex(n uint64) (uint64, error) {
 type ZipfianDistribution struct {
 	Theta float64 `json:"theta"`
 
-	stream *rng.Stream
-
 	mu    sync.Mutex
 	state *zipfState // memoized for state.n; recomputed when n changes.
 }
@@ -169,7 +146,7 @@ func (z *ZipfianDistribution) validate() error {
 // SampleIndex draws a Zipf-skewed index in [0, n). n must be stable per sampler:
 // the zeta cache is keyed on n, so a changing n recomputes O(n) every draw. See
 // package doc.
-func (z *ZipfianDistribution) SampleIndex(n uint64) (uint64, error) {
+func (z *ZipfianDistribution) SampleIndex(rng *mrand.Rand, n uint64) (uint64, error) {
 	if n == 0 {
 		return 0, fmt.Errorf("zipfian sample: empty keyspace (n == 0)")
 	}
@@ -181,12 +158,7 @@ func (z *ZipfianDistribution) SampleIndex(n uint64) (uint64, error) {
 	st := z.state
 	z.mu.Unlock()
 
-	var u float64
-	if z.stream != nil {
-		u = z.stream.Float64()
-	} else {
-		u = rand.Float64()
-	}
+	u := rng.Float64()
 	uz := u * st.zetaN
 	if uz < 1.0 {
 		return 0, nil