diff --git a/cmd/prometheus/dashboards/erigon_internals.json b/cmd/prometheus/dashboards/erigon_internals.json index e7459459dd0..d176e6d52ad 100644 --- a/cmd/prometheus/dashboards/erigon_internals.json +++ b/cmd/prometheus/dashboards/erigon_internals.json @@ -5144,7 +5144,7 @@ "targets": [ { "editorMode": "code", - "expr": "commit_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_keys_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "total keys", "range": true, "refId": "A", @@ -5155,7 +5155,7 @@ }, { "editorMode": "code", - "expr": "commit_account_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_key_traversals_total{kind=\"address\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "account keys", "range": true, "refId": "B", @@ -5166,7 +5166,7 @@ }, { "editorMode": "code", - "expr": "commit_storage_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_key_traversals_total{kind=\"storage\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "storage keys", "range": true, "refId": "C", @@ -5265,7 +5265,7 @@ "targets": [ { "editorMode": "code", - "expr": "commit_fold_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_folds_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "fold", "range": true, "refId": "A", @@ -5276,7 +5276,7 @@ }, { "editorMode": "code", - "expr": "commit_unfold_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_unfolds_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "unfold", "range": true, "refId": "B", @@ -5375,7 +5375,7 @@ "targets": [ { "editorMode": "code", - "expr": "commit_branch_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"branch\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "branch reads", "range": true, "refId": "A", @@ -5386,7 +5386,7 @@ }, { "editorMode": "code", - "expr": "commit_branch_write_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_branch_writes_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "branch writes", "range": true, "refId": "B", @@ -5397,7 +5397,7 @@ }, { "editorMode": "code", - "expr": "commit_read_rate{instance=~\"$instance\"}", + "expr": "sum without (kind) (rate(commitment_reads_total{instance=~\"$instance\"}[$__rate_interval]))", "legendFormat": "total reads", "range": true, "refId": "C", @@ -5408,7 +5408,7 @@ }, { "editorMode": "code", - "expr": "commit_account_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"account\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "account reads", "range": true, "refId": "D", @@ -5419,7 +5419,7 @@ }, { "editorMode": "code", - "expr": "commit_storage_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"storage\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "storage reads", "range": true, "refId": "E", @@ -5535,8 +5535,8 @@ "targets": [ { "editorMode": "code", - "expr": "commit_block_dur{instance=~\"$instance\"}", - "legendFormat": "block dur (ns)", + "expr": "1e9 * histogram_quantile(0.9, sum by (le, instance) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", + "legendFormat": "block dur (ns) {{instance}}", "range": true, "refId": "A", "datasource": { diff --git a/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json b/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json index 184cc4e86b6..1025f8b9b21 100644 --- a/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json +++ b/dashboards/erigon_custom_metrics/erigon_custom_metrics.internal.json @@ -5997,7 +5997,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "builder", - "expr": "commit_block_dur{instance=\"$instance\"}", + "expr": "1e9 * histogram_quantile(0.9, sum by (le, instance) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", "instant": false, "legendFormat": "commitment: {{instance}}", "range": true, @@ -9290,7 +9290,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_keys_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "total keys {{instance}}", "range": true, "refId": "A" @@ -9301,7 +9301,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_account_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_key_traversals_total{kind=\"address\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "account keys {{instance}}", "range": true, "refId": "B" @@ -9312,7 +9312,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_storage_key_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_key_traversals_total{kind=\"storage\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "storage keys {{instance}}", "range": true, "refId": "C" @@ -9417,7 +9417,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_fold_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_folds_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "fold {{instance}}", "range": true, "refId": "A" @@ -9428,7 +9428,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_unfold_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_unfolds_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "unfold {{instance}}", "range": true, "refId": "B" @@ -9533,7 +9533,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_branch_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"branch\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "reads {{instance}}", "range": true, "refId": "A" @@ -9544,7 +9544,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_branch_write_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_branch_writes_total{instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "writes {{instance}}", "range": true, "refId": "B" @@ -9555,7 +9555,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_read_rate{instance=~\"$instance\"}", + "expr": "sum without (kind) (rate(commitment_reads_total{instance=~\"$instance\"}[$__rate_interval]))", "legendFormat": "total reads {{instance}}", "range": true, "refId": "C" @@ -9566,7 +9566,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_account_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"account\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "account reads {{instance}}", "range": true, "refId": "D" @@ -9577,7 +9577,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_storage_read_rate{instance=~\"$instance\"}", + "expr": "rate(commitment_reads_total{kind=\"storage\",instance=~\"$instance\"}[$__rate_interval])", "legendFormat": "storage reads {{instance}}", "range": true, "refId": "E" @@ -9699,7 +9699,7 @@ "uid": "grafanacloud-prom" }, "editorMode": "code", - "expr": "commit_block_dur{instance=~\"$instance\"}", + "expr": "1e9 * histogram_quantile(0.9, sum by (le, instance) (rate(commitment_round_duration_seconds_bucket{instance=~\"$instance\"}[$__rate_interval])))", "legendFormat": "block dur (ns) {{instance}}", "range": true, "refId": "A" diff --git a/db/state/execctx/domain_shared.go b/db/state/execctx/domain_shared.go index dcc7fecdf58..4702a156a63 100644 --- a/db/state/execctx/domain_shared.go +++ b/db/state/execctx/domain_shared.go @@ -539,7 +539,7 @@ func (sd *SharedDomains) FlushPendingUpdatesWithoutChangeset(tx kv.TemporalTx) e putBranch := func(prefix, data, prevData []byte) error { return sd.DomainPutCommitmentDiff(tx, prefix, data, upd.TxNum, prevData, nil) } - _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch) + _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch, upd.Metrics) return err } @@ -561,7 +561,7 @@ func (sd *SharedDomains) flushPendingUpdates(ctx context.Context, tx kv.Temporal switcher, ok := sd.mem.(changesetSwitcher) if !ok { - _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch) + _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch, upd.Metrics) return err } @@ -585,7 +585,7 @@ func (sd *SharedDomains) flushPendingUpdates(ctx context.Context, tx kv.Temporal // see concurrency contract on the wrappers above. defer sd.SwapCommitmentDiffLocked(cs)() - if _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch); err != nil { + if _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch, upd.Metrics); err != nil { return err } @@ -594,7 +594,7 @@ func (sd *SharedDomains) flushPendingUpdates(ctx context.Context, tx kv.Temporal } // No past changeset found — write into whatever is current. - _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch) + _, err := commitment.ApplyDeferredBranchUpdates(upd.Deferred, runtime.NumCPU(), putBranch, upd.Metrics) return err } diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go index 9a706ca3cca..9f7a1972819 100644 --- a/execution/commitment/commitment.go +++ b/execution/commitment/commitment.go @@ -281,6 +281,10 @@ type PendingCommitmentUpdate struct { BlockHash common.Hash TxNum uint64 Deferred []*DeferredBranchUpdate + // Metrics is the producing trie's, carried so the later apply still reaches + // that trie's log and CSV counters. The Prometheus counters do not depend on + // it — publishBranchWrites bills those where the write lands. + Metrics *Metrics } func (p *PendingCommitmentUpdate) Clear() { @@ -369,26 +373,29 @@ func (be *BranchEncoder) ApplyDeferredUpdates( numWorkers int, putBranch func(prefix []byte, data []byte, prevData []byte) error, ) error { - written, err := ApplyDeferredBranchUpdates(be.deferred, numWorkers, putBranch) - if err != nil { + if _, err := ApplyDeferredBranchUpdates(be.deferred, numWorkers, putBranch, be.metrics); err != nil { return err } - if be.metrics != nil { - be.metrics.updateBranch.Add(uint64(written)) - } return nil } var workerMergerPool = sync.Pool{New: func() any { return NewHexBranchMerger(512) }} -// Returns the number of updates written. putBranch must copy prefix and data rather than -// retain them: they are pooled and reused for a later, unrelated update. prevData is -// cloned per update and carries no such constraint. +// ApplyDeferredBranchUpdates applies the queued branch writes and returns how many +// were written. Writes are published to the branch-write counters as they land, +// not against a round: the caller-owned path applies from SharedDomains after the +// producing round has already closed, so there is no round left to bill. m, when +// non-nil, additionally carries them into that trie's log and CSV counters. +// +// putBranch must copy prefix and data rather than retain them: they are pooled and +// reused for a later, unrelated update. prevData is cloned per update and carries +// no such constraint. func ApplyDeferredBranchUpdates( deferred []*DeferredBranchUpdate, numWorkers int, putBranch func(prefix []byte, data []byte, prevData []byte) error, -) (int, error) { + m *Metrics, +) (n int, err error) { if len(deferred) == 0 { return 0, nil } @@ -400,20 +407,24 @@ func ApplyDeferredBranchUpdates( merger := workerMergerPool.Get().(*BranchMerger) defer workerMergerPool.Put(merger) - var written int + var written, bytesOut int for _, upd := range deferred { if err := mergeDeferredUpdate(upd, merger); err != nil { + publishBranchWrites(written, bytesOut, m) return written, err } if upd.encoded == nil { continue } if err := putBranch(capLen(upd.prefix), capLen(upd.encoded), capLen(upd.prev)); err != nil { + publishBranchWrites(written, bytesOut, m) return written, err } written++ + bytesOut += len(upd.encoded) } mxTrieBranchesUpdated.AddInt(written) + publishBranchWrites(written, bytesOut, m) return written, nil } @@ -445,17 +456,20 @@ func ApplyDeferredBranchUpdates( } } - var written int + var written, bytesOut int for _, upd := range deferred { if upd.encoded == nil { continue } if err := putBranch(capLen(upd.prefix), capLen(upd.encoded), capLen(upd.prev)); err != nil { + publishBranchWrites(written, bytesOut, m) return written, err } written++ + bytesOut += len(upd.encoded) } mxTrieBranchesUpdated.AddInt(written) + publishBranchWrites(written, bytesOut, m) return written, nil } @@ -503,9 +517,7 @@ func (be *BranchEncoder) CollectUpdate( if err := ctx.PutBranch(prefixCopy, updateCopy, prev); err != nil { return err } - if be.metrics != nil { - be.metrics.updateBranch.Add(1) - } + publishBranchWrites(1, len(updateCopy), be.metrics) mxTrieBranchesUpdated.Inc() return nil } diff --git a/execution/commitment/commitment_test.go b/execution/commitment/commitment_test.go index fbfb32aba6a..2294171ef11 100644 --- a/execution/commitment/commitment_test.go +++ b/execution/commitment/commitment_test.go @@ -1115,7 +1115,7 @@ func TestApplyDeferred_CallbackSeesInputDerivedCapacity(t *testing.T) { require.Equal(t, len(data), cap(data), "data carries leftover pool capacity") require.Equal(t, len(prevData), cap(prevData), "prevData carries leftover pool capacity") return nil - }) + }, nil) require.NoError(t, err) require.Equal(t, tc.updates, written) require.Equal(t, tc.updates, seen, "callback must run for every update") diff --git a/execution/commitment/commitmentdb/commitment_context.go b/execution/commitment/commitmentdb/commitment_context.go index 431768e17ef..d0d8bc9ec2a 100644 --- a/execution/commitment/commitmentdb/commitment_context.go +++ b/execution/commitment/commitmentdb/commitment_context.go @@ -666,6 +666,7 @@ func (sdc *SharedDomainsCommitmentContext) computeCommitment(ctx context.Context BlockNum: blockNum, TxNum: txNum, Deferred: trie.TakeDeferredUpdates(), + Metrics: trie.Metrics(), } } case *commitment.ParallelPatriciaHashed: @@ -674,6 +675,7 @@ func (sdc *SharedDomainsCommitmentContext) computeCommitment(ctx context.Context BlockNum: blockNum, TxNum: txNum, Deferred: trie.TakeDeferredUpdates(), + Metrics: trie.Metrics(), } } } diff --git a/execution/commitment/hex_patricia_hashed.go b/execution/commitment/hex_patricia_hashed.go index 8417c960edf..8aceffb6817 100644 --- a/execution/commitment/hex_patricia_hashed.go +++ b/execution/commitment/hex_patricia_hashed.go @@ -233,6 +233,10 @@ func newHexPatriciaHashed() *HexPatriciaHashed { return hph } +// Metrics exposes the trie's counters so a caller applying its deferred writes +// can carry them into this trie's log and CSV totals. +func (hph *HexPatriciaHashed) Metrics() *Metrics { return hph.metrics } + // SetCollapseTracer sets a callback that will be invoked when a node collapse occurs // during commitment calculation. This is used by witness generation to capture paths // to HashNodes that need resolution when a FullNode is reduced to a single child. @@ -1464,6 +1468,7 @@ func (hph *HexPatriciaHashed) unfoldBranchNode(row int, depth int16, deleted boo if err != nil { return err } + hph.metrics.AddBranchRead(len(branchData)) // depthsToTxNum is used for per-file metrics; step is no longer available // from the cache-or-DB helper (cache never had a meaningful step anyway). @@ -2483,6 +2488,9 @@ func (hph *HexPatriciaHashed) Process(ctx context.Context, updates *Updates, log hph.metrics.Reset() hph.metrics.updates.Store(updatesCount) + hph.metrics.AddRoundKeys(updatesCount) + roundStart := time.Now() + defer func() { observeRound(hph.metrics, roundStart) }() if hph.metrics.collectCommitmentMetrics { defer func() { hph.metrics.TotalProcessingTimeInc(start) diff --git a/execution/commitment/metrics.go b/execution/commitment/metrics.go index e2c92defb99..c18223272cd 100644 --- a/execution/commitment/metrics.go +++ b/execution/commitment/metrics.go @@ -42,6 +42,9 @@ type Metrics struct { loadDepths [10]uint64 unfolds atomic.Uint64 folds atomic.Uint64 + roundKeys atomic.Uint64 + branchReadBytes atomic.Uint64 + branchWriteBytes atomic.Uint64 spentUnfolding atomic.Int64 spentFolding atomic.Int64 spentProcessing atomic.Int64 @@ -51,28 +54,35 @@ type Metrics struct { } type MetricValues struct { - mu *sync.RWMutex - Accounts map[string]*AccountStats - Branches map[string]*BranchStats - Updates uint64 - AddressKeys uint64 - StorageKeys uint64 - LoadBranch uint64 - LoadAccount uint64 - LoadStorage uint64 - CacheBranch uint64 - CacheAccount uint64 - CacheStorage uint64 - MissBranch uint64 - MissAccount uint64 - MissStorage uint64 - UpdateBranch uint64 - LoadDepths [10]uint64 - Unfolds uint64 - Folds uint64 - SpentUnfolding time.Duration - SpentFolding time.Duration - SpentProcessing time.Duration + mu *sync.RWMutex + Accounts map[string]*AccountStats + Branches map[string]*BranchStats + Updates uint64 + AddressKeys uint64 + StorageKeys uint64 + LoadBranch uint64 + LoadAccount uint64 + LoadStorage uint64 + CacheBranch uint64 + CacheAccount uint64 + CacheStorage uint64 + MissBranch uint64 + MissAccount uint64 + MissStorage uint64 + UpdateBranch uint64 + LoadDepths [10]uint64 + Unfolds uint64 + Folds uint64 + // RoundKeys is the distinct key count of one round — Process resets the + // counters on both engines, so nothing here spans rounds. AddressKeys and + // StorageKeys count cell traversals instead, which the parallel engine + // inflates by re-walking subtrees on mount+replay. + RoundKeys uint64 + BranchReadBytes uint64 + BranchWriteBytes uint64 + SpentUnfolding time.Duration + SpentFolding time.Duration + SpentProcessing time.Duration } func (m MetricValues) RLock() { @@ -126,28 +136,31 @@ func (m *Metrics) EnableCsvMetrics(filePathPrefix string) { func (m *Metrics) AsValues() MetricValues { return MetricValues{ - mu: &m.Accounts.m, - Accounts: m.Accounts.AccountStats, - Branches: m.Branches.BranchStats, - Updates: m.updates.Load(), - AddressKeys: m.addressKeys.Load(), - StorageKeys: m.storageKeys.Load(), - LoadBranch: m.loadBranch.Load(), - LoadAccount: m.loadAccount.Load(), - LoadStorage: m.loadStorage.Load(), - CacheBranch: m.cacheBranch.Load(), - CacheAccount: m.cacheAccount.Load(), - CacheStorage: m.cacheStorage.Load(), - MissBranch: m.missBranch.Load(), - MissAccount: m.missAccount.Load(), - MissStorage: m.missStorage.Load(), - UpdateBranch: m.updateBranch.Load(), - LoadDepths: m.loadDepths, - Unfolds: m.unfolds.Load(), - Folds: m.folds.Load(), - SpentUnfolding: time.Duration(m.spentUnfolding.Load()), - SpentFolding: time.Duration(m.spentFolding.Load()), - SpentProcessing: time.Duration(m.spentProcessing.Load()), + mu: &m.Accounts.m, + Accounts: m.Accounts.AccountStats, + Branches: m.Branches.BranchStats, + Updates: m.updates.Load(), + AddressKeys: m.addressKeys.Load(), + StorageKeys: m.storageKeys.Load(), + LoadBranch: m.loadBranch.Load(), + LoadAccount: m.loadAccount.Load(), + LoadStorage: m.loadStorage.Load(), + CacheBranch: m.cacheBranch.Load(), + CacheAccount: m.cacheAccount.Load(), + CacheStorage: m.cacheStorage.Load(), + MissBranch: m.missBranch.Load(), + MissAccount: m.missAccount.Load(), + MissStorage: m.missStorage.Load(), + UpdateBranch: m.updateBranch.Load(), + LoadDepths: m.loadDepths, + Unfolds: m.unfolds.Load(), + Folds: m.folds.Load(), + RoundKeys: m.roundKeys.Load(), + BranchReadBytes: m.branchReadBytes.Load(), + BranchWriteBytes: m.branchWriteBytes.Load(), + SpentUnfolding: time.Duration(m.spentUnfolding.Load()), + SpentFolding: time.Duration(m.spentFolding.Load()), + SpentProcessing: time.Duration(m.spentProcessing.Load()), } } @@ -314,6 +327,9 @@ func (m *Metrics) Reset() { m.cacheStorage.Store(0) m.unfolds.Store(0) m.folds.Store(0) + m.roundKeys.Store(0) + m.branchReadBytes.Store(0) + m.branchWriteBytes.Store(0) m.spentUnfolding.Store(0) m.spentFolding.Store(0) m.spentProcessing.Store(0) @@ -416,6 +432,17 @@ func (m *Metrics) StartFolding(plainKey []byte) func() { return nil } +// AddBranchRead records one branch read of n bytes. +func (m *Metrics) AddBranchRead(n int) { m.branchReadBytes.Add(uint64(n)) } + +// AddBranchWrite records one branch write of n bytes. +func (m *Metrics) AddBranchWrite(n int) { m.branchWriteBytes.Add(uint64(n)) } + +// AddRoundKeys records the distinct-key count of one finished round. Not +// merged between tries: the engine that ran the round owns this number, while +// each worker only sees its own subtree. +func (m *Metrics) AddRoundKeys(n uint64) { m.roundKeys.Add(n) } + // Merge folds src's counters into m. The parallel trie gives every mount // worker its own Metrics — an atomic add on a shared line in the fold loop // would cost more than the counter is worth — and merges once per round. @@ -437,6 +464,8 @@ func (m *Metrics) Merge(src *Metrics) { m.updateBranch.Add(src.updateBranch.Load()) m.unfolds.Add(src.unfolds.Load()) m.folds.Add(src.folds.Load()) + m.branchReadBytes.Add(src.branchReadBytes.Load()) + m.branchWriteBytes.Add(src.branchWriteBytes.Load()) m.spentUnfolding.Add(src.spentUnfolding.Load()) m.spentFolding.Add(src.spentFolding.Load()) m.spentProcessing.Add(src.spentProcessing.Load()) diff --git a/execution/commitment/parallel_metrics_test.go b/execution/commitment/parallel_metrics_test.go index 02ae7e618d6..07fe71b722e 100644 --- a/execution/commitment/parallel_metrics_test.go +++ b/execution/commitment/parallel_metrics_test.go @@ -104,3 +104,98 @@ func TestMetricsResetClearsEveryCounter(t *testing.T) { assert.Zero(t, v.Unfolds) assert.Zero(t, v.SpentFolding) } + +func TestRoundKeysAreDistinctNotTraversals(t *testing.T) { + ms := NewMockState(t) + keys, upds := buildNibbleSpread(t, 16, 4) + require.NoError(t, ms.applyPlainUpdates(keys, upds)) + + tr := newParTrie(t, ms, 4) + defer tr.Release() + ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash) + defer ut.Close() + for _, k := range keys { + ut.TouchPlainKey(string(k), nil, nil) + } + + var got *CommitProgress + _, err := tr.Process(context.Background(), ut, "", func(p *CommitProgress) { got = p }, WarmupConfig{}) + require.NoError(t, err) + require.NotNil(t, got) + + m := got.Metrics + assert.EqualValues(t, len(keys), m.RoundKeys, + "RoundKeys is the distinct key count handed to the trie") + assert.GreaterOrEqual(t, m.AddressKeys+m.StorageKeys, m.RoundKeys, + "traversals count cell visits, so they never undercount distinct keys") + assert.Positive(t, m.BranchWriteBytes, "branch write bytes are counted") +} + +// Two rounds on one trie must report the second round's own numbers. Tries come +// from a pool whose Release does not clear counters, and the parallel trie used +// to accumulate across rounds, so both ends of the merge could carry history in. +func TestRoundCountersDoNotAccumulateAcrossRounds(t *testing.T) { + ms := NewMockState(t) + keys, upds := buildNibbleSpread(t, 16, 4) + require.NoError(t, ms.applyPlainUpdates(keys, upds)) + + tr := newParTrie(t, ms, 4) + defer tr.Release() + + round := func() *CommitProgress { + t.Helper() + ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash) + defer ut.Close() + for _, k := range keys { + ut.TouchPlainKey(string(k), nil, nil) + } + var got *CommitProgress + _, err := tr.Process(context.Background(), ut, "", func(p *CommitProgress) { got = p }, WarmupConfig{}) + require.NoError(t, err) + require.NotNil(t, got) + return got + } + + first := round() + second := round() + + assert.EqualValues(t, len(keys), first.Metrics.RoundKeys) + assert.EqualValues(t, len(keys), second.Metrics.RoundKeys, + "the second round reports its own key count, not the running total") + // Strictly less than 2x: with the pooled reset removed a worker enters the + // second round holding the first's traversals and adds its own, landing on + // exactly 2x — which an inclusive bound would admit. + assert.Less(t, second.Metrics.AddressKeys, first.Metrics.AddressKeys*2, + "traversals are per-round; a pooled worker must not carry its last round in") +} + +// A branch write must reach the Prometheus counter exactly once. Billing it both +// where it lands and again from the round's snapshot is invisible in the trie's +// own MetricValues, so this reads the published counter instead. +func TestBranchWritesArePublishedOnce(t *testing.T) { + ms := NewMockState(t) + keys, upds := buildNibbleSpread(t, 16, 4) + require.NoError(t, ms.applyPlainUpdates(keys, upds)) + + tr := newParTrie(t, ms, 4) + defer tr.Release() + ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash) + defer ut.Close() + for _, k := range keys { + ut.TouchPlainKey(string(k), nil, nil) + } + + beforePuts := mxBranchPuts.GetValueUint64() + beforeBytes := mxWriteBytes.GetValueUint64() + + var got *CommitProgress + _, err := tr.Process(context.Background(), ut, "", func(p *CommitProgress) { got = p }, WarmupConfig{}) + require.NoError(t, err) + require.NotNil(t, got) + + require.Positive(t, got.Metrics.UpdateBranch, "the round wrote branches at all") + assert.EqualValues(t, got.Metrics.UpdateBranch, mxBranchPuts.GetValueUint64()-beforePuts, + "commitment_branch_writes_total counts each write once") + assert.EqualValues(t, got.Metrics.BranchWriteBytes, mxWriteBytes.GetValueUint64()-beforeBytes, + "commitment_branch_write_bytes_total counts each write once") +} diff --git a/execution/commitment/parallel_mount.go b/execution/commitment/parallel_mount.go index f4f1c0c9978..6e6505ca73a 100644 --- a/execution/commitment/parallel_mount.go +++ b/execution/commitment/parallel_mount.go @@ -142,7 +142,15 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up ni, ch := nib, child g.Go(func() error { w := NewHexPatriciaHashed(p.accountKeyLen, nil, p.cfg) - defer p.metrics.Merge(w.metrics) + // Tries come from a pool and Release does not clear their counters, + // so a checkout carries the previous round's numbers into the merge. + w.metrics.Reset() + // Merge before releasing: Release pools the trie, after which another + // goroutine may check it out and write these same counters. + release := func() { + p.metrics.Merge(w.metrics) + w.Release() + } w.mountTo(base, ni) if p.template != nil && p.template.traceW != nil { w.traceW = tracePrefix(p.template.traceW, fmt.Sprintf("[mnt %x] ", ni)) @@ -173,7 +181,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up return sr, err }) if buildErr != nil { - w.Release() + release() return fmt.Errorf("mount[%x] build: %w", ni, buildErr) } var tf time.Time @@ -186,7 +194,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up foldDur[ni] = time.Since(tf) } if err != nil { - w.Release() + release() return fmt.Errorf("mount[%x] fold: %w", ni, err) } cells[ni] = c @@ -194,7 +202,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up if deferred := w.TakeDeferredUpdates(); len(deferred) > 0 { pu.appendDeferred(deferred) } - w.Release() + release() return nil }) childIdx++ diff --git a/execution/commitment/parallel_patricia_hashed.go b/execution/commitment/parallel_patricia_hashed.go index b3cc5bdef16..238ce1d6e52 100644 --- a/execution/commitment/parallel_patricia_hashed.go +++ b/execution/commitment/parallel_patricia_hashed.go @@ -25,6 +25,7 @@ import ( "runtime" "sync" "sync/atomic" + "time" ) var defaultParallelCommitmentWorkers = runtime.NumCPU() @@ -52,6 +53,9 @@ type ParallelPatriciaHashed struct { func (p *ParallelPatriciaHashed) DeepLocalFolds() uint64 { return p.deepLocalFolds.Load() } +// Metrics exposes the round's counters; see HexPatriciaHashed.Metrics. +func (p *ParallelPatriciaHashed) Metrics() *Metrics { return p.metrics } + func NewParallelPatriciaHashed(ctxFactory TrieContextFactory, accountKeyLen int16, cfg TrieConfig) *ParallelPatriciaHashed { p := &ParallelPatriciaHashed{ template: NewHexPatriciaHashed(accountKeyLen, nil, cfg), @@ -231,6 +235,13 @@ func (p *ParallelPatriciaHashed) Process( p.rootHash.Store(nil) p.deepLocalFolds.Store(0) + // Per-round, matching HexPatriciaHashed.Process: the counters published for + // a round have to describe that round alone. + p.metrics.Reset() + p.metrics.AddRoundKeys(updates.Size()) + roundStart := time.Now() + defer func() { observeRound(p.metrics, roundStart) }() + pu := updates.parallel if pu.trie == nil || pu.trie.root == nil || pu.trie.root.subtreeCount == 0 { // A consumed (or never-touched) collection must return the carried root; folding @@ -329,12 +340,8 @@ func (p *ParallelPatriciaHashed) applyDeferredUpdates(ctx context.Context, pu *p // This path calls PutBranch directly rather than through a BranchEncoder, // so it is the only place the parallel engine's branch writes get counted. - n, err := ApplyDeferredBranchUpdates(deferred, p.numWorkers, applyCtx.PutBranch) - if err != nil { + if _, err := ApplyDeferredBranchUpdates(deferred, p.numWorkers, applyCtx.PutBranch, p.metrics); err != nil { return fmt.Errorf("apply deferred branch updates: %w", err) } - if p.metrics != nil { - p.metrics.updateBranch.Add(uint64(n)) - } return nil } diff --git a/execution/commitment/prom_metrics.go b/execution/commitment/prom_metrics.go new file mode 100644 index 00000000000..a507343ab3b --- /dev/null +++ b/execution/commitment/prom_metrics.go @@ -0,0 +1,106 @@ +// Copyright 2026 The Erigon Authors +// This file is part of Erigon. +// +// Erigon is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Erigon is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with Erigon. If not, see . + +package commitment + +import ( + "time" + + "github.com/erigontech/erigon/diagnostics/metrics" +) + +// Commitment metrics are counters and a histogram rather than gauges holding a +// pre-divided rate: rate() belongs in the query, so the averaging window stays +// the reader's choice and a missed scrape costs a sample, not an interval. +// +// They are emitted per round, from the trie itself, because a round's Metrics +// is per-round on both engines and there is no cumulative series to difference: +// HexPatriciaHashed.Process resets at the top of every round, and the parallel +// trie now does the same. Publishing from the executor's log ticker instead +// would both mis-difference that snapshot and drop every round between ticks. +var ( + mxRounds = metrics.GetOrCreateCounter("commitment_rounds_total") + + // Buckets span a fast incremental block through a whale fold. + mxRoundDuration = metrics.NewHistogram("commitment_round_duration_seconds", + []float64{0.001, 0.005, 0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60}) + + mxKeys = metrics.GetOrCreateCounter("commitment_keys_total") + mxFolds = metrics.GetOrCreateCounter("commitment_folds_total") + mxUnfolds = metrics.GetOrCreateCounter("commitment_unfolds_total") + mxBranchPuts = metrics.GetOrCreateCounter("commitment_branch_writes_total") + mxReadBytes = metrics.GetOrCreateCounter("commitment_branch_read_bytes_total") + mxWriteBytes = metrics.GetOrCreateCounter("commitment_branch_write_bytes_total") + + // kind=address|storage. Cell traversals, not distinct keys: the parallel + // engine re-walks subtrees on mount+replay, so this exceeds commitment_keys_total. + mxTraversals = metrics.GetOrCreateCounterVec("commitment_key_traversals_total", + []string{"kind"}, "cell traversals during commitment, by key kind") + + // kind=account|storage|branch + mxReads = metrics.GetOrCreateCounterVec("commitment_reads_total", + []string{"kind"}, "PatriciaContext reads during commitment") +) + +func addU64(c metrics.Counter, v uint64) { + if v > 0 { + c.AddUint64(v) + } +} + +func addVec(v *metrics.CounterVec, kind string, n uint64) { + if n > 0 { + v.WithLabelValues(kind).Add(float64(n)) + } +} + +// publishBranchWrites bills n branch writes of bytesOut bytes. Called where the +// write lands rather than at a round boundary, because deferred writes can be +// applied after their round has closed. m, when non-nil, also gets them for the +// trie's own log and CSV counters. +func publishBranchWrites(n, bytesOut int, m *Metrics) { + if n <= 0 { + return + } + mxBranchPuts.AddInt(n) + if bytesOut > 0 { + mxWriteBytes.AddInt(bytesOut) + } + if m != nil { + m.updateBranch.Add(uint64(n)) + m.AddBranchWrite(bytesOut) + } +} + +// observeRound publishes one finished round. Branch writes are not published +// here — publishBranchWrites bills those where they land. +func observeRound(m *Metrics, start time.Time) { + mxRounds.Inc() + mxRoundDuration.ObserveDuration(start) + if m == nil { + return + } + v := m.AsValues() + addU64(mxKeys, v.RoundKeys) + addU64(mxFolds, v.Folds) + addU64(mxUnfolds, v.Unfolds) + addU64(mxReadBytes, v.BranchReadBytes) + addVec(mxTraversals, "address", v.AddressKeys) + addVec(mxTraversals, "storage", v.StorageKeys) + addVec(mxReads, "account", v.LoadAccount) + addVec(mxReads, "storage", v.LoadStorage) + addVec(mxReads, "branch", v.LoadBranch) +} diff --git a/execution/stagedsync/exec3.go b/execution/stagedsync/exec3.go index cce54d21e4c..5f10da51c61 100644 --- a/execution/stagedsync/exec3.go +++ b/execution/stagedsync/exec3.go @@ -171,7 +171,6 @@ func execV3(ctx context.Context, logEvery := time.NewTicker(20 * time.Second) defer logEvery.Stop() defer resetExecGauges(ctx) - defer resetCommitmentGauges(ctx) defer resetDomainGauges(ctx) stepsInDb := rawdbhelpers.IdxStepsCountV3(applyTx, doms.StepSize()) @@ -299,7 +298,6 @@ func execV3Serial(ctx context.Context, logEvery := time.NewTicker(20 * time.Second) defer logEvery.Stop() defer resetExecGauges(ctx) - defer resetCommitmentGauges(ctx) defer resetDomainGauges(ctx) stepsInDb := rawdbhelpers.IdxStepsCountV3(applyTx, doms.StepSize()) diff --git a/execution/stagedsync/exec3_metrics.go b/execution/stagedsync/exec3_metrics.go index 8dfe130000f..907e97205f9 100644 --- a/execution/stagedsync/exec3_metrics.go +++ b/execution/stagedsync/exec3_metrics.go @@ -106,19 +106,6 @@ var ( mxExecCodeDomainFileReads = metrics.NewGauge(`exec_domain_file_read_rate{domain="code"}`) mxExecCodeDomainFileReadDuration = metrics.NewGauge(`exec_domain_file_read_dur{domain="code"}`) - mxCommitmentTransactions = metrics.NewGauge(`commit_txns`) - mxCommitmentBlocks = metrics.NewGauge("commit_blocks") - mxCommitmentBlockDuration = metrics.NewGauge("commit_block_dur") - mxCommitmentReadRate = metrics.NewGauge("commit_read_rate") - mxCommitmentAccountReadRate = metrics.NewGauge("commit_account_read_rate") - mxCommitmentStorageReadRate = metrics.NewGauge("commit_storage_read_rate") - mxCommitmentBranchReadRate = metrics.NewGauge("commit_branch_read_rate") - mxCommitmentBranchWriteRate = metrics.NewGauge("commit_branch_write_rate") - mxCommitmentKeyRate = metrics.NewGauge("commit_key_rate") - mxCommitmentAccountKeyRate = metrics.NewGauge("commit_account_key_rate") - mxCommitmentStorageKeyRate = metrics.NewGauge("commit_storage_key_rate") - mxCommitmentFoldRate = metrics.NewGauge("commit_fold_rate") - mxCommitmentUnfoldRate = metrics.NewGauge("commit_unfold_rate") mxCommitmentDomainReads = metrics.NewGauge(`exec_domain_read_rate{domain="commitment"}`) mxCommitmentDomainReadDuration = metrics.NewGauge(`exec_domain_read_dur{domain="commitment"}`) mxCommitmentDomainCacheReads = metrics.NewGauge(`exec_domain_cache_read_rate{domain="commitment"}`) @@ -175,7 +162,6 @@ func (g *gaugeResetTask) reset() { } var execResetTask = gaugeResetTask{} -var commitResetTask = gaugeResetTask{} var domainResetTask = gaugeResetTask{} // enough time to alow the sampler to scrape @@ -204,25 +190,6 @@ func resetExecGauges(ctx context.Context) { } } -func resetCommitmentGauges(ctx context.Context) { - commitResetTask.Lock() - defer commitResetTask.Unlock() - if commitResetTask.Timer != nil { - if commitResetTask.stopped { - commitResetTask.Timer = time.NewTimer(resetDelay) - } else { - commitResetTask.Reset(resetDelay) - } - } else { - commitResetTask.Timer = time.NewTimer(resetDelay) - commitResetTask.ctx = ctx - commitResetTask.gauges = []metrics.Gauge{ - mxCommitmentTransactions, mxCommitmentBlocks, mxCommitmentBlockDuration, - } - commitResetTask.run(ctx) - } -} - func resetDomainGauges(ctx context.Context) { domainResetTask.Lock() defer domainResetTask.Unlock() @@ -248,9 +215,8 @@ func resetDomainGauges(ctx context.Context) { mxExecDomainPutKeySize, mxExecDomainPutValueSize, mxExecAccountDomainPutRate, mxExecAccountDomainPutSize, mxExecAccountDomainPutKeySize, mxExecAccountDomainPutValueSize, mxExecStorageDomainPutRate, mxExecStorageDomainPutSize, mxExecStorageDomainPutKeySize, mxExecStorageDomainPutValueSize, mxExecCodeDomainPutRate, mxExecCodeDomainPutSize, - mxExecCodeDomainPutKeySize, mxExecCodeDomainPutValueSize, mxCommitmentReadRate, mxCommitmentAccountReadRate, - mxCommitmentStorageReadRate, mxCommitmentBranchReadRate, mxCommitmentBranchWriteRate, mxCommitmentKeyRate, - mxCommitmentAccountKeyRate, mxCommitmentStorageKeyRate, mxCommitmentFoldRate, mxCommitmentUnfoldRate, mxCommitmentDomainReads, + mxExecCodeDomainPutKeySize, mxExecCodeDomainPutValueSize, + mxCommitmentDomainReads, mxCommitmentDomainReadDuration, mxCommitmentDomainCacheReads, mxCommitmentDomainCacheReadDuration, mxCommitmentDomainDbReads, mxCommitmentDomainDbReadDuration, mxCommitmentDomainFileReads, mxCommitmentDomainFileReadDuration, mxCommitmentDomainPutRate, mxCommitmentDomainPutSize, mxCommitmentDomainPutKeySize, mxCommitmentDomainPutValueSize, @@ -463,45 +429,38 @@ func NewProgress(initialBlockNum, initialTxNum, commitThreshold uint64, logPrefi } type Progress struct { - initialTime time.Time - initialTxNum uint64 - initialBlockNum uint64 - prevExecTime time.Time - prevExecutedBlockNum int64 - prevExecutedTxNum uint64 - prevExecutedGas int64 - prevExecCount uint64 - prevActivations int64 - prevTaskDuration time.Duration - prevTaskReadDuration time.Duration - prevAccountReadDuration time.Duration - prevStorageReadDuration time.Duration - prevCodeReadDuration time.Duration - prevTaskGas int64 - prevBlockCount int64 - prevBlockDuration time.Duration - prevAbortCount uint64 - prevInvalidCount uint64 - prevReadCount int64 - prevAccountReadCount int64 - prevStorageReadCount int64 - prevCodeReadCount int64 - prevWriteCount uint64 - prevCommitTime time.Time - prevCommittedBlockNum uint64 - prevCommittedTxNum uint64 - prevCommitLogGas int64 - prevCommitmentKeyCount uint64 - prevCommitmentAccountKeyCount uint64 - prevCommitmentStorageKeyCount uint64 - prevCommitmentAccountReadCount uint64 - prevCommitmentStorageReadCount uint64 - prevBranchReadCount uint64 - prevBranchWriteCount uint64 - commitThreshold uint64 - prevDomainMetrics *kvmetrics.DomainMetrics - logPrefix string - logger log.Logger + initialTime time.Time + initialTxNum uint64 + initialBlockNum uint64 + prevExecTime time.Time + prevExecutedBlockNum int64 + prevExecutedTxNum uint64 + prevExecutedGas int64 + prevExecCount uint64 + prevActivations int64 + prevTaskDuration time.Duration + prevTaskReadDuration time.Duration + prevAccountReadDuration time.Duration + prevStorageReadDuration time.Duration + prevCodeReadDuration time.Duration + prevTaskGas int64 + prevBlockCount int64 + prevBlockDuration time.Duration + prevAbortCount uint64 + prevInvalidCount uint64 + prevReadCount int64 + prevAccountReadCount int64 + prevStorageReadCount int64 + prevCodeReadCount int64 + prevWriteCount uint64 + prevCommitTime time.Time + prevCommittedBlockNum uint64 + prevCommittedTxNum uint64 + prevCommitLogGas int64 + commitThreshold uint64 + prevDomainMetrics *kvmetrics.DomainMetrics + logPrefix string + logger log.Logger } type executor interface { @@ -774,13 +733,6 @@ func (p *Progress) LogCommitments(rs *state.StateV3, ex executor, stepsInDb floa } lastProgress.Metrics.RLock() - accountKeyCount := lastProgress.Metrics.AddressKeys - storageKeyCount := lastProgress.Metrics.StorageKeys - keyCount := accountKeyCount + storageKeyCount - accountReadCount := lastProgress.Metrics.LoadAccount - storageReadCount := lastProgress.Metrics.LoadStorage - branchReadCount := lastProgress.Metrics.LoadBranch - branchWriteCount := lastProgress.Metrics.UpdateBranch cacheBranchHits := lastProgress.Metrics.CacheBranch cacheAccountHits := lastProgress.Metrics.CacheAccount cacheStorageHits := lastProgress.Metrics.CacheStorage @@ -789,33 +741,6 @@ func (p *Progress) LogCommitments(rs *state.StateV3, ex executor, stepsInDb floa missStorageCount := lastProgress.Metrics.MissStorage lastProgress.Metrics.RUnlock() - curKeyCount := int64(keyCount - p.prevCommitmentKeyCount) - curAccountKeyCount := int64(accountKeyCount - p.prevCommitmentAccountKeyCount) - curStorageKeyCount := int64(storageKeyCount - p.prevCommitmentStorageKeyCount) - - mxCommitmentKeyRate.Set(float64(curKeyCount) / interval.Seconds()) - mxCommitmentAccountKeyRate.Set(float64(curAccountKeyCount) / interval.Seconds()) - mxCommitmentStorageKeyRate.Set(float64(curStorageKeyCount) / interval.Seconds()) - - curAccountReadCount := int64(accountReadCount - p.prevCommitmentAccountReadCount) - curStorageReadCount := int64(storageReadCount - p.prevCommitmentStorageReadCount) - curBranchReadCount := int64(branchReadCount - p.prevBranchReadCount) - curBranchWriteCount := int64(branchWriteCount - p.prevBranchWriteCount) - - curReadCount := curAccountReadCount + curStorageReadCount + curBranchReadCount - curReadRate := uint64(float64(curReadCount) / interval.Seconds()) - curBranchWriteRate := uint64(float64(curBranchWriteCount) / interval.Seconds()) - - mxCommitmentReadRate.SetUint64(curReadRate) - mxCommitmentAccountReadRate.Set(float64(curAccountReadCount) / interval.Seconds()) - mxCommitmentStorageReadRate.Set(float64(curStorageReadCount) / interval.Seconds()) - mxCommitmentBranchReadRate.Set(float64(curBranchReadCount) / interval.Seconds()) - mxCommitmentBranchWriteRate.SetUint64(curBranchWriteRate) - - mxCommitmentTransactions.Set(float64(committedTxSec)) - mxCommitmentBlocks.Set(float64(committedDiffBlocks)) - mxCommitmentBlockDuration.Set(float64(commitedBlockDur)) - totalCacheHits := cacheBranchHits + cacheAccountHits + cacheStorageHits totalCacheMisses := missBranchCount + missAccountCount + missStorageCount diff --git a/execution/stagedsync/exec3_serial.go b/execution/stagedsync/exec3_serial.go index 4bebda74da6..d253a5834be 100644 --- a/execution/stagedsync/exec3_serial.go +++ b/execution/stagedsync/exec3_serial.go @@ -251,7 +251,6 @@ func (se *serialExecutor) exec(ctx context.Context, execStage *StageState, u Unw if !ok { return b.HeaderNoCopy(), rwTx, nil } - resetCommitmentGauges(ctx) se.txExecutor.lastCommittedBlockNum.Store(b.NumberU64()) se.txExecutor.lastCommittedTxNum.Store(inputTxNum) se.logger.Info(