From f4d42ae1b4ffe5545c0e3ccd40c497ae9f246ad7 Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 22 Sep 2026 17:06:10 +0700 Subject: [PATCH 01/18] commitment: benchmark ModeParallel collect and Process together --- execution/commitment/presort_bench_test.go | 83 ++++++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 execution/commitment/presort_bench_test.go diff --git a/execution/commitment/presort_bench_test.go b/execution/commitment/presort_bench_test.go new file mode 100644 index 00000000000..c9a3cc105f9 --- /dev/null +++ b/execution/commitment/presort_bench_test.go @@ -0,0 +1,83 @@ +// 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 ( + "context" + "fmt" + "runtime" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/common/length" +) + +func benchTouchAndProcess(b *testing.B, pk [][]byte, updates []Update, workers int) { + ctx := context.Background() + tmp := b.TempDir() + b.ReportAllocs() + var pph *ParallelPatriciaHashed + defer func() { + if pph != nil { + pph.Release() + } + }() + for b.Loop() { + b.StopTimer() + ms := NewMockState(b) + ms.SetConcurrentCommitment(true) + require.NoError(b, ms.applyPlainUpdates(pk, updates)) + factory := mockTrieCtxFactory(ms) + if pph == nil { + pph = NewParallelPatriciaHashed(factory, length.Addr, DefaultTrieConfig()) + pph.SetNumWorkers(workers) + } else { + pph.SetTrieContextFactory(factory) + pph.ResetContext(ms) + } + pph.RootTrie().Reset() + upds := NewUpdates(ModeParallel, tmp, KeyToHexNibbleHash) + b.StartTimer() + + WrapKeyUpdatesInto(b, upds, pk, updates) + _, err := pph.Process(ctx, upds, "", nil, WarmupConfig{}) + + b.StopTimer() + require.NoError(b, err) + upds.Close() + b.StartTimer() + } +} + +func Benchmark_ModeParallel_TouchAndProcess(b *testing.B) { + ncpu := runtime.NumCPU() + for _, c := range []struct { + name string + build func(testing.TB) ([][]byte, []Update) + }{ + {"100K-AccountsOnly", build100KAccountsCorpus}, + {"500K-StorageHeavy", build500KStorageHeavyCorpus}, + } { + pk, updates := c.build(b) + for _, w := range []int{4, ncpu} { + b.Run(fmt.Sprintf("%s/w%d", c.name, w), func(b *testing.B) { + benchTouchAndProcess(b, pk, updates, w) + }) + } + } +} From 837ba6b85a4fd62ea9114289eabeb0172bc15d1f Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 22 Sep 2026 17:13:39 +0700 Subject: [PATCH 02/18] commitment: ModeParallel collects hashed keys and pre-sorts them in parallel before Process --- execution/commitment/commitment.go | 6 +- execution/commitment/commitment_test.go | 6 + execution/commitment/fork_walk_test.go | 11 +- .../commitment/parallel_patricia_hashed.go | 1 + execution/commitment/parallel_presort.go | 125 +++++++++++++ execution/commitment/parallel_presort_test.go | 170 ++++++++++++++++++ execution/commitment/parallel_update.go | 36 +++- execution/commitment/prefix_trie_test.go | 17 +- 8 files changed, 354 insertions(+), 18 deletions(-) create mode 100644 execution/commitment/parallel_presort.go create mode 100644 execution/commitment/parallel_presort_test.go diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go index 7f9aa3dbf93..64986957435 100644 --- a/execution/commitment/commitment.go +++ b/execution/commitment/commitment.go @@ -1499,7 +1499,7 @@ func (t *Updates) TouchPlainKey(key string, val []byte, fn func(c *KeyUpdate, va hashedKey := t.hashKey(keyBytes) ik := t.parallel.internKey(keyBytes) t.keys[key] = struct{}{} - t.parallel.Insert(hashedKey, ik, nil) + t.parallel.Collect(hashedKey, ik, nil) default: } } @@ -1560,7 +1560,7 @@ func (t *Updates) TouchPlainKeyDirect(key string, update *Update) { ik = t.parallel.internKey(keyBytes) t.keys[key] = struct{}{} } - t.parallel.Insert(hashedKey, ik, u) + t.parallel.Collect(hashedKey, ik, u) default: } } @@ -1582,7 +1582,7 @@ func (t *Updates) TouchHashedKey(hashedKey []byte) { } dedupKey := string(hashedKey) if _, ok := t.keys[dedupKey]; !ok { - t.parallel.Insert(hashedKey, nil, nil) + t.parallel.Collect(hashedKey, nil, nil) t.keys[dedupKey] = struct{}{} } case ModeUpdate: diff --git a/execution/commitment/commitment_test.go b/execution/commitment/commitment_test.go index 1745cd835d8..ff75568693d 100644 --- a/execution/commitment/commitment_test.go +++ b/execution/commitment/commitment_test.go @@ -882,12 +882,14 @@ func TestUpdatesModeParallel_TouchPlainKeyRoutes(t *testing.T) { } require.Equal(t, uint64(len(keys)), ut.Size()) + ut.parallel.Build() require.NotNil(t, ut.parallel.trie.root) require.EqualValues(t, len(keys), ut.parallel.trie.root.subtreeCount, "every touched key must show up in the prefix trie") ut.TouchPlainKey(string(keys[0]), []byte("v2"), ut.TouchStorage) require.Equal(t, uint64(len(keys)), ut.Size()) + ut.parallel.Build() require.EqualValues(t, len(keys), ut.parallel.trie.root.subtreeCount, "duplicate TouchPlainKey must not double-count in the trie") } @@ -912,6 +914,7 @@ func TestUpdatesModeParallel_RepeatTouchDoesNotRehash(t *testing.T) { require.EqualValues(t, 1, hashCalls.Load(), "a repeat touch must not rehash the key") require.Equal(t, uint64(1), ut.Size()) + ut.parallel.Build() require.NotNil(t, ut.parallel.trie.root) require.EqualValues(t, 1, ut.parallel.trie.root.subtreeCount) } @@ -930,6 +933,7 @@ func TestUpdatesModeParallel_TouchHashedKey(t *testing.T) { ut.TouchHashedKey(hk1) require.Equal(t, uint64(2), ut.Size()) + ut.parallel.Build() require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount) } @@ -947,6 +951,7 @@ func TestUpdatesModeParallel_Reset(t *testing.T) { ut.TouchPlainKey(string(k), []byte("v"), ut.TouchStorage) } require.Equal(t, uint64(2), ut.Size()) + ut.parallel.Build() require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount) ut.Reset() @@ -962,6 +967,7 @@ func TestUpdatesModeParallel_Reset(t *testing.T) { ut.TouchPlainKey(string(k), []byte("v"), ut.TouchStorage) } require.Equal(t, uint64(2), ut.Size()) + ut.parallel.Build() require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount) } diff --git a/execution/commitment/fork_walk_test.go b/execution/commitment/fork_walk_test.go index 0ca0033d461..76b3c30ed73 100644 --- a/execution/commitment/fork_walk_test.go +++ b/execution/commitment/fork_walk_test.go @@ -400,7 +400,8 @@ func TestForkWalk_ForkStartsHelpersOnlyOnIdleLeases(t *testing.T) { } bw := &walker{trie: base} require.NoError(t, fw.attach(ctx, bw)) - root := ut.parallel.trie.root + ut.parallel.Build() + root := ut.parallel.trie.root err := fw.walk(ctx, bw, root, append(make([]byte, 0, forkPathCap), root.ext...)) fw.detach(bw) require.NoError(t, err) @@ -493,6 +494,7 @@ func TestForkWalk_HelperErrorIsNotMaskedByCancellation(t *testing.T) { bw := &walker{trie: base} require.NoError(t, fw.attach(ctx, bw)) require.Same(t, first, bw.lease, "the forking walker must hold the lease whose reads wait for the helper to fail") + ut.parallel.Build() root := ut.parallel.trie.root err = fw.walk(ctx, bw, root, append(make([]byte, 0, forkPathCap), root.ext...)) fw.detach(bw) @@ -573,9 +575,10 @@ func absentDeleteCorpus(freshRegion bool) (k1 [][]byte, u1 []Update, k2 [][]byte func TestForkWalk_SplitsExcludesTheNodesOwnKey(t *testing.T) { pu := newParallelUpdate() - pu.Insert(nibs(0x01, 0x02), []byte("pk-D"), nil) - pu.Insert(nibs(0x01, 0x02, 0x03), []byte("pk-A"), nil) - pu.Insert(nibs(0x01, 0x02, 0x04), []byte("pk-B"), nil) + pu.Collect(nibs(0x01, 0x02), []byte("pk-D"), nil) + pu.Collect(nibs(0x01, 0x02, 0x03), []byte("pk-A"), nil) + pu.Collect(nibs(0x01, 0x02, 0x04), []byte("pk-B"), nil) + pu.Build() node := pu.trie.root.children[0] require.NotNil(t, node.plainKey) require.Equal(t, uint32(3), node.subtreeCount) diff --git a/execution/commitment/parallel_patricia_hashed.go b/execution/commitment/parallel_patricia_hashed.go index 76b02e0ac0c..b42c59981c5 100644 --- a/execution/commitment/parallel_patricia_hashed.go +++ b/execution/commitment/parallel_patricia_hashed.go @@ -253,6 +253,7 @@ func (p *ParallelPatriciaHashed) Process( defer func() { observeRound(p.metrics, roundStart) }() pu := updates.parallel + pu.Build() 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 // an empty base would publish the empty-trie root instead. diff --git a/execution/commitment/parallel_presort.go b/execution/commitment/parallel_presort.go new file mode 100644 index 00000000000..ce369c7e131 --- /dev/null +++ b/execution/commitment/parallel_presort.go @@ -0,0 +1,125 @@ +// 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 ( + "bytes" + "cmp" + "runtime" + "slices" + "sync" + "sync/atomic" +) + +const presortBuckets = 256 + +const presortEntrySize = 64 + +const presortBucketKeep = 8192 + +const presortParallelMin = 4096 + +type presortEntry struct { + hashedKey []byte + plainKey []byte + update *Update + seq uint32 +} + +type presorter struct { + buckets [presortBuckets][]presortEntry + count int + bytes int + seq uint32 +} + +func presortBucketOf(hashedKey []byte) int { + switch len(hashedKey) { + case 0: + return 0 + case 1: + return int(hashedKey[0]&0x0f) << 4 + default: + return int(hashedKey[0]&0x0f)<<4 | int(hashedKey[1]&0x0f) + } +} + +func (p *presorter) collect(hashedKey, plainKey []byte, update *Update) { + i := presortBucketOf(hashedKey) + p.buckets[i] = append(p.buckets[i], presortEntry{ + hashedKey: hashedKey, + plainKey: plainKey, + update: update, + seq: p.seq, + }) + p.seq++ + p.count++ + p.bytes += presortEntrySize + len(hashedKey) +} + +func presortLess(a, b presortEntry) int { + if c := bytes.Compare(a.hashedKey, b.hashedKey); c != 0 { + return c + } + return cmp.Compare(a.seq, b.seq) +} + +func (p *presorter) sortBuckets() { + nw := min(runtime.GOMAXPROCS(0), presortBuckets) + if nw <= 1 || p.count < presortParallelMin { + for i := range p.buckets { + if len(p.buckets[i]) > 1 { + slices.SortFunc(p.buckets[i], presortLess) + } + } + return + } + var next atomic.Int32 + var wg sync.WaitGroup + wg.Add(nw) + for range nw { + go func() { + defer wg.Done() + for { + i := int(next.Add(1)) - 1 + if i >= presortBuckets { + return + } + if len(p.buckets[i]) > 1 { + slices.SortFunc(p.buckets[i], presortLess) + } + } + }() + } + wg.Wait() +} + +func (p *presorter) releaseBucket(i int) { + if cap(p.buckets[i]) > presortBucketKeep { + p.buckets[i] = nil + return + } + clear(p.buckets[i]) + p.buckets[i] = p.buckets[i][:0] +} + +func (p *presorter) reset() { + for i := range p.buckets { + p.releaseBucket(i) + } + p.count, p.bytes, p.seq = 0, 0, 0 +} diff --git a/execution/commitment/parallel_presort_test.go b/execution/commitment/parallel_presort_test.go new file mode 100644 index 00000000000..98041bb6d50 --- /dev/null +++ b/execution/commitment/parallel_presort_test.go @@ -0,0 +1,170 @@ +// 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 ( + "math/rand" + "slices" + "testing" + + "github.com/holiman/uint256" + "github.com/stretchr/testify/require" +) + +type flatNode struct { + path string + plainKey string + update string + subtree uint32 +} + +func flattenPrefixTrie(t *testing.T, tr *prefixTrie) []flatNode { + t.Helper() + var out []flatNode + var walk func(n *prefixNode, path []byte) + walk = func(n *prefixNode, path []byte) { + path = append(path, n.ext...) + rec := flatNode{path: NibblesToString(path), subtree: n.subtreeCount} + if n.plainKey != nil { + rec.plainKey = string(n.plainKey) + } + if n.update != nil { + rec.update = n.update.String() + } + out = append(out, rec) + for i, c := range n.children { + nib := nthSetNibble(n.bitmap, i) + walk(c, append(slices.Clone(path), nib)) + } + } + walk(tr.root, nil) + return out +} + +func nthSetNibble(bitmap uint16, n int) byte { + seen := 0 + for i := range 16 { + if bitmap&(1<> 8)} + } + var upd *Update + if i%3 != 0 { + upd = &Update{Flags: BalanceUpdate, Balance: *uint256.NewInt(uint64(i) + 1)} + } + cases = append(cases, presortCase{hashedKey: hk, plainKey: pk, update: upd}) + if dupEvery > 0 && i%dupEvery == 0 { + cases = append(cases, presortCase{ + hashedKey: slices.Clone(hk), + plainKey: pk, + update: &Update{Flags: NonceUpdate, Nonce: uint64(i) + 1}, + }) + } + } + return cases +} + +func buildReferenceTrie(cases []presortCase) *prefixTrie { + tr := newPrefixTrie() + for _, c := range cases { + var upd *Update + if c.update != nil { + cp := *c.update + upd = &cp + } + tr.Insert(c.hashedKey, c.plainKey, upd) + } + return tr +} + +func buildPresortedTrie(cases []presortCase, memLimit int) *prefixTrie { + pu := newParallelUpdate() + if memLimit > 0 { + pu.memLimit = memLimit + } + for _, c := range cases { + var upd *Update + if c.update != nil { + cp := *c.update + upd = &cp + } + pu.Collect(c.hashedKey, c.plainKey, upd) + } + pu.Build() + return pu.trie +} + +func TestPresort_MatchesInsertionOrderTrie(t *testing.T) { + t.Parallel() + + for _, n := range []int{1, 2, 17, 5000} { + cases := randomPresortCases(int64(n)*7919+3, n, 4) + want := flattenPrefixTrie(t, buildReferenceTrie(cases)) + got := flattenPrefixTrie(t, buildPresortedTrie(cases, 0)) + require.Equal(t, want, got, "presorted build must match insertion-order build for n=%d", n) + } +} + +func TestPresort_MidBatchFlushKeepsMergeOrder(t *testing.T) { + t.Parallel() + + cases := randomPresortCases(4242, 3000, 3) + want := flattenPrefixTrie(t, buildReferenceTrie(cases)) + got := flattenPrefixTrie(t, buildPresortedTrie(cases, 8*presortEntrySize)) + require.Equal(t, want, got, "a memory-limited flush must not change the merge order") +} + +func TestPresort_BucketOfIsOrderPreserving(t *testing.T) { + t.Parallel() + + prev := -1 + for hi := range 16 { + require.Equal(t, hi<<4, presortBucketOf([]byte{byte(hi)}), "a one-nibble key sorts before its extensions") + for lo := range 16 { + b := presortBucketOf([]byte{byte(hi), byte(lo), 0x0a}) + require.Greater(t, b, prev, "bucket index must rise with the first two nibbles") + prev = b + } + } + require.Equal(t, 0, presortBucketOf(nil)) + require.Equal(t, presortBuckets-1, prev) +} diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index a231b4d9d60..8b6861c6b3f 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -47,6 +47,9 @@ func (a *plainKeyArena) reset() { a.buf = a.buf[:0] } type parallelUpdate struct { trie *prefixTrie + pending presorter + memLimit int + deferredMu sync.Mutex deferredCombined []*DeferredBranchUpdate @@ -55,13 +58,36 @@ type parallelUpdate struct { func newParallelUpdate() *parallelUpdate { return ¶llelUpdate{ - trie: newPrefixTrie(), + trie: newPrefixTrie(), + memLimit: defaultDirectMemLimit, } } -// Insert is not safe for concurrent calls; the caller must serialize them. -func (pu *parallelUpdate) Insert(hashedKey, plainKey []byte, update *Update) { - pu.trie.Insert(hashedKey, plainKey, update) +// Collect is not safe for concurrent calls; the caller must serialize them. +func (pu *parallelUpdate) Collect(hashedKey, plainKey []byte, update *Update) { + pu.pending.collect(hashedKey, plainKey, update) + if pu.pending.bytes >= pu.memLimit { + pu.Build() + } +} + +func (pu *parallelUpdate) Build() { + if pu.pending.count == 0 { + return + } + if pu.trie == nil { + pu.pending.reset() + return + } + pu.pending.sortBuckets() + for i := range pu.pending.buckets { + b := pu.pending.buckets[i] + for j := range b { + pu.trie.Insert(b[j].hashedKey, b[j].plainKey, b[j].update) + } + pu.pending.releaseBucket(i) + } + pu.pending.count, pu.pending.bytes, pu.pending.seq = 0, 0, 0 } func (pu *parallelUpdate) internKey(plainKey []byte) []byte { @@ -78,6 +104,7 @@ func (pu *parallelUpdate) drainDeferred() { } func (pu *parallelUpdate) Reset() { + pu.pending.reset() if pu.trie != nil { pu.trie.Reset() } @@ -86,6 +113,7 @@ func (pu *parallelUpdate) Reset() { } func (pu *parallelUpdate) Close() { + pu.pending.reset() pu.trie = nil pu.drainDeferred() pu.keyArena.reset() diff --git a/execution/commitment/prefix_trie_test.go b/execution/commitment/prefix_trie_test.go index b1aefc567e8..5fd2fb3b3cf 100644 --- a/execution/commitment/prefix_trie_test.go +++ b/execution/commitment/prefix_trie_test.go @@ -400,7 +400,7 @@ func TestParallelUpdateLifecycle(t *testing.T) { t.Run("close", func(t *testing.T) { pu := newParallelUpdate() - pu.Insert(nibs(0x01, 0x02), nil, nil) + pu.Collect(nibs(0x01, 0x02), nil, nil) pu.deferredCombined = append(pu.deferredCombined, &DeferredBranchUpdate{}) pu.Close() @@ -413,9 +413,10 @@ func TestParallelUpdateLifecycle(t *testing.T) { func TestParallelUpdateInsertDelegates(t *testing.T) { pu := newParallelUpdate() - pu.Insert(nibs(0x01, 0x02, 0x03), nil, nil) - pu.Insert(nibs(0x01, 0x02, 0x04), nil, nil) - pu.Insert(nibs(0x05, 0x06, 0x07), nil, nil) + pu.Collect(nibs(0x01, 0x02, 0x03), nil, nil) + pu.Collect(nibs(0x01, 0x02, 0x04), nil, nil) + pu.Collect(nibs(0x05, 0x06, 0x07), nil, nil) + pu.Build() expected := newPrefixTrie() expected.Insert(nibs(0x01, 0x02, 0x03), nil, nil) @@ -431,9 +432,10 @@ func TestParallelUpdateInsertDelegates(t *testing.T) { func TestParallelUpdateResetClearsAllState(t *testing.T) { pu := newParallelUpdate() - pu.Insert(nibs(0x01, 0x02, 0x03), nil, nil) - pu.Insert(nibs(0x05, 0x06, 0x07), nil, nil) + pu.Collect(nibs(0x01, 0x02, 0x03), nil, nil) + pu.Collect(nibs(0x05, 0x06, 0x07), nil, nil) pu.deferredCombined = append(pu.deferredCombined, &DeferredBranchUpdate{}) + pu.Build() require.Equal(t, uint32(2), pu.trie.root.subtreeCount) require.Len(t, pu.deferredCombined, 1) @@ -444,7 +446,8 @@ func TestParallelUpdateResetClearsAllState(t *testing.T) { assert.Equal(t, 1, pu.trie.arena.nodeCount(), "arena must be reset to a single root node") assert.Empty(t, pu.deferredCombined, "deferredCombined must be cleared") - pu.Insert(nibs(0x0A, 0x0B), nil, nil) + pu.Collect(nibs(0x0A, 0x0B), nil, nil) + pu.Build() assert.Equal(t, uint32(1), pu.trie.root.subtreeCount) } From 52eafe445e90bfe46f55a2e5e82c7edf02e69eda Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 22 Sep 2026 17:15:07 +0700 Subject: [PATCH 03/18] commitment: copy the caller buffer in ModeParallel TouchHashedKey, size presort entries with unsafe.Sizeof --- execution/commitment/commitment.go | 2 +- execution/commitment/parallel_presort.go | 5 +++-- execution/commitment/parallel_presort_test.go | 21 +++++++++++++++++++ 3 files changed, 25 insertions(+), 3 deletions(-) diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go index 64986957435..3019348b774 100644 --- a/execution/commitment/commitment.go +++ b/execution/commitment/commitment.go @@ -1582,7 +1582,7 @@ func (t *Updates) TouchHashedKey(hashedKey []byte) { } dedupKey := string(hashedKey) if _, ok := t.keys[dedupKey]; !ok { - t.parallel.Collect(hashedKey, nil, nil) + t.parallel.Collect(common.ToBytesZeroCopy(dedupKey), nil, nil) t.keys[dedupKey] = struct{}{} } case ModeUpdate: diff --git a/execution/commitment/parallel_presort.go b/execution/commitment/parallel_presort.go index ce369c7e131..a04ffa0631a 100644 --- a/execution/commitment/parallel_presort.go +++ b/execution/commitment/parallel_presort.go @@ -23,12 +23,11 @@ import ( "slices" "sync" "sync/atomic" + "unsafe" ) const presortBuckets = 256 -const presortEntrySize = 64 - const presortBucketKeep = 8192 const presortParallelMin = 4096 @@ -40,6 +39,8 @@ type presortEntry struct { seq uint32 } +const presortEntrySize = int(unsafe.Sizeof(presortEntry{})) + type presorter struct { buckets [presortBuckets][]presortEntry count int diff --git a/execution/commitment/parallel_presort_test.go b/execution/commitment/parallel_presort_test.go index 98041bb6d50..f1183a2d25f 100644 --- a/execution/commitment/parallel_presort_test.go +++ b/execution/commitment/parallel_presort_test.go @@ -153,6 +153,27 @@ func TestPresort_MidBatchFlushKeepsMergeOrder(t *testing.T) { require.Equal(t, want, got, "a memory-limited flush must not change the merge order") } +func TestPresort_TouchHashedKeyCopiesCallerBuffer(t *testing.T) { + t.Parallel() + + ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash) + defer ut.Close() + + buf := make([]byte, 64) + for v := byte(1); v <= 2; v++ { + for i := range buf { + buf[i] = v + } + ut.TouchHashedKey(buf) + } + + require.Equal(t, uint64(2), ut.Size()) + ut.parallel.Build() + require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount, + "a reused caller buffer must not collapse two keys into one") + require.Len(t, ut.parallel.trie.root.children, 2) +} + func TestPresort_BucketOfIsOrderPreserving(t *testing.T) { t.Parallel() From 86c5c43a45b6ffbd9f5cf74a291891d412a4ae9f Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 22 Sep 2026 17:38:31 +0700 Subject: [PATCH 04/18] commitment: sort and insert ModeParallel key chunks on a background builder so Process finds the trie built --- execution/commitment/parallel_presort.go | 9 +-- execution/commitment/parallel_presort_test.go | 33 +++++--- execution/commitment/parallel_update.go | 76 +++++++++++++++---- 3 files changed, 86 insertions(+), 32 deletions(-) diff --git a/execution/commitment/parallel_presort.go b/execution/commitment/parallel_presort.go index a04ffa0631a..587748b5270 100644 --- a/execution/commitment/parallel_presort.go +++ b/execution/commitment/parallel_presort.go @@ -23,7 +23,6 @@ import ( "slices" "sync" "sync/atomic" - "unsafe" ) const presortBuckets = 256 @@ -39,12 +38,9 @@ type presortEntry struct { seq uint32 } -const presortEntrySize = int(unsafe.Sizeof(presortEntry{})) - type presorter struct { buckets [presortBuckets][]presortEntry count int - bytes int seq uint32 } @@ -69,7 +65,6 @@ func (p *presorter) collect(hashedKey, plainKey []byte, update *Update) { }) p.seq++ p.count++ - p.bytes += presortEntrySize + len(hashedKey) } func presortLess(a, b presortEntry) int { @@ -80,7 +75,7 @@ func presortLess(a, b presortEntry) int { } func (p *presorter) sortBuckets() { - nw := min(runtime.GOMAXPROCS(0), presortBuckets) + nw := min(runtime.GOMAXPROCS(0), presortBuckets, 1+p.count/presortParallelMin) if nw <= 1 || p.count < presortParallelMin { for i := range p.buckets { if len(p.buckets[i]) > 1 { @@ -122,5 +117,5 @@ func (p *presorter) reset() { for i := range p.buckets { p.releaseBucket(i) } - p.count, p.bytes, p.seq = 0, 0, 0 + p.count, p.seq = 0, 0 } diff --git a/execution/commitment/parallel_presort_test.go b/execution/commitment/parallel_presort_test.go index f1183a2d25f..fea1d85cf1e 100644 --- a/execution/commitment/parallel_presort_test.go +++ b/execution/commitment/parallel_presort_test.go @@ -116,10 +116,10 @@ func buildReferenceTrie(cases []presortCase) *prefixTrie { return tr } -func buildPresortedTrie(cases []presortCase, memLimit int) *prefixTrie { +func buildPresortedTrie(cases []presortCase, chunkKeys int) (*prefixTrie, bool) { pu := newParallelUpdate() - if memLimit > 0 { - pu.memLimit = memLimit + if chunkKeys > 0 { + pu.chunkKeys = chunkKeys } for _, c := range cases { var upd *Update @@ -130,7 +130,12 @@ func buildPresortedTrie(cases []presortCase, memLimit int) *prefixTrie { pu.Collect(c.hashedKey, c.plainKey, upd) } pu.Build() - return pu.trie + tr, backgrounded := pu.trie, pu.buildCh != nil + if backgrounded { + close(pu.buildCh) + pu.buildCh = nil + } + return tr, backgrounded } func TestPresort_MatchesInsertionOrderTrie(t *testing.T) { @@ -139,18 +144,24 @@ func TestPresort_MatchesInsertionOrderTrie(t *testing.T) { for _, n := range []int{1, 2, 17, 5000} { cases := randomPresortCases(int64(n)*7919+3, n, 4) want := flattenPrefixTrie(t, buildReferenceTrie(cases)) - got := flattenPrefixTrie(t, buildPresortedTrie(cases, 0)) - require.Equal(t, want, got, "presorted build must match insertion-order build for n=%d", n) + got, backgrounded := buildPresortedTrie(cases, 1<<20) + require.False(t, backgrounded, "n=%d must stay under the hand-off threshold", n) + require.Equal(t, want, flattenPrefixTrie(t, got), + "presorted build must match insertion-order build for n=%d", n) } } -func TestPresort_MidBatchFlushKeepsMergeOrder(t *testing.T) { +func TestPresort_BackgroundBuildKeepsMergeOrder(t *testing.T) { t.Parallel() - cases := randomPresortCases(4242, 3000, 3) - want := flattenPrefixTrie(t, buildReferenceTrie(cases)) - got := flattenPrefixTrie(t, buildPresortedTrie(cases, 8*presortEntrySize)) - require.Equal(t, want, got, "a memory-limited flush must not change the merge order") + for _, n := range []int{1, 2, 17, 3000} { + cases := randomPresortCases(int64(n)*4242+1, n, 3) + want := flattenPrefixTrie(t, buildReferenceTrie(cases)) + got, backgrounded := buildPresortedTrie(cases, 8) + require.Equal(t, n > 8, backgrounded, "n=%d must cross the hand-off threshold", n) + require.Equal(t, want, flattenPrefixTrie(t, got), + "chunks built in the background must not change the merge order for n=%d", n) + } } func TestPresort_TouchHashedKeyCopiesCallerBuffer(t *testing.T) { diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index 8b6861c6b3f..5ca2de3f21e 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -44,11 +44,18 @@ func (a *plainKeyArena) intern(b []byte) []byte { func (a *plainKeyArena) reset() { a.buf = a.buf[:0] } +const presortChunkKeys = 8192 + +const presortBuffers = 2 + type parallelUpdate struct { trie *prefixTrie - pending presorter - memLimit int + pending *presorter + chunkKeys int + buildCh chan *presorter + freeCh chan *presorter + inflight sync.WaitGroup deferredMu sync.Mutex deferredCombined []*DeferredBranchUpdate @@ -58,36 +65,71 @@ type parallelUpdate struct { func newParallelUpdate() *parallelUpdate { return ¶llelUpdate{ - trie: newPrefixTrie(), - memLimit: defaultDirectMemLimit, + trie: newPrefixTrie(), + pending: new(presorter), + chunkKeys: presortChunkKeys, } } // Collect is not safe for concurrent calls; the caller must serialize them. func (pu *parallelUpdate) Collect(hashedKey, plainKey []byte, update *Update) { pu.pending.collect(hashedKey, plainKey, update) - if pu.pending.bytes >= pu.memLimit { - pu.Build() + if pu.pending.count >= pu.chunkKeys { + pu.handOff() } } -func (pu *parallelUpdate) Build() { - if pu.pending.count == 0 { - return +func (pu *parallelUpdate) startBuilder() { + pu.buildCh = make(chan *presorter, presortBuffers) + pu.freeCh = make(chan *presorter, presortBuffers) + for range presortBuffers { + pu.freeCh <- new(presorter) } + go func() { + for p := range pu.buildCh { + pu.insertSorted(p) + p.reset() + pu.freeCh <- p + pu.inflight.Done() + } + }() +} + +func (pu *parallelUpdate) handOff() { if pu.trie == nil { pu.pending.reset() return } - pu.pending.sortBuckets() - for i := range pu.pending.buckets { - b := pu.pending.buckets[i] + if pu.buildCh == nil { + pu.startBuilder() + } + pu.inflight.Add(1) + pu.buildCh <- pu.pending + pu.pending = <-pu.freeCh +} + +func (pu *parallelUpdate) insertSorted(p *presorter) { + p.sortBuckets() + for i := range p.buckets { + b := p.buckets[i] for j := range b { pu.trie.Insert(b[j].hashedKey, b[j].plainKey, b[j].update) } - pu.pending.releaseBucket(i) } - pu.pending.count, pu.pending.bytes, pu.pending.seq = 0, 0, 0 +} + +func (pu *parallelUpdate) Build() { + if pu.pending.count > 0 { + if pu.buildCh != nil { + pu.handOff() + } else if pu.trie != nil { + pu.insertSorted(pu.pending) + pu.pending.reset() + } else { + pu.pending.reset() + } + } + pu.inflight.Wait() } func (pu *parallelUpdate) internKey(plainKey []byte) []byte { @@ -104,6 +146,7 @@ func (pu *parallelUpdate) drainDeferred() { } func (pu *parallelUpdate) Reset() { + pu.inflight.Wait() pu.pending.reset() if pu.trie != nil { pu.trie.Reset() @@ -113,6 +156,11 @@ func (pu *parallelUpdate) Reset() { } func (pu *parallelUpdate) Close() { + pu.inflight.Wait() + if pu.buildCh != nil { + close(pu.buildCh) + pu.buildCh, pu.freeCh = nil, nil + } pu.pending.reset() pu.trie = nil pu.drainDeferred() From 2f96b56a989ecae7b2c59321c0d076ee546fb8c1 Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 22 Sep 2026 18:06:49 +0700 Subject: [PATCH 05/18] commitment: reap the ModeParallel builder goroutine in Build and pool its chunk buffers --- execution/commitment/fork_walk_test.go | 2 +- execution/commitment/parallel_presort_test.go | 36 +++++++++++------ execution/commitment/parallel_update.go | 39 ++++++++++++++----- 3 files changed, 56 insertions(+), 21 deletions(-) diff --git a/execution/commitment/fork_walk_test.go b/execution/commitment/fork_walk_test.go index 76b3c30ed73..ea307ca7e4b 100644 --- a/execution/commitment/fork_walk_test.go +++ b/execution/commitment/fork_walk_test.go @@ -401,7 +401,7 @@ func TestForkWalk_ForkStartsHelpersOnlyOnIdleLeases(t *testing.T) { bw := &walker{trie: base} require.NoError(t, fw.attach(ctx, bw)) ut.parallel.Build() - root := ut.parallel.trie.root + root := ut.parallel.trie.root err := fw.walk(ctx, bw, root, append(make([]byte, 0, forkPathCap), root.ext...)) fw.detach(bw) require.NoError(t, err) diff --git a/execution/commitment/parallel_presort_test.go b/execution/commitment/parallel_presort_test.go index fea1d85cf1e..f8de8e2d1c9 100644 --- a/execution/commitment/parallel_presort_test.go +++ b/execution/commitment/parallel_presort_test.go @@ -116,7 +116,7 @@ func buildReferenceTrie(cases []presortCase) *prefixTrie { return tr } -func buildPresortedTrie(cases []presortCase, chunkKeys int) (*prefixTrie, bool) { +func buildPresorted(cases []presortCase, chunkKeys int) (*parallelUpdate, bool) { pu := newParallelUpdate() if chunkKeys > 0 { pu.chunkKeys = chunkKeys @@ -129,13 +129,9 @@ func buildPresortedTrie(cases []presortCase, chunkKeys int) (*prefixTrie, bool) } pu.Collect(c.hashedKey, c.plainKey, upd) } + backgrounded := pu.buildCh != nil pu.Build() - tr, backgrounded := pu.trie, pu.buildCh != nil - if backgrounded { - close(pu.buildCh) - pu.buildCh = nil - } - return tr, backgrounded + return pu, backgrounded } func TestPresort_MatchesInsertionOrderTrie(t *testing.T) { @@ -144,9 +140,9 @@ func TestPresort_MatchesInsertionOrderTrie(t *testing.T) { for _, n := range []int{1, 2, 17, 5000} { cases := randomPresortCases(int64(n)*7919+3, n, 4) want := flattenPrefixTrie(t, buildReferenceTrie(cases)) - got, backgrounded := buildPresortedTrie(cases, 1<<20) + pu, backgrounded := buildPresorted(cases, 1<<20) require.False(t, backgrounded, "n=%d must stay under the hand-off threshold", n) - require.Equal(t, want, flattenPrefixTrie(t, got), + require.Equal(t, want, flattenPrefixTrie(t, pu.trie), "presorted build must match insertion-order build for n=%d", n) } } @@ -157,13 +153,31 @@ func TestPresort_BackgroundBuildKeepsMergeOrder(t *testing.T) { for _, n := range []int{1, 2, 17, 3000} { cases := randomPresortCases(int64(n)*4242+1, n, 3) want := flattenPrefixTrie(t, buildReferenceTrie(cases)) - got, backgrounded := buildPresortedTrie(cases, 8) + pu, backgrounded := buildPresorted(cases, 8) require.Equal(t, n > 8, backgrounded, "n=%d must cross the hand-off threshold", n) - require.Equal(t, want, flattenPrefixTrie(t, got), + require.Nil(t, pu.buildCh, "Build must reap the builder goroutine") + require.Equal(t, want, flattenPrefixTrie(t, pu.trie), "chunks built in the background must not change the merge order for n=%d", n) } } +func TestPresort_BuffersReturnToThePoolAcrossBatches(t *testing.T) { + t.Parallel() + + cases := randomPresortCases(777, 200, 0) + pu, backgrounded := buildPresorted(cases, 8) + require.True(t, backgrounded) + require.Len(t, pu.pool, presortBuffers, "every chunk buffer must come back to the pool") + + pu.Reset() + for _, c := range cases { + pu.Collect(c.hashedKey, c.plainKey, nil) + } + pu.Build() + require.Len(t, pu.pool, presortBuffers, "a second batch must reuse the pooled buffers") + require.EqualValues(t, len(cases), pu.trie.root.subtreeCount) +} + func TestPresort_TouchHashedKeyCopiesCallerBuffer(t *testing.T) { t.Parallel() diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index 5ca2de3f21e..dc0a2743792 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -55,6 +55,7 @@ type parallelUpdate struct { chunkKeys int buildCh chan *presorter freeCh chan *presorter + pool []*presorter inflight sync.WaitGroup deferredMu sync.Mutex @@ -83,16 +84,36 @@ func (pu *parallelUpdate) startBuilder() { pu.buildCh = make(chan *presorter, presortBuffers) pu.freeCh = make(chan *presorter, presortBuffers) for range presortBuffers { - pu.freeCh <- new(presorter) + if n := len(pu.pool); n > 0 { + pu.freeCh <- pu.pool[n-1] + pu.pool = pu.pool[:n-1] + } else { + pu.freeCh <- new(presorter) + } } - go func() { - for p := range pu.buildCh { + go func(build <-chan *presorter, free chan<- *presorter) { + for p := range build { pu.insertSorted(p) p.reset() - pu.freeCh <- p + free <- p pu.inflight.Done() } - }() + }(pu.buildCh, pu.freeCh) +} + +func (pu *parallelUpdate) stopBuilder() { + if pu.buildCh == nil { + return + } + close(pu.buildCh) + for range presortBuffers { + select { + case p := <-pu.freeCh: + pu.pool = append(pu.pool, p) + default: + } + } + pu.buildCh, pu.freeCh = nil, nil } func (pu *parallelUpdate) handOff() { @@ -130,6 +151,7 @@ func (pu *parallelUpdate) Build() { } } pu.inflight.Wait() + pu.stopBuilder() } func (pu *parallelUpdate) internKey(plainKey []byte) []byte { @@ -147,6 +169,7 @@ func (pu *parallelUpdate) drainDeferred() { func (pu *parallelUpdate) Reset() { pu.inflight.Wait() + pu.stopBuilder() pu.pending.reset() if pu.trie != nil { pu.trie.Reset() @@ -157,10 +180,8 @@ func (pu *parallelUpdate) Reset() { func (pu *parallelUpdate) Close() { pu.inflight.Wait() - if pu.buildCh != nil { - close(pu.buildCh) - pu.buildCh, pu.freeCh = nil, nil - } + pu.stopBuilder() + pu.pool = nil pu.pending.reset() pu.trie = nil pu.drainDeferred() From b8aadae02954ea307dad626f21952c8e91abd684 Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 22 Sep 2026 18:38:13 +0700 Subject: [PATCH 06/18] commitment: sort each ModeParallel chunk as one flat pre-sized slice instead of 256 nibble buckets --- execution/commitment/parallel_presort.go | 75 +++---------------- execution/commitment/parallel_presort_test.go | 20 +---- execution/commitment/parallel_update.go | 15 ++-- 3 files changed, 20 insertions(+), 90 deletions(-) diff --git a/execution/commitment/parallel_presort.go b/execution/commitment/parallel_presort.go index 587748b5270..79824a717c8 100644 --- a/execution/commitment/parallel_presort.go +++ b/execution/commitment/parallel_presort.go @@ -19,18 +19,9 @@ package commitment import ( "bytes" "cmp" - "runtime" "slices" - "sync" - "sync/atomic" ) -const presortBuckets = 256 - -const presortBucketKeep = 8192 - -const presortParallelMin = 4096 - type presortEntry struct { hashedKey []byte plainKey []byte @@ -39,34 +30,22 @@ type presortEntry struct { } type presorter struct { - buckets [presortBuckets][]presortEntry - count int + entries []presortEntry seq uint32 } -func presortBucketOf(hashedKey []byte) int { - switch len(hashedKey) { - case 0: - return 0 - case 1: - return int(hashedKey[0]&0x0f) << 4 - default: - return int(hashedKey[0]&0x0f)<<4 | int(hashedKey[1]&0x0f) - } -} - func (p *presorter) collect(hashedKey, plainKey []byte, update *Update) { - i := presortBucketOf(hashedKey) - p.buckets[i] = append(p.buckets[i], presortEntry{ + p.entries = append(p.entries, presortEntry{ hashedKey: hashedKey, plainKey: plainKey, update: update, seq: p.seq, }) p.seq++ - p.count++ } +func (p *presorter) count() int { return len(p.entries) } + func presortLess(a, b presortEntry) int { if c := bytes.Compare(a.hashedKey, b.hashedKey); c != 0 { return c @@ -74,48 +53,14 @@ func presortLess(a, b presortEntry) int { return cmp.Compare(a.seq, b.seq) } -func (p *presorter) sortBuckets() { - nw := min(runtime.GOMAXPROCS(0), presortBuckets, 1+p.count/presortParallelMin) - if nw <= 1 || p.count < presortParallelMin { - for i := range p.buckets { - if len(p.buckets[i]) > 1 { - slices.SortFunc(p.buckets[i], presortLess) - } - } - return - } - var next atomic.Int32 - var wg sync.WaitGroup - wg.Add(nw) - for range nw { - go func() { - defer wg.Done() - for { - i := int(next.Add(1)) - 1 - if i >= presortBuckets { - return - } - if len(p.buckets[i]) > 1 { - slices.SortFunc(p.buckets[i], presortLess) - } - } - }() - } - wg.Wait() -} - -func (p *presorter) releaseBucket(i int) { - if cap(p.buckets[i]) > presortBucketKeep { - p.buckets[i] = nil - return +func (p *presorter) sort() { + if len(p.entries) > 1 { + slices.SortFunc(p.entries, presortLess) } - clear(p.buckets[i]) - p.buckets[i] = p.buckets[i][:0] } func (p *presorter) reset() { - for i := range p.buckets { - p.releaseBucket(i) - } - p.count, p.seq = 0, 0 + clear(p.entries) + p.entries = p.entries[:0] + p.seq = 0 } diff --git a/execution/commitment/parallel_presort_test.go b/execution/commitment/parallel_presort_test.go index f8de8e2d1c9..501b566a07e 100644 --- a/execution/commitment/parallel_presort_test.go +++ b/execution/commitment/parallel_presort_test.go @@ -175,6 +175,10 @@ func TestPresort_BuffersReturnToThePoolAcrossBatches(t *testing.T) { } pu.Build() require.Len(t, pu.pool, presortBuffers, "a second batch must reuse the pooled buffers") + for _, p := range pu.pool { + require.GreaterOrEqual(t, cap(p.entries), pu.chunkKeys, + "a pooled chunk buffer must keep its capacity, or every chunk re-grows it") + } require.EqualValues(t, len(cases), pu.trie.root.subtreeCount) } @@ -198,19 +202,3 @@ func TestPresort_TouchHashedKeyCopiesCallerBuffer(t *testing.T) { "a reused caller buffer must not collapse two keys into one") require.Len(t, ut.parallel.trie.root.children, 2) } - -func TestPresort_BucketOfIsOrderPreserving(t *testing.T) { - t.Parallel() - - prev := -1 - for hi := range 16 { - require.Equal(t, hi<<4, presortBucketOf([]byte{byte(hi)}), "a one-nibble key sorts before its extensions") - for lo := range 16 { - b := presortBucketOf([]byte{byte(hi), byte(lo), 0x0a}) - require.Greater(t, b, prev, "bucket index must rise with the first two nibbles") - prev = b - } - } - require.Equal(t, 0, presortBucketOf(nil)) - require.Equal(t, presortBuckets-1, prev) -} diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index dc0a2743792..7bd89ea8e75 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -75,7 +75,7 @@ func newParallelUpdate() *parallelUpdate { // Collect is not safe for concurrent calls; the caller must serialize them. func (pu *parallelUpdate) Collect(hashedKey, plainKey []byte, update *Update) { pu.pending.collect(hashedKey, plainKey, update) - if pu.pending.count >= pu.chunkKeys { + if pu.pending.count() >= pu.chunkKeys { pu.handOff() } } @@ -88,7 +88,7 @@ func (pu *parallelUpdate) startBuilder() { pu.freeCh <- pu.pool[n-1] pu.pool = pu.pool[:n-1] } else { - pu.freeCh <- new(presorter) + pu.freeCh <- &presorter{entries: make([]presortEntry, 0, pu.chunkKeys)} } } go func(build <-chan *presorter, free chan<- *presorter) { @@ -130,17 +130,14 @@ func (pu *parallelUpdate) handOff() { } func (pu *parallelUpdate) insertSorted(p *presorter) { - p.sortBuckets() - for i := range p.buckets { - b := p.buckets[i] - for j := range b { - pu.trie.Insert(b[j].hashedKey, b[j].plainKey, b[j].update) - } + p.sort() + for i := range p.entries { + pu.trie.Insert(p.entries[i].hashedKey, p.entries[i].plainKey, p.entries[i].update) } } func (pu *parallelUpdate) Build() { - if pu.pending.count > 0 { + if pu.pending.count() > 0 { if pu.buildCh != nil { pu.handOff() } else if pu.trie != nil { From 84af2dc814348ac7216872274d8227d605e77bfb Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 22 Sep 2026 21:07:57 +0700 Subject: [PATCH 07/18] commitment: benchmark the cost of dedup by uniq map against dedup by hash plus prefix-trie lookup --- execution/commitment/dedup_bench_test.go | 106 +++++++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 execution/commitment/dedup_bench_test.go diff --git a/execution/commitment/dedup_bench_test.go b/execution/commitment/dedup_bench_test.go new file mode 100644 index 00000000000..7d4ffcd5306 --- /dev/null +++ b/execution/commitment/dedup_bench_test.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 ( + "encoding/binary" + "math/rand" + "testing" + + "github.com/erigontech/erigon/common" + "github.com/erigontech/erigon/common/length" +) + +func dedupBenchKeys(n, keyLen int) []string { + rnd := rand.New(rand.NewSource(20260922)) + keys := make([]string, n) + for i := range keys { + k := make([]byte, keyLen) + for j := 0; j+8 <= keyLen; j += 8 { + binary.BigEndian.PutUint64(k[j:], rnd.Uint64()) + } + keys[i] = string(k) + } + return keys +} + +func BenchmarkDedupCost(b *testing.B) { + const n = 50_000 + + for _, c := range []struct { + name string + keyLen int + }{ + {"account", length.Addr}, + {"storage", length.Addr + length.Hash}, + } { + keys := dedupBenchKeys(n, c.keyLen) + + b.Run(c.name+"/mapHit", func(b *testing.B) { + seen := make(map[string]struct{}, n) + for _, k := range keys { + seen[k] = struct{}{} + } + b.ReportAllocs() + i := 0 + for b.Loop() { + if _, ok := seen[keys[i]]; !ok { + b.Fatal("key must be present") + } + i = (i + 1) % n + } + }) + + b.Run(c.name+"/hash", func(b *testing.B) { + b.ReportAllocs() + i := 0 + for b.Loop() { + sinkHashedKey = KeyToHexNibbleHash(common.ToBytesZeroCopy(keys[i])) + i = (i + 1) % n + } + }) + + b.Run(c.name+"/hashCached", func(b *testing.B) { + b.ReportAllocs() + var cache addrHashCache + i := 0 + for b.Loop() { + sinkHashedKey = keyToHexNibbleHashCached(common.ToBytesZeroCopy(keys[i]), &cache) + i = (i + 1) % n + } + }) + + b.Run(c.name+"/trieHit", func(b *testing.B) { + hashed := make([][]byte, n) + tr := newPrefixTrie() + for i, k := range keys { + hashed[i] = KeyToHexNibbleHash(common.ToBytesZeroCopy(k)) + tr.Insert(hashed[i], common.ToBytesZeroCopy(k), nil) + } + b.ReportAllocs() + i := 0 + for b.Loop() { + if tr.Insert(hashed[i], common.ToBytesZeroCopy(keys[i]), nil) { + b.Fatal("key must already be in the trie") + } + i = (i + 1) % n + } + }) + } +} + +var sinkHashedKey []byte From 75b3d9603186d98da42a17b5791c14c703dc84d6 Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 22 Sep 2026 21:17:04 +0700 Subject: [PATCH 08/18] commitment: insert ModeParallel chunks in arrival order, the pre-sort cost 2.5x what it saved --- ...{parallel_presort.go => parallel_chunk.go} | 42 ++----- ...presort_test.go => parallel_chunk_test.go} | 38 +++--- execution/commitment/parallel_update.go | 35 +++--- .../prefix_trie_order_bench_test.go | 113 ++++++++++++++++++ 4 files changed, 158 insertions(+), 70 deletions(-) rename execution/commitment/{parallel_presort.go => parallel_chunk.go} (55%) rename execution/commitment/{parallel_presort_test.go => parallel_chunk_test.go} (79%) create mode 100644 execution/commitment/prefix_trie_order_bench_test.go diff --git a/execution/commitment/parallel_presort.go b/execution/commitment/parallel_chunk.go similarity index 55% rename from execution/commitment/parallel_presort.go rename to execution/commitment/parallel_chunk.go index 79824a717c8..c02cd7e4f0f 100644 --- a/execution/commitment/parallel_presort.go +++ b/execution/commitment/parallel_chunk.go @@ -16,51 +16,27 @@ package commitment -import ( - "bytes" - "cmp" - "slices" -) - -type presortEntry struct { +type touchEntry struct { hashedKey []byte plainKey []byte update *Update - seq uint32 } -type presorter struct { - entries []presortEntry - seq uint32 +type touchChunk struct { + entries []touchEntry } -func (p *presorter) collect(hashedKey, plainKey []byte, update *Update) { - p.entries = append(p.entries, presortEntry{ +func (c *touchChunk) collect(hashedKey, plainKey []byte, update *Update) { + c.entries = append(c.entries, touchEntry{ hashedKey: hashedKey, plainKey: plainKey, update: update, - seq: p.seq, }) - p.seq++ } -func (p *presorter) count() int { return len(p.entries) } - -func presortLess(a, b presortEntry) int { - if c := bytes.Compare(a.hashedKey, b.hashedKey); c != 0 { - return c - } - return cmp.Compare(a.seq, b.seq) -} - -func (p *presorter) sort() { - if len(p.entries) > 1 { - slices.SortFunc(p.entries, presortLess) - } -} +func (c *touchChunk) count() int { return len(c.entries) } -func (p *presorter) reset() { - clear(p.entries) - p.entries = p.entries[:0] - p.seq = 0 +func (c *touchChunk) reset() { + clear(c.entries) + c.entries = c.entries[:0] } diff --git a/execution/commitment/parallel_presort_test.go b/execution/commitment/parallel_chunk_test.go similarity index 79% rename from execution/commitment/parallel_presort_test.go rename to execution/commitment/parallel_chunk_test.go index 501b566a07e..64cf31d6a6f 100644 --- a/execution/commitment/parallel_presort_test.go +++ b/execution/commitment/parallel_chunk_test.go @@ -69,15 +69,15 @@ func nthSetNibble(bitmap uint16, n int) byte { panic("bitmap has fewer set bits than requested") } -type presortCase struct { +type touchCase struct { hashedKey []byte plainKey []byte update *Update } -func randomPresortCases(seed int64, n, dupEvery int) []presortCase { +func randomTouchCases(seed int64, n, dupEvery int) []touchCase { rnd := rand.New(rand.NewSource(seed)) - cases := make([]presortCase, 0, n) + cases := make([]touchCase, 0, n) for i := range n { hk := make([]byte, 64) for j := range hk { @@ -91,9 +91,9 @@ func randomPresortCases(seed int64, n, dupEvery int) []presortCase { if i%3 != 0 { upd = &Update{Flags: BalanceUpdate, Balance: *uint256.NewInt(uint64(i) + 1)} } - cases = append(cases, presortCase{hashedKey: hk, plainKey: pk, update: upd}) + cases = append(cases, touchCase{hashedKey: hk, plainKey: pk, update: upd}) if dupEvery > 0 && i%dupEvery == 0 { - cases = append(cases, presortCase{ + cases = append(cases, touchCase{ hashedKey: slices.Clone(hk), plainKey: pk, update: &Update{Flags: NonceUpdate, Nonce: uint64(i) + 1}, @@ -103,7 +103,7 @@ func randomPresortCases(seed int64, n, dupEvery int) []presortCase { return cases } -func buildReferenceTrie(cases []presortCase) *prefixTrie { +func buildReferenceTrie(cases []touchCase) *prefixTrie { tr := newPrefixTrie() for _, c := range cases { var upd *Update @@ -116,7 +116,7 @@ func buildReferenceTrie(cases []presortCase) *prefixTrie { return tr } -func buildPresorted(cases []presortCase, chunkKeys int) (*parallelUpdate, bool) { +func buildChunked(cases []touchCase, chunkKeys int) (*parallelUpdate, bool) { pu := newParallelUpdate() if chunkKeys > 0 { pu.chunkKeys = chunkKeys @@ -134,26 +134,26 @@ func buildPresorted(cases []presortCase, chunkKeys int) (*parallelUpdate, bool) return pu, backgrounded } -func TestPresort_MatchesInsertionOrderTrie(t *testing.T) { +func TestChunkBuild_MatchesInsertionOrderTrie(t *testing.T) { t.Parallel() for _, n := range []int{1, 2, 17, 5000} { - cases := randomPresortCases(int64(n)*7919+3, n, 4) + cases := randomTouchCases(int64(n)*7919+3, n, 4) want := flattenPrefixTrie(t, buildReferenceTrie(cases)) - pu, backgrounded := buildPresorted(cases, 1<<20) + pu, backgrounded := buildChunked(cases, 1<<20) require.False(t, backgrounded, "n=%d must stay under the hand-off threshold", n) require.Equal(t, want, flattenPrefixTrie(t, pu.trie), "presorted build must match insertion-order build for n=%d", n) } } -func TestPresort_BackgroundBuildKeepsMergeOrder(t *testing.T) { +func TestChunkBuild_BackgroundBuildKeepsMergeOrder(t *testing.T) { t.Parallel() for _, n := range []int{1, 2, 17, 3000} { - cases := randomPresortCases(int64(n)*4242+1, n, 3) + cases := randomTouchCases(int64(n)*4242+1, n, 3) want := flattenPrefixTrie(t, buildReferenceTrie(cases)) - pu, backgrounded := buildPresorted(cases, 8) + pu, backgrounded := buildChunked(cases, 8) require.Equal(t, n > 8, backgrounded, "n=%d must cross the hand-off threshold", n) require.Nil(t, pu.buildCh, "Build must reap the builder goroutine") require.Equal(t, want, flattenPrefixTrie(t, pu.trie), @@ -161,20 +161,20 @@ func TestPresort_BackgroundBuildKeepsMergeOrder(t *testing.T) { } } -func TestPresort_BuffersReturnToThePoolAcrossBatches(t *testing.T) { +func TestChunkBuild_BuffersReturnToThePoolAcrossBatches(t *testing.T) { t.Parallel() - cases := randomPresortCases(777, 200, 0) - pu, backgrounded := buildPresorted(cases, 8) + cases := randomTouchCases(777, 200, 0) + pu, backgrounded := buildChunked(cases, 8) require.True(t, backgrounded) - require.Len(t, pu.pool, presortBuffers, "every chunk buffer must come back to the pool") + require.Len(t, pu.pool, touchChunkBuffers, "every chunk buffer must come back to the pool") pu.Reset() for _, c := range cases { pu.Collect(c.hashedKey, c.plainKey, nil) } pu.Build() - require.Len(t, pu.pool, presortBuffers, "a second batch must reuse the pooled buffers") + require.Len(t, pu.pool, touchChunkBuffers, "a second batch must reuse the pooled buffers") for _, p := range pu.pool { require.GreaterOrEqual(t, cap(p.entries), pu.chunkKeys, "a pooled chunk buffer must keep its capacity, or every chunk re-grows it") @@ -182,7 +182,7 @@ func TestPresort_BuffersReturnToThePoolAcrossBatches(t *testing.T) { require.EqualValues(t, len(cases), pu.trie.root.subtreeCount) } -func TestPresort_TouchHashedKeyCopiesCallerBuffer(t *testing.T) { +func TestChunkBuild_TouchHashedKeyCopiesCallerBuffer(t *testing.T) { t.Parallel() ut := NewUpdates(ModeParallel, t.TempDir(), KeyToHexNibbleHash) diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index 7bd89ea8e75..268e28640b8 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -44,18 +44,18 @@ func (a *plainKeyArena) intern(b []byte) []byte { func (a *plainKeyArena) reset() { a.buf = a.buf[:0] } -const presortChunkKeys = 8192 +const touchChunkKeys = 8192 -const presortBuffers = 2 +const touchChunkBuffers = 2 type parallelUpdate struct { trie *prefixTrie - pending *presorter + pending *touchChunk chunkKeys int - buildCh chan *presorter - freeCh chan *presorter - pool []*presorter + buildCh chan *touchChunk + freeCh chan *touchChunk + pool []*touchChunk inflight sync.WaitGroup deferredMu sync.Mutex @@ -67,8 +67,8 @@ type parallelUpdate struct { func newParallelUpdate() *parallelUpdate { return ¶llelUpdate{ trie: newPrefixTrie(), - pending: new(presorter), - chunkKeys: presortChunkKeys, + pending: new(touchChunk), + chunkKeys: touchChunkKeys, } } @@ -81,19 +81,19 @@ func (pu *parallelUpdate) Collect(hashedKey, plainKey []byte, update *Update) { } func (pu *parallelUpdate) startBuilder() { - pu.buildCh = make(chan *presorter, presortBuffers) - pu.freeCh = make(chan *presorter, presortBuffers) - for range presortBuffers { + pu.buildCh = make(chan *touchChunk, touchChunkBuffers) + pu.freeCh = make(chan *touchChunk, touchChunkBuffers) + for range touchChunkBuffers { if n := len(pu.pool); n > 0 { pu.freeCh <- pu.pool[n-1] pu.pool = pu.pool[:n-1] } else { - pu.freeCh <- &presorter{entries: make([]presortEntry, 0, pu.chunkKeys)} + pu.freeCh <- &touchChunk{entries: make([]touchEntry, 0, pu.chunkKeys)} } } - go func(build <-chan *presorter, free chan<- *presorter) { + go func(build <-chan *touchChunk, free chan<- *touchChunk) { for p := range build { - pu.insertSorted(p) + pu.insertChunk(p) p.reset() free <- p pu.inflight.Done() @@ -106,7 +106,7 @@ func (pu *parallelUpdate) stopBuilder() { return } close(pu.buildCh) - for range presortBuffers { + for range touchChunkBuffers { select { case p := <-pu.freeCh: pu.pool = append(pu.pool, p) @@ -129,8 +129,7 @@ func (pu *parallelUpdate) handOff() { pu.pending = <-pu.freeCh } -func (pu *parallelUpdate) insertSorted(p *presorter) { - p.sort() +func (pu *parallelUpdate) insertChunk(p *touchChunk) { for i := range p.entries { pu.trie.Insert(p.entries[i].hashedKey, p.entries[i].plainKey, p.entries[i].update) } @@ -141,7 +140,7 @@ func (pu *parallelUpdate) Build() { if pu.buildCh != nil { pu.handOff() } else if pu.trie != nil { - pu.insertSorted(pu.pending) + pu.insertChunk(pu.pending) pu.pending.reset() } else { pu.pending.reset() diff --git a/execution/commitment/prefix_trie_order_bench_test.go b/execution/commitment/prefix_trie_order_bench_test.go new file mode 100644 index 00000000000..88a25a43926 --- /dev/null +++ b/execution/commitment/prefix_trie_order_bench_test.go @@ -0,0 +1,113 @@ +// 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 ( + "bytes" + "encoding/binary" + "fmt" + "math/rand" + "slices" + "testing" + + "github.com/erigontech/erigon/common/length" +) + +func orderBenchAccountKeys(n int, seed int64) [][]byte { + rnd := rand.New(rand.NewSource(seed)) + out := make([][]byte, n) + addr := make([]byte, length.Addr) + for i := range out { + binary.BigEndian.PutUint64(addr, rnd.Uint64()) + binary.BigEndian.PutUint64(addr[8:], rnd.Uint64()) + out[i] = KeyToHexNibbleHash(addr) + } + return out +} + +func orderBenchStorageKeys(n, slotsPerAddr int, seed int64) [][]byte { + rnd := rand.New(rand.NewSource(seed)) + out := make([][]byte, 0, n) + key := make([]byte, length.Addr+length.Hash) + for len(out) < n { + binary.BigEndian.PutUint64(key, rnd.Uint64()) + binary.BigEndian.PutUint64(key[8:], rnd.Uint64()) + for s := 0; s < slotsPerAddr && len(out) < n; s++ { + binary.BigEndian.PutUint64(key[length.Addr:], rnd.Uint64()) + binary.BigEndian.PutUint64(key[length.Addr+8:], rnd.Uint64()) + out = append(out, KeyToHexNibbleHash(key)) + } + } + return out +} + +func insertAll(tr *prefixTrie, keys [][]byte) { + for _, k := range keys { + tr.Insert(k, k, nil) + } +} + +func BenchmarkPrefixTrieInsertOrder(b *testing.B) { + const n = touchChunkKeys + + for _, c := range []struct { + name string + keys [][]byte + }{ + {"accounts", orderBenchAccountKeys(n, 11)}, + {"storage-clustered", orderBenchStorageKeys(n, 500, 22)}, + } { + sorted := slices.Clone(c.keys) + slices.SortFunc(sorted, bytes.Compare) + + b.Run(fmt.Sprintf("%s/arrival", c.name), func(b *testing.B) { + tr := newPrefixTrie() + b.ReportAllocs() + for b.Loop() { + insertAll(tr, c.keys) + b.StopTimer() + tr.Reset() + b.StartTimer() + } + }) + + b.Run(fmt.Sprintf("%s/presorted", c.name), func(b *testing.B) { + tr := newPrefixTrie() + b.ReportAllocs() + for b.Loop() { + insertAll(tr, sorted) + b.StopTimer() + tr.Reset() + b.StartTimer() + } + }) + + b.Run(fmt.Sprintf("%s/sort+insert", c.name), func(b *testing.B) { + tr := newPrefixTrie() + scratch := make([][]byte, n) + b.ReportAllocs() + for b.Loop() { + copy(scratch, c.keys) + slices.SortFunc(scratch, bytes.Compare) + insertAll(tr, scratch) + b.StopTimer() + tr.Reset() + b.StartTimer() + } + }) + } +} From 5898cd3b6deb1d12a6bdd2fd7433b0a7fb63d4ce Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 23 Sep 2026 08:33:26 +0700 Subject: [PATCH 09/18] commitment: drop the chunk wrapper type, the buffer pool and the investigation benchmarks --- execution/commitment/dedup_bench_test.go | 106 ---------------- execution/commitment/parallel_chunk.go | 42 ------- ...chunk_test.go => parallel_collect_test.go} | 12 +- .../parallel_streaming_bench_test.go | 36 +++++- execution/commitment/parallel_update.go | 80 ++++++------- .../prefix_trie_order_bench_test.go | 113 ------------------ execution/commitment/presort_bench_test.go | 83 ------------- 7 files changed, 76 insertions(+), 396 deletions(-) delete mode 100644 execution/commitment/dedup_bench_test.go delete mode 100644 execution/commitment/parallel_chunk.go rename execution/commitment/{parallel_chunk_test.go => parallel_collect_test.go} (92%) delete mode 100644 execution/commitment/prefix_trie_order_bench_test.go delete mode 100644 execution/commitment/presort_bench_test.go diff --git a/execution/commitment/dedup_bench_test.go b/execution/commitment/dedup_bench_test.go deleted file mode 100644 index 7d4ffcd5306..00000000000 --- a/execution/commitment/dedup_bench_test.go +++ /dev/null @@ -1,106 +0,0 @@ -// 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 ( - "encoding/binary" - "math/rand" - "testing" - - "github.com/erigontech/erigon/common" - "github.com/erigontech/erigon/common/length" -) - -func dedupBenchKeys(n, keyLen int) []string { - rnd := rand.New(rand.NewSource(20260922)) - keys := make([]string, n) - for i := range keys { - k := make([]byte, keyLen) - for j := 0; j+8 <= keyLen; j += 8 { - binary.BigEndian.PutUint64(k[j:], rnd.Uint64()) - } - keys[i] = string(k) - } - return keys -} - -func BenchmarkDedupCost(b *testing.B) { - const n = 50_000 - - for _, c := range []struct { - name string - keyLen int - }{ - {"account", length.Addr}, - {"storage", length.Addr + length.Hash}, - } { - keys := dedupBenchKeys(n, c.keyLen) - - b.Run(c.name+"/mapHit", func(b *testing.B) { - seen := make(map[string]struct{}, n) - for _, k := range keys { - seen[k] = struct{}{} - } - b.ReportAllocs() - i := 0 - for b.Loop() { - if _, ok := seen[keys[i]]; !ok { - b.Fatal("key must be present") - } - i = (i + 1) % n - } - }) - - b.Run(c.name+"/hash", func(b *testing.B) { - b.ReportAllocs() - i := 0 - for b.Loop() { - sinkHashedKey = KeyToHexNibbleHash(common.ToBytesZeroCopy(keys[i])) - i = (i + 1) % n - } - }) - - b.Run(c.name+"/hashCached", func(b *testing.B) { - b.ReportAllocs() - var cache addrHashCache - i := 0 - for b.Loop() { - sinkHashedKey = keyToHexNibbleHashCached(common.ToBytesZeroCopy(keys[i]), &cache) - i = (i + 1) % n - } - }) - - b.Run(c.name+"/trieHit", func(b *testing.B) { - hashed := make([][]byte, n) - tr := newPrefixTrie() - for i, k := range keys { - hashed[i] = KeyToHexNibbleHash(common.ToBytesZeroCopy(k)) - tr.Insert(hashed[i], common.ToBytesZeroCopy(k), nil) - } - b.ReportAllocs() - i := 0 - for b.Loop() { - if tr.Insert(hashed[i], common.ToBytesZeroCopy(keys[i]), nil) { - b.Fatal("key must already be in the trie") - } - i = (i + 1) % n - } - }) - } -} - -var sinkHashedKey []byte diff --git a/execution/commitment/parallel_chunk.go b/execution/commitment/parallel_chunk.go deleted file mode 100644 index c02cd7e4f0f..00000000000 --- a/execution/commitment/parallel_chunk.go +++ /dev/null @@ -1,42 +0,0 @@ -// 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 - -type touchEntry struct { - hashedKey []byte - plainKey []byte - update *Update -} - -type touchChunk struct { - entries []touchEntry -} - -func (c *touchChunk) collect(hashedKey, plainKey []byte, update *Update) { - c.entries = append(c.entries, touchEntry{ - hashedKey: hashedKey, - plainKey: plainKey, - update: update, - }) -} - -func (c *touchChunk) count() int { return len(c.entries) } - -func (c *touchChunk) reset() { - clear(c.entries) - c.entries = c.entries[:0] -} diff --git a/execution/commitment/parallel_chunk_test.go b/execution/commitment/parallel_collect_test.go similarity index 92% rename from execution/commitment/parallel_chunk_test.go rename to execution/commitment/parallel_collect_test.go index 64cf31d6a6f..bbef0c4c76a 100644 --- a/execution/commitment/parallel_chunk_test.go +++ b/execution/commitment/parallel_collect_test.go @@ -161,23 +161,23 @@ func TestChunkBuild_BackgroundBuildKeepsMergeOrder(t *testing.T) { } } -func TestChunkBuild_BuffersReturnToThePoolAcrossBatches(t *testing.T) { +func TestChunkBuild_BuffersSurviveBatches(t *testing.T) { t.Parallel() cases := randomTouchCases(777, 200, 0) pu, backgrounded := buildChunked(cases, 8) require.True(t, backgrounded) - require.Len(t, pu.pool, touchChunkBuffers, "every chunk buffer must come back to the pool") + require.Len(t, pu.freeCh, touchChunkBuffers, "every chunk buffer must be back in the free list") pu.Reset() for _, c := range cases { pu.Collect(c.hashedKey, c.plainKey, nil) } pu.Build() - require.Len(t, pu.pool, touchChunkBuffers, "a second batch must reuse the pooled buffers") - for _, p := range pu.pool { - require.GreaterOrEqual(t, cap(p.entries), pu.chunkKeys, - "a pooled chunk buffer must keep its capacity, or every chunk re-grows it") + require.Len(t, pu.freeCh, touchChunkBuffers, "a second batch must reuse the same buffers") + for range touchChunkBuffers { + require.GreaterOrEqual(t, cap(<-pu.freeCh), pu.chunkKeys, + "a recycled buffer must keep its capacity, or every chunk re-grows it") } require.EqualValues(t, len(cases), pu.trie.root.subtreeCount) } diff --git a/execution/commitment/parallel_streaming_bench_test.go b/execution/commitment/parallel_streaming_bench_test.go index a4f4e4b47f3..01d0914b260 100644 --- a/execution/commitment/parallel_streaming_bench_test.go +++ b/execution/commitment/parallel_streaming_bench_test.go @@ -56,16 +56,22 @@ func runDirectBench(b *testing.B, pk [][]byte, updates []Update) { func runParallelBench(b *testing.B, pk [][]byte, updates []Update, workers int) { runParallelBenchWith(b, pk, updates, workers, - func(ms *MockState) (TrieContextFactory, func()) { return mockTrieCtxFactory(ms), func() {} }) + func(ms *MockState) (TrieContextFactory, func()) { return mockTrieCtxFactory(ms), func() {} }, false) } func runCollectingParallelBench(b *testing.B, pk [][]byte, updates []Update, workers int) { - runParallelBenchWith(b, pk, updates, workers, collectingTrieCtxFactory) + runParallelBenchWith(b, pk, updates, workers, collectingTrieCtxFactory, false) +} + +func runCollectAndProcessBench(b *testing.B, pk [][]byte, updates []Update, workers int) { + runParallelBenchWith(b, pk, updates, workers, + func(ms *MockState) (TrieContextFactory, func()) { return mockTrieCtxFactory(ms), func() {} }, true) } func runParallelBenchWith(b *testing.B, pk [][]byte, updates []Update, workers int, - newFactory func(*MockState) (TrieContextFactory, func())) { + newFactory func(*MockState) (TrieContextFactory, func()), timeCollect bool) { ctx := context.Background() + tmp := b.TempDir() b.ReportAllocs() var pph *ParallelPatriciaHashed defer func() { @@ -88,8 +94,14 @@ func runParallelBenchWith(b *testing.B, pk [][]byte, updates []Update, workers i } pph.RootTrie().Reset() upds := WrapKeyUpdates(b, ModeParallel, KeyToHexNibbleHash, pk, updates) + if timeCollect { + upds = NewUpdates(ModeParallel, tmp, KeyToHexNibbleHash) + } b.StartTimer() + if timeCollect { + WrapKeyUpdatesInto(b, upds, pk, updates) + } _, err := pph.Process(ctx, upds, "", nil, WarmupConfig{}) b.StopTimer() @@ -623,3 +635,21 @@ func collectingTrieCtxFactory(ms *MockState) (TrieContextFactory, func()) { } return f, drain } + +func Benchmark_ModeParallel_TouchAndProcess(b *testing.B) { + ncpu := runtime.NumCPU() + for _, c := range []struct { + name string + build func(testing.TB) ([][]byte, []Update) + }{ + {"100K-AccountsOnly", build100KAccountsCorpus}, + {"500K-StorageHeavy", build500KStorageHeavyCorpus}, + } { + pk, updates := c.build(b) + for _, w := range []int{4, ncpu} { + b.Run(fmt.Sprintf("%s/w%d", c.name, w), func(b *testing.B) { + runCollectAndProcessBench(b, pk, updates, w) + }) + } + } +} diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index 268e28640b8..2385e0498d5 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -48,14 +48,19 @@ const touchChunkKeys = 8192 const touchChunkBuffers = 2 +type touchEntry struct { + hashedKey []byte + plainKey []byte + update *Update +} + type parallelUpdate struct { trie *prefixTrie - pending *touchChunk + pending []touchEntry chunkKeys int - buildCh chan *touchChunk - freeCh chan *touchChunk - pool []*touchChunk + buildCh chan []touchEntry + freeCh chan []touchEntry inflight sync.WaitGroup deferredMu sync.Mutex @@ -67,58 +72,46 @@ type parallelUpdate struct { func newParallelUpdate() *parallelUpdate { return ¶llelUpdate{ trie: newPrefixTrie(), - pending: new(touchChunk), chunkKeys: touchChunkKeys, } } // Collect is not safe for concurrent calls; the caller must serialize them. func (pu *parallelUpdate) Collect(hashedKey, plainKey []byte, update *Update) { - pu.pending.collect(hashedKey, plainKey, update) - if pu.pending.count() >= pu.chunkKeys { + pu.pending = append(pu.pending, touchEntry{hashedKey: hashedKey, plainKey: plainKey, update: update}) + if len(pu.pending) >= pu.chunkKeys { pu.handOff() } } func (pu *parallelUpdate) startBuilder() { - pu.buildCh = make(chan *touchChunk, touchChunkBuffers) - pu.freeCh = make(chan *touchChunk, touchChunkBuffers) - for range touchChunkBuffers { - if n := len(pu.pool); n > 0 { - pu.freeCh <- pu.pool[n-1] - pu.pool = pu.pool[:n-1] - } else { - pu.freeCh <- &touchChunk{entries: make([]touchEntry, 0, pu.chunkKeys)} + if pu.freeCh == nil { + pu.freeCh = make(chan []touchEntry, touchChunkBuffers) + for range touchChunkBuffers { + pu.freeCh <- make([]touchEntry, 0, pu.chunkKeys) } } - go func(build <-chan *touchChunk, free chan<- *touchChunk) { - for p := range build { - pu.insertChunk(p) - p.reset() - free <- p + pu.buildCh = make(chan []touchEntry, touchChunkBuffers) + go func(build <-chan []touchEntry, free chan<- []touchEntry) { + for c := range build { + pu.insertChunk(c) + clear(c) + free <- c[:0] pu.inflight.Done() } }(pu.buildCh, pu.freeCh) } func (pu *parallelUpdate) stopBuilder() { - if pu.buildCh == nil { - return + if pu.buildCh != nil { + close(pu.buildCh) + pu.buildCh = nil } - close(pu.buildCh) - for range touchChunkBuffers { - select { - case p := <-pu.freeCh: - pu.pool = append(pu.pool, p) - default: - } - } - pu.buildCh, pu.freeCh = nil, nil } func (pu *parallelUpdate) handOff() { if pu.trie == nil { - pu.pending.reset() + pu.pending = pu.pending[:0] return } if pu.buildCh == nil { @@ -129,21 +122,20 @@ func (pu *parallelUpdate) handOff() { pu.pending = <-pu.freeCh } -func (pu *parallelUpdate) insertChunk(p *touchChunk) { - for i := range p.entries { - pu.trie.Insert(p.entries[i].hashedKey, p.entries[i].plainKey, p.entries[i].update) +func (pu *parallelUpdate) insertChunk(c []touchEntry) { + for i := range c { + pu.trie.Insert(c[i].hashedKey, c[i].plainKey, c[i].update) } } func (pu *parallelUpdate) Build() { - if pu.pending.count() > 0 { + if len(pu.pending) > 0 { if pu.buildCh != nil { pu.handOff() - } else if pu.trie != nil { - pu.insertChunk(pu.pending) - pu.pending.reset() } else { - pu.pending.reset() + pu.insertChunk(pu.pending) + clear(pu.pending) + pu.pending = pu.pending[:0] } } pu.inflight.Wait() @@ -166,7 +158,8 @@ func (pu *parallelUpdate) drainDeferred() { func (pu *parallelUpdate) Reset() { pu.inflight.Wait() pu.stopBuilder() - pu.pending.reset() + clear(pu.pending) + pu.pending = pu.pending[:0] if pu.trie != nil { pu.trie.Reset() } @@ -177,8 +170,9 @@ func (pu *parallelUpdate) Reset() { func (pu *parallelUpdate) Close() { pu.inflight.Wait() pu.stopBuilder() - pu.pool = nil - pu.pending.reset() + pu.freeCh = nil + clear(pu.pending) + pu.pending = nil pu.trie = nil pu.drainDeferred() pu.keyArena.reset() diff --git a/execution/commitment/prefix_trie_order_bench_test.go b/execution/commitment/prefix_trie_order_bench_test.go deleted file mode 100644 index 88a25a43926..00000000000 --- a/execution/commitment/prefix_trie_order_bench_test.go +++ /dev/null @@ -1,113 +0,0 @@ -// 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 ( - "bytes" - "encoding/binary" - "fmt" - "math/rand" - "slices" - "testing" - - "github.com/erigontech/erigon/common/length" -) - -func orderBenchAccountKeys(n int, seed int64) [][]byte { - rnd := rand.New(rand.NewSource(seed)) - out := make([][]byte, n) - addr := make([]byte, length.Addr) - for i := range out { - binary.BigEndian.PutUint64(addr, rnd.Uint64()) - binary.BigEndian.PutUint64(addr[8:], rnd.Uint64()) - out[i] = KeyToHexNibbleHash(addr) - } - return out -} - -func orderBenchStorageKeys(n, slotsPerAddr int, seed int64) [][]byte { - rnd := rand.New(rand.NewSource(seed)) - out := make([][]byte, 0, n) - key := make([]byte, length.Addr+length.Hash) - for len(out) < n { - binary.BigEndian.PutUint64(key, rnd.Uint64()) - binary.BigEndian.PutUint64(key[8:], rnd.Uint64()) - for s := 0; s < slotsPerAddr && len(out) < n; s++ { - binary.BigEndian.PutUint64(key[length.Addr:], rnd.Uint64()) - binary.BigEndian.PutUint64(key[length.Addr+8:], rnd.Uint64()) - out = append(out, KeyToHexNibbleHash(key)) - } - } - return out -} - -func insertAll(tr *prefixTrie, keys [][]byte) { - for _, k := range keys { - tr.Insert(k, k, nil) - } -} - -func BenchmarkPrefixTrieInsertOrder(b *testing.B) { - const n = touchChunkKeys - - for _, c := range []struct { - name string - keys [][]byte - }{ - {"accounts", orderBenchAccountKeys(n, 11)}, - {"storage-clustered", orderBenchStorageKeys(n, 500, 22)}, - } { - sorted := slices.Clone(c.keys) - slices.SortFunc(sorted, bytes.Compare) - - b.Run(fmt.Sprintf("%s/arrival", c.name), func(b *testing.B) { - tr := newPrefixTrie() - b.ReportAllocs() - for b.Loop() { - insertAll(tr, c.keys) - b.StopTimer() - tr.Reset() - b.StartTimer() - } - }) - - b.Run(fmt.Sprintf("%s/presorted", c.name), func(b *testing.B) { - tr := newPrefixTrie() - b.ReportAllocs() - for b.Loop() { - insertAll(tr, sorted) - b.StopTimer() - tr.Reset() - b.StartTimer() - } - }) - - b.Run(fmt.Sprintf("%s/sort+insert", c.name), func(b *testing.B) { - tr := newPrefixTrie() - scratch := make([][]byte, n) - b.ReportAllocs() - for b.Loop() { - copy(scratch, c.keys) - slices.SortFunc(scratch, bytes.Compare) - insertAll(tr, scratch) - b.StopTimer() - tr.Reset() - b.StartTimer() - } - }) - } -} diff --git a/execution/commitment/presort_bench_test.go b/execution/commitment/presort_bench_test.go deleted file mode 100644 index c9a3cc105f9..00000000000 --- a/execution/commitment/presort_bench_test.go +++ /dev/null @@ -1,83 +0,0 @@ -// 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 ( - "context" - "fmt" - "runtime" - "testing" - - "github.com/stretchr/testify/require" - - "github.com/erigontech/erigon/common/length" -) - -func benchTouchAndProcess(b *testing.B, pk [][]byte, updates []Update, workers int) { - ctx := context.Background() - tmp := b.TempDir() - b.ReportAllocs() - var pph *ParallelPatriciaHashed - defer func() { - if pph != nil { - pph.Release() - } - }() - for b.Loop() { - b.StopTimer() - ms := NewMockState(b) - ms.SetConcurrentCommitment(true) - require.NoError(b, ms.applyPlainUpdates(pk, updates)) - factory := mockTrieCtxFactory(ms) - if pph == nil { - pph = NewParallelPatriciaHashed(factory, length.Addr, DefaultTrieConfig()) - pph.SetNumWorkers(workers) - } else { - pph.SetTrieContextFactory(factory) - pph.ResetContext(ms) - } - pph.RootTrie().Reset() - upds := NewUpdates(ModeParallel, tmp, KeyToHexNibbleHash) - b.StartTimer() - - WrapKeyUpdatesInto(b, upds, pk, updates) - _, err := pph.Process(ctx, upds, "", nil, WarmupConfig{}) - - b.StopTimer() - require.NoError(b, err) - upds.Close() - b.StartTimer() - } -} - -func Benchmark_ModeParallel_TouchAndProcess(b *testing.B) { - ncpu := runtime.NumCPU() - for _, c := range []struct { - name string - build func(testing.TB) ([][]byte, []Update) - }{ - {"100K-AccountsOnly", build100KAccountsCorpus}, - {"500K-StorageHeavy", build500KStorageHeavyCorpus}, - } { - pk, updates := c.build(b) - for _, w := range []int{4, ncpu} { - b.Run(fmt.Sprintf("%s/w%d", c.name, w), func(b *testing.B) { - benchTouchAndProcess(b, pk, updates, w) - }) - } - } -} From d3751a9ace93bc79d54d5e20ff1cf53ca93aa17e Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 23 Sep 2026 08:38:23 +0700 Subject: [PATCH 10/18] commitment: fold builder drain into one helper, drop the test-only chunk threshold and the unreachable nil-trie guard --- execution/commitment/parallel_collect_test.go | 33 +++++------------- execution/commitment/parallel_update.go | 34 +++++++------------ 2 files changed, 21 insertions(+), 46 deletions(-) diff --git a/execution/commitment/parallel_collect_test.go b/execution/commitment/parallel_collect_test.go index bbef0c4c76a..b18eb64989c 100644 --- a/execution/commitment/parallel_collect_test.go +++ b/execution/commitment/parallel_collect_test.go @@ -116,11 +116,8 @@ func buildReferenceTrie(cases []touchCase) *prefixTrie { return tr } -func buildChunked(cases []touchCase, chunkKeys int) (*parallelUpdate, bool) { +func buildChunked(cases []touchCase) (*parallelUpdate, bool) { pu := newParallelUpdate() - if chunkKeys > 0 { - pu.chunkKeys = chunkKeys - } for _, c := range cases { var upd *Update if c.update != nil { @@ -137,35 +134,23 @@ func buildChunked(cases []touchCase, chunkKeys int) (*parallelUpdate, bool) { func TestChunkBuild_MatchesInsertionOrderTrie(t *testing.T) { t.Parallel() - for _, n := range []int{1, 2, 17, 5000} { + for _, n := range []int{1, 2, 17, 5000, 8200} { cases := randomTouchCases(int64(n)*7919+3, n, 4) want := flattenPrefixTrie(t, buildReferenceTrie(cases)) - pu, backgrounded := buildChunked(cases, 1<<20) - require.False(t, backgrounded, "n=%d must stay under the hand-off threshold", n) - require.Equal(t, want, flattenPrefixTrie(t, pu.trie), - "presorted build must match insertion-order build for n=%d", n) - } -} - -func TestChunkBuild_BackgroundBuildKeepsMergeOrder(t *testing.T) { - t.Parallel() - - for _, n := range []int{1, 2, 17, 3000} { - cases := randomTouchCases(int64(n)*4242+1, n, 3) - want := flattenPrefixTrie(t, buildReferenceTrie(cases)) - pu, backgrounded := buildChunked(cases, 8) - require.Equal(t, n > 8, backgrounded, "n=%d must cross the hand-off threshold", n) + pu, backgrounded := buildChunked(cases) + require.Equal(t, len(cases) > touchChunkKeys, backgrounded, + "n=%d: hand-off must trigger exactly when the batch outgrows one chunk", n) require.Nil(t, pu.buildCh, "Build must reap the builder goroutine") require.Equal(t, want, flattenPrefixTrie(t, pu.trie), - "chunks built in the background must not change the merge order for n=%d", n) + "chunked build must match insertion-order build for n=%d", n) } } func TestChunkBuild_BuffersSurviveBatches(t *testing.T) { t.Parallel() - cases := randomTouchCases(777, 200, 0) - pu, backgrounded := buildChunked(cases, 8) + cases := randomTouchCases(777, touchChunkKeys+8, 0) + pu, backgrounded := buildChunked(cases) require.True(t, backgrounded) require.Len(t, pu.freeCh, touchChunkBuffers, "every chunk buffer must be back in the free list") @@ -176,7 +161,7 @@ func TestChunkBuild_BuffersSurviveBatches(t *testing.T) { pu.Build() require.Len(t, pu.freeCh, touchChunkBuffers, "a second batch must reuse the same buffers") for range touchChunkBuffers { - require.GreaterOrEqual(t, cap(<-pu.freeCh), pu.chunkKeys, + require.GreaterOrEqual(t, cap(<-pu.freeCh), touchChunkKeys, "a recycled buffer must keep its capacity, or every chunk re-grows it") } require.EqualValues(t, len(cases), pu.trie.root.subtreeCount) diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index 2385e0498d5..1f7ee9f5f59 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -57,11 +57,10 @@ type touchEntry struct { type parallelUpdate struct { trie *prefixTrie - pending []touchEntry - chunkKeys int - buildCh chan []touchEntry - freeCh chan []touchEntry - inflight sync.WaitGroup + pending []touchEntry + buildCh chan []touchEntry + freeCh chan []touchEntry + inflight sync.WaitGroup deferredMu sync.Mutex deferredCombined []*DeferredBranchUpdate @@ -70,16 +69,13 @@ type parallelUpdate struct { } func newParallelUpdate() *parallelUpdate { - return ¶llelUpdate{ - trie: newPrefixTrie(), - chunkKeys: touchChunkKeys, - } + return ¶llelUpdate{trie: newPrefixTrie()} } // Collect is not safe for concurrent calls; the caller must serialize them. func (pu *parallelUpdate) Collect(hashedKey, plainKey []byte, update *Update) { pu.pending = append(pu.pending, touchEntry{hashedKey: hashedKey, plainKey: plainKey, update: update}) - if len(pu.pending) >= pu.chunkKeys { + if len(pu.pending) >= touchChunkKeys { pu.handOff() } } @@ -88,7 +84,7 @@ func (pu *parallelUpdate) startBuilder() { if pu.freeCh == nil { pu.freeCh = make(chan []touchEntry, touchChunkBuffers) for range touchChunkBuffers { - pu.freeCh <- make([]touchEntry, 0, pu.chunkKeys) + pu.freeCh <- make([]touchEntry, 0, touchChunkKeys) } } pu.buildCh = make(chan []touchEntry, touchChunkBuffers) @@ -102,7 +98,8 @@ func (pu *parallelUpdate) startBuilder() { }(pu.buildCh, pu.freeCh) } -func (pu *parallelUpdate) stopBuilder() { +func (pu *parallelUpdate) drainBuilder() { + pu.inflight.Wait() if pu.buildCh != nil { close(pu.buildCh) pu.buildCh = nil @@ -110,10 +107,6 @@ func (pu *parallelUpdate) stopBuilder() { } func (pu *parallelUpdate) handOff() { - if pu.trie == nil { - pu.pending = pu.pending[:0] - return - } if pu.buildCh == nil { pu.startBuilder() } @@ -138,8 +131,7 @@ func (pu *parallelUpdate) Build() { pu.pending = pu.pending[:0] } } - pu.inflight.Wait() - pu.stopBuilder() + pu.drainBuilder() } func (pu *parallelUpdate) internKey(plainKey []byte) []byte { @@ -156,8 +148,7 @@ func (pu *parallelUpdate) drainDeferred() { } func (pu *parallelUpdate) Reset() { - pu.inflight.Wait() - pu.stopBuilder() + pu.drainBuilder() clear(pu.pending) pu.pending = pu.pending[:0] if pu.trie != nil { @@ -168,8 +159,7 @@ func (pu *parallelUpdate) Reset() { } func (pu *parallelUpdate) Close() { - pu.inflight.Wait() - pu.stopBuilder() + pu.drainBuilder() pu.freeCh = nil clear(pu.pending) pu.pending = nil From 4a13d50f4425a538299ca8407b5fa543db79736d Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 23 Sep 2026 08:44:33 +0700 Subject: [PATCH 11/18] commitment: inline the chunk builder start, drop defensive update copies and pin the trie flattener --- execution/commitment/parallel_collect_test.go | 43 +++++++++--------- execution/commitment/parallel_update.go | 44 +++++++++---------- 2 files changed, 42 insertions(+), 45 deletions(-) diff --git a/execution/commitment/parallel_collect_test.go b/execution/commitment/parallel_collect_test.go index b18eb64989c..e6e3d6d9f23 100644 --- a/execution/commitment/parallel_collect_test.go +++ b/execution/commitment/parallel_collect_test.go @@ -17,6 +17,7 @@ package commitment import ( + "math/bits" "math/rand" "slices" "testing" @@ -56,17 +57,10 @@ func flattenPrefixTrie(t *testing.T, tr *prefixTrie) []flatNode { } func nthSetNibble(bitmap uint16, n int) byte { - seen := 0 - for i := range 16 { - if bitmap&(1< Date: Wed, 23 Sep 2026 10:42:52 +0700 Subject: [PATCH 12/18] commitment: chunk ModeParallel touches at 16384 keys and drain the builder inside the touch benchmark --- execution/commitment/parallel_update.go | 2 +- execution/commitment/touch_collect_bench_test.go | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index 8187a3409dd..0b7ac01629a 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -44,7 +44,7 @@ func (a *plainKeyArena) intern(b []byte) []byte { func (a *plainKeyArena) reset() { a.buf = a.buf[:0] } -const touchChunkKeys = 8192 +const touchChunkKeys = 16384 const touchChunkBuffers = 2 diff --git a/execution/commitment/touch_collect_bench_test.go b/execution/commitment/touch_collect_bench_test.go index 5c82ef22e97..df2446be11e 100644 --- a/execution/commitment/touch_collect_bench_test.go +++ b/execution/commitment/touch_collect_bench_test.go @@ -65,6 +65,9 @@ func BenchmarkTouchDirect(b *testing.B) { Balance: *uint256.NewInt(uint64(i)), }) } + if ut.parallel != nil { + ut.parallel.Build() + } b.StopTimer() ut.Reset() b.StartTimer() From 242c5201b7546516d93792b0f9cec28deb97f16d Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 23 Sep 2026 11:53:37 +0700 Subject: [PATCH 13/18] commitment: stop the collect benchmark discarding a populated Updates and leaking its builder --- execution/commitment/commitment_test.go | 35 +++++++------------ execution/commitment/parallel_collect_test.go | 31 ++++++++++++++++ .../parallel_streaming_bench_test.go | 7 ++-- 3 files changed, 48 insertions(+), 25 deletions(-) diff --git a/execution/commitment/commitment_test.go b/execution/commitment/commitment_test.go index ff75568693d..f63c6170c70 100644 --- a/execution/commitment/commitment_test.go +++ b/execution/commitment/commitment_test.go @@ -865,6 +865,13 @@ func TestUpdatesModeParallel_NewAllocates(t *testing.T) { require.Equal(t, uint64(0), ut.Size()) } +func requireParallelCount(t *testing.T, ut *Updates, n int, msg string) { + t.Helper() + require.Equal(t, uint64(n), ut.Size()) + ut.parallel.Build() + require.EqualValues(t, n, ut.parallel.trie.root.subtreeCount, msg) +} + func TestUpdatesModeParallel_TouchPlainKeyRoutes(t *testing.T) { t.Parallel() @@ -881,17 +888,10 @@ func TestUpdatesModeParallel_TouchPlainKeyRoutes(t *testing.T) { ut.TouchPlainKey(string(k), []byte("v"), ut.TouchStorage) } - require.Equal(t, uint64(len(keys)), ut.Size()) - ut.parallel.Build() - require.NotNil(t, ut.parallel.trie.root) - require.EqualValues(t, len(keys), ut.parallel.trie.root.subtreeCount, - "every touched key must show up in the prefix trie") + requireParallelCount(t, ut, len(keys), "every touched key must show up in the prefix trie") ut.TouchPlainKey(string(keys[0]), []byte("v2"), ut.TouchStorage) - require.Equal(t, uint64(len(keys)), ut.Size()) - ut.parallel.Build() - require.EqualValues(t, len(keys), ut.parallel.trie.root.subtreeCount, - "duplicate TouchPlainKey must not double-count in the trie") + requireParallelCount(t, ut, len(keys), "duplicate TouchPlainKey must not double-count in the trie") } func TestUpdatesModeParallel_RepeatTouchDoesNotRehash(t *testing.T) { @@ -913,10 +913,7 @@ func TestUpdatesModeParallel_RepeatTouchDoesNotRehash(t *testing.T) { } require.EqualValues(t, 1, hashCalls.Load(), "a repeat touch must not rehash the key") - require.Equal(t, uint64(1), ut.Size()) - ut.parallel.Build() - require.NotNil(t, ut.parallel.trie.root) - require.EqualValues(t, 1, ut.parallel.trie.root.subtreeCount) + requireParallelCount(t, ut, 1, "the single touched key must reach the trie") } func TestUpdatesModeParallel_TouchHashedKey(t *testing.T) { @@ -932,9 +929,7 @@ func TestUpdatesModeParallel_TouchHashedKey(t *testing.T) { ut.TouchHashedKey(hk2) ut.TouchHashedKey(hk1) - require.Equal(t, uint64(2), ut.Size()) - ut.parallel.Build() - require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount) + requireParallelCount(t, ut, 2, "a repeated hashed key must not double-count in the trie") } func TestUpdatesModeParallel_Reset(t *testing.T) { @@ -950,9 +945,7 @@ func TestUpdatesModeParallel_Reset(t *testing.T) { for _, k := range keys { ut.TouchPlainKey(string(k), []byte("v"), ut.TouchStorage) } - require.Equal(t, uint64(2), ut.Size()) - ut.parallel.Build() - require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount) + requireParallelCount(t, ut, 2, "both touched keys must reach the trie") ut.Reset() @@ -966,9 +959,7 @@ func TestUpdatesModeParallel_Reset(t *testing.T) { for _, k := range keys { ut.TouchPlainKey(string(k), []byte("v"), ut.TouchStorage) } - require.Equal(t, uint64(2), ut.Size()) - ut.parallel.Build() - require.EqualValues(t, 2, ut.parallel.trie.root.subtreeCount) + requireParallelCount(t, ut, 2, "a collection reused after Reset must refill the trie") } func TestUpdatesModeParallel_Close(t *testing.T) { diff --git a/execution/commitment/parallel_collect_test.go b/execution/commitment/parallel_collect_test.go index e6e3d6d9f23..3356c1e9599 100644 --- a/execution/commitment/parallel_collect_test.go +++ b/execution/commitment/parallel_collect_test.go @@ -130,6 +130,37 @@ func TestFlattenPrefixTrie_Paths(t *testing.T) { require.Equal(t, []string{"", "12", "123", "12f", "5"}, got) } +func TestChunkBuild_WalkOrderIsSortedWhateverTheArrivalOrder(t *testing.T) { + t.Parallel() + + cases := randomTouchCases(31337, 2000, 0) + shuffled := slices.Clone(cases) + rnd := rand.New(rand.NewSource(9)) + rnd.Shuffle(len(shuffled), func(i, j int) { shuffled[i], shuffled[j] = shuffled[j], shuffled[i] }) + + pu, _ := buildChunked(shuffled) + flat := flattenPrefixTrie(t, pu.trie) + + paths := make([]string, 0, len(flat)) + terminals := make([]string, 0, len(cases)) + for _, n := range flat { + paths = append(paths, n.path) + if n.plainKey != "" { + terminals = append(terminals, n.path) + } + } + require.True(t, slices.IsSorted(paths), "the walk must emit nodes in hashed-key order") + + want := make([]string, 0, len(cases)) + for _, c := range cases { + if c.plainKey != nil { + want = append(want, NibblesToString(c.hashedKey)) + } + } + slices.Sort(want) + require.Equal(t, want, terminals, "every collected key must appear once, in sorted order") +} + func TestChunkBuild_MatchesInsertionOrderTrie(t *testing.T) { t.Parallel() diff --git a/execution/commitment/parallel_streaming_bench_test.go b/execution/commitment/parallel_streaming_bench_test.go index 01d0914b260..112a0797e1a 100644 --- a/execution/commitment/parallel_streaming_bench_test.go +++ b/execution/commitment/parallel_streaming_bench_test.go @@ -71,7 +71,6 @@ func runCollectAndProcessBench(b *testing.B, pk [][]byte, updates []Update, work func runParallelBenchWith(b *testing.B, pk [][]byte, updates []Update, workers int, newFactory func(*MockState) (TrieContextFactory, func()), timeCollect bool) { ctx := context.Background() - tmp := b.TempDir() b.ReportAllocs() var pph *ParallelPatriciaHashed defer func() { @@ -93,9 +92,11 @@ func runParallelBenchWith(b *testing.B, pk [][]byte, updates []Update, workers i pph.ResetContext(ms) } pph.RootTrie().Reset() - upds := WrapKeyUpdates(b, ModeParallel, KeyToHexNibbleHash, pk, updates) + var upds *Updates if timeCollect { - upds = NewUpdates(ModeParallel, tmp, KeyToHexNibbleHash) + upds = NewUpdates(ModeParallel, "", KeyToHexNibbleHash) + } else { + upds = WrapKeyUpdates(b, ModeParallel, KeyToHexNibbleHash, pk, updates) } b.StartTimer() From 127478c62d65d7eee8d5bde2cca8323105ef0bd4 Mon Sep 17 00:00:00 2001 From: awskii Date: Wed, 23 Sep 2026 12:21:14 +0700 Subject: [PATCH 14/18] commitment: cross the chunk threshold in the differential test and drain the builder on SetMode --- execution/commitment/commitment.go | 3 +++ execution/commitment/parallel_collect_test.go | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go index 3019348b774..9aec43e243e 100644 --- a/execution/commitment/commitment.go +++ b/execution/commitment/commitment.go @@ -1381,6 +1381,9 @@ func NewUpdates(m Mode, tmpdir string, hasher keyHasher) *Updates { } func (t *Updates) SetMode(m Mode) { + if t.parallel != nil { + t.parallel.Reset() + } t.mode = m switch t.mode { case ModeDirect: diff --git a/execution/commitment/parallel_collect_test.go b/execution/commitment/parallel_collect_test.go index 3356c1e9599..e2d63d7dddf 100644 --- a/execution/commitment/parallel_collect_test.go +++ b/execution/commitment/parallel_collect_test.go @@ -164,7 +164,7 @@ func TestChunkBuild_WalkOrderIsSortedWhateverTheArrivalOrder(t *testing.T) { func TestChunkBuild_MatchesInsertionOrderTrie(t *testing.T) { t.Parallel() - for _, n := range []int{1, 2, 17, 5000, 8200} { + for _, n := range []int{1, 2, 17, 5000, touchChunkKeys + 1, 3*touchChunkKeys + 7} { cases := randomTouchCases(int64(n)*7919+3, n, 4) want := flattenPrefixTrie(t, buildReferenceTrie(cases)) pu, backgrounded := buildChunked(cases) From 789aa098535c893d1235494728147de69efb67d8 Mon Sep 17 00:00:00 2001 From: awskii Date: Mon, 12 Oct 2026 13:03:05 +0700 Subject: [PATCH 15/18] commitment: give each touch chunk its own goroutine so no builder outlives its work; tighten chunk tests --- execution/commitment/parallel_collect_test.go | 158 +++++++----------- execution/commitment/parallel_update.go | 59 ++----- execution/commitment/prefix_trie_test.go | 4 +- 3 files changed, 74 insertions(+), 147 deletions(-) diff --git a/execution/commitment/parallel_collect_test.go b/execution/commitment/parallel_collect_test.go index e2d63d7dddf..5775e3fd1ad 100644 --- a/execution/commitment/parallel_collect_test.go +++ b/execution/commitment/parallel_collect_test.go @@ -17,61 +17,20 @@ package commitment import ( - "math/bits" + "bytes" "math/rand" "slices" "testing" + "testing/synctest" + "unsafe" "github.com/holiman/uint256" "github.com/stretchr/testify/require" ) -type flatNode struct { - path string - plainKey string - update string - subtree uint32 -} - -func flattenPrefixTrie(t *testing.T, tr *prefixTrie) []flatNode { - t.Helper() - var out []flatNode - var walk func(n *prefixNode, path []byte) - walk = func(n *prefixNode, path []byte) { - path = append(path, n.ext...) - rec := flatNode{path: NibblesToString(path), subtree: n.subtreeCount} - if n.plainKey != nil { - rec.plainKey = string(n.plainKey) - } - if n.update != nil { - rec.update = n.update.String() - } - out = append(out, rec) - for i, c := range n.children { - nib := nthSetNibble(n.bitmap, i) - walk(c, append(slices.Clone(path), nib)) - } - } - walk(tr.root, nil) - return out -} - -func nthSetNibble(bitmap uint16, n int) byte { - for range n { - bitmap &= bitmap - 1 - } - return byte(bits.TrailingZeros16(bitmap)) -} - -type touchCase struct { - hashedKey []byte - plainKey []byte - update *Update -} - -func randomTouchCases(seed int64, n, dupEvery int) []touchCase { +func randomTouchCases(seed int64, n, dupEvery int) []touchEntry { rnd := rand.New(rand.NewSource(seed)) - cases := make([]touchCase, 0, n) + cases := make([]touchEntry, 0, n) for i := range n { hk := make([]byte, 64) for j := range hk { @@ -85,9 +44,9 @@ func randomTouchCases(seed int64, n, dupEvery int) []touchCase { if i%3 != 0 { upd = &Update{Flags: BalanceUpdate, Balance: *uint256.NewInt(uint64(i) + 1)} } - cases = append(cases, touchCase{hashedKey: hk, plainKey: pk, update: upd}) + cases = append(cases, touchEntry{hashedKey: hk, plainKey: pk, update: upd}) if dupEvery > 0 && i%dupEvery == 0 { - cases = append(cases, touchCase{ + cases = append(cases, touchEntry{ hashedKey: slices.Clone(hk), plainKey: pk, update: &Update{Flags: NonceUpdate, Nonce: uint64(i) + 1}, @@ -97,7 +56,7 @@ func randomTouchCases(seed int64, n, dupEvery int) []touchCase { return cases } -func buildReferenceTrie(cases []touchCase) *prefixTrie { +func buildReferenceTrie(cases []touchEntry) *prefixTrie { tr := newPrefixTrie() for _, c := range cases { tr.Insert(c.hashedKey, c.plainKey, c.update) @@ -105,73 +64,59 @@ func buildReferenceTrie(cases []touchCase) *prefixTrie { return tr } -func buildChunked(cases []touchCase) (*parallelUpdate, bool) { - pu := newParallelUpdate() +func collectAll(pu *parallelUpdate, cases []touchEntry) { for _, c := range cases { pu.Collect(c.hashedKey, c.plainKey, c.update) } - backgrounded := pu.buildCh != nil - pu.Build() - return pu, backgrounded } -func TestFlattenPrefixTrie_Paths(t *testing.T) { - t.Parallel() - - tr := newPrefixTrie() - for _, hk := range [][]byte{nibs(1, 2, 3), nibs(1, 2, 0xf), nibs(5)} { - tr.Insert(hk, hk, nil) - } - - var got []string - for _, n := range flattenPrefixTrie(t, tr) { - got = append(got, n.path) - } - require.Equal(t, []string{"", "12", "123", "12f", "5"}, got) +func buildChunked(cases []touchEntry) (*parallelUpdate, bool) { + pu := newParallelUpdate() + collectAll(pu, cases) + backgrounded := pu.spare != nil + pu.Build() + return pu, backgrounded } func TestChunkBuild_WalkOrderIsSortedWhateverTheArrivalOrder(t *testing.T) { t.Parallel() - cases := randomTouchCases(31337, 2000, 0) + cases := randomTouchCases(31337, 2*touchChunkKeys+3, 0) shuffled := slices.Clone(cases) rnd := rand.New(rand.NewSource(9)) rnd.Shuffle(len(shuffled), func(i, j int) { shuffled[i], shuffled[j] = shuffled[j], shuffled[i] }) - pu, _ := buildChunked(shuffled) - flat := flattenPrefixTrie(t, pu.trie) + pu, backgrounded := buildChunked(shuffled) + require.True(t, backgrounded, "the batch must cross a chunk boundary") + walk := collectWalk(pu.trie) + require.True(t, slices.IsSortedFunc(walk, func(a, b walkEntry) int { return bytes.Compare(a.prefix, b.prefix) }), + "the walk must emit nodes in hashed-key order") - paths := make([]string, 0, len(flat)) - terminals := make([]string, 0, len(cases)) - for _, n := range flat { - paths = append(paths, n.path) - if n.plainKey != "" { - terminals = append(terminals, n.path) + var terminals, want [][]byte + for _, e := range walk { + if e.plainKey != nil { + terminals = append(terminals, e.prefix) } } - require.True(t, slices.IsSorted(paths), "the walk must emit nodes in hashed-key order") - - want := make([]string, 0, len(cases)) for _, c := range cases { if c.plainKey != nil { - want = append(want, NibblesToString(c.hashedKey)) + want = append(want, c.hashedKey) } } - slices.Sort(want) + slices.SortFunc(want, bytes.Compare) require.Equal(t, want, terminals, "every collected key must appear once, in sorted order") } func TestChunkBuild_MatchesInsertionOrderTrie(t *testing.T) { t.Parallel() - for _, n := range []int{1, 2, 17, 5000, touchChunkKeys + 1, 3*touchChunkKeys + 7} { - cases := randomTouchCases(int64(n)*7919+3, n, 4) - want := flattenPrefixTrie(t, buildReferenceTrie(cases)) + for _, n := range []int{1, 2, 17, 5000, touchChunkKeys - 1, touchChunkKeys, touchChunkKeys + 1, 3*touchChunkKeys + 7} { + cases := randomTouchCases(int64(n)*7919+3, n, 4)[:n] + want := collectWalk(buildReferenceTrie(cases)) pu, backgrounded := buildChunked(cases) - require.Equal(t, len(cases) > touchChunkKeys, backgrounded, - "n=%d: hand-off must trigger exactly when the batch outgrows one chunk", n) - require.Nil(t, pu.buildCh, "Build must reap the builder goroutine") - require.Equal(t, want, flattenPrefixTrie(t, pu.trie), + require.Equal(t, n >= touchChunkKeys, backgrounded, + "n=%d: hand-off must trigger exactly when the batch fills a chunk", n) + require.Equal(t, want, collectWalk(pu.trie), "chunked build must match insertion-order build for n=%d", n) } } @@ -180,21 +125,23 @@ func TestChunkBuild_BuffersSurviveBatches(t *testing.T) { t.Parallel() cases := randomTouchCases(777, touchChunkKeys+8, 0) - pu, backgrounded := buildChunked(cases) - require.True(t, backgrounded) - require.Len(t, pu.freeCh, touchChunkBuffers, "every chunk buffer must be back in the free list") - - pu.Reset() - for _, c := range cases { - pu.Collect(c.hashedKey, c.plainKey, nil) + pu := newParallelUpdate() + buffers := func() []*touchEntry { + return []*touchEntry{unsafe.SliceData(pu.pending), unsafe.SliceData(pu.spare)} } - pu.Build() - require.Len(t, pu.freeCh, touchChunkBuffers, "a second batch must reuse the same buffers") - for range touchChunkBuffers { - require.GreaterOrEqual(t, cap(<-pu.freeCh), touchChunkKeys, - "a recycled buffer must keep its capacity, or every chunk re-grows it") + var first []*touchEntry + for batch := range 3 { + pu.Reset() + collectAll(pu, cases) + pu.Build() + require.EqualValues(t, len(cases), pu.trie.root.subtreeCount) + if batch == 1 { + first = buffers() + } } - require.EqualValues(t, len(cases), pu.trie.root.subtreeCount) + require.GreaterOrEqual(t, cap(pu.pending), touchChunkKeys, "a recycled buffer must keep its capacity") + require.GreaterOrEqual(t, cap(pu.spare), touchChunkKeys, "a recycled buffer must keep its capacity") + require.ElementsMatch(t, first, buffers(), "later batches must reuse the same two buffers") } func TestChunkBuild_TouchHashedKeyCopiesCallerBuffer(t *testing.T) { @@ -217,3 +164,12 @@ func TestChunkBuild_TouchHashedKeyCopiesCallerBuffer(t *testing.T) { "a reused caller buffer must not collapse two keys into one") require.Len(t, ut.parallel.trie.root.children, 2) } + +func TestChunkBuild_AbandonedCollectionLeavesNoGoroutine(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + collectAll(newParallelUpdate(), randomTouchCases(5, touchChunkKeys, 0)) + synctest.Wait() + }) +} diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index a2564152480..5df23429560 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -48,8 +48,6 @@ func (a *plainKeyArena) reset() { a.buf = a.buf[:0] } const touchChunkKeys = 16384 -const touchChunkBuffers = 2 - type touchEntry struct { hashedKey []byte plainKey []byte @@ -60,8 +58,7 @@ type parallelUpdate struct { trie *prefixTrie pending []touchEntry - buildCh chan []touchEntry - freeCh chan []touchEntry + spare []touchEntry inflight sync.WaitGroup deferredMu sync.Mutex @@ -82,39 +79,19 @@ func (pu *parallelUpdate) Collect(hashedKey, plainKey []byte, update *Update) { } } -func (pu *parallelUpdate) drainBuilder() { - pu.inflight.Wait() - if pu.buildCh != nil { - close(pu.buildCh) - pu.buildCh = nil - } -} - func resetEntries(c []touchEntry) []touchEntry { clear(c) return c[:0] } func (pu *parallelUpdate) handOff() { - if pu.buildCh == nil { - if pu.freeCh == nil { - pu.freeCh = make(chan []touchEntry, touchChunkBuffers) - for range touchChunkBuffers { - pu.freeCh <- make([]touchEntry, 0, touchChunkKeys) - } - } - pu.buildCh = make(chan []touchEntry, touchChunkBuffers) - go func(build <-chan []touchEntry, free chan<- []touchEntry) { - for c := range build { - pu.insertChunk(c) - free <- resetEntries(c) - pu.inflight.Done() - } - }(pu.buildCh, pu.freeCh) - } - pu.inflight.Add(1) - pu.buildCh <- pu.pending - pu.pending = <-pu.freeCh + pu.inflight.Wait() + c := pu.pending + pu.pending, pu.spare = pu.spare[:0], c + pu.inflight.Go(func() { + pu.insertChunk(c) + clear(c) + }) } func (pu *parallelUpdate) insertChunk(c []touchEntry) { @@ -124,15 +101,9 @@ func (pu *parallelUpdate) insertChunk(c []touchEntry) { } func (pu *parallelUpdate) Build() { - if len(pu.pending) > 0 { - if pu.buildCh != nil { - pu.handOff() - } else { - pu.insertChunk(pu.pending) - pu.pending = resetEntries(pu.pending) - } - } - pu.drainBuilder() + pu.inflight.Wait() + pu.insertChunk(pu.pending) + pu.pending = resetEntries(pu.pending) } func (pu *parallelUpdate) internKey(plainKey []byte) []byte { @@ -149,7 +120,7 @@ func (pu *parallelUpdate) drainDeferred() { } func (pu *parallelUpdate) Reset() { - pu.drainBuilder() + pu.inflight.Wait() pu.pending = resetEntries(pu.pending) if pu.trie != nil { pu.trie.Reset() @@ -159,10 +130,8 @@ func (pu *parallelUpdate) Reset() { } func (pu *parallelUpdate) Close() { - pu.drainBuilder() - pu.freeCh = nil - clear(pu.pending) - pu.pending = nil + pu.inflight.Wait() + pu.pending, pu.spare = nil, nil pu.trie = nil pu.drainDeferred() pu.keyArena.reset() diff --git a/execution/commitment/prefix_trie_test.go b/execution/commitment/prefix_trie_test.go index 5fd2fb3b3cf..d62e3d5d012 100644 --- a/execution/commitment/prefix_trie_test.go +++ b/execution/commitment/prefix_trie_test.go @@ -31,6 +31,8 @@ type walkEntry struct { prefix []byte subtreeCount uint32 bitmap uint16 + plainKey []byte + update *Update } func collectWalk(t *prefixTrie) []walkEntry { @@ -43,7 +45,7 @@ func collectWalk(t *prefixTrie) []walkEntry { prefix := make([]byte, len(accPrefix)+len(n.ext)) copy(prefix, accPrefix) copy(prefix[len(accPrefix):], n.ext) - out = append(out, walkEntry{prefix: prefix, subtreeCount: n.subtreeCount, bitmap: n.bitmap}) + out = append(out, walkEntry{prefix: prefix, subtreeCount: n.subtreeCount, bitmap: n.bitmap, plainKey: n.plainKey, update: n.update}) childIdx := 0 for bm := n.bitmap; bm != 0; { nib := byte(bits.TrailingZeros16(bm)) From 481f11b6001f8d655a0cdfc7e28e62c721a8254c Mon Sep 17 00:00:00 2001 From: awskii Date: Mon, 12 Oct 2026 13:08:11 +0700 Subject: [PATCH 16/18] commitment: allocate the second touch chunk buffer at full size instead of growing it --- execution/commitment/parallel_update.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/execution/commitment/parallel_update.go b/execution/commitment/parallel_update.go index 5df23429560..6d0954c1a4d 100644 --- a/execution/commitment/parallel_update.go +++ b/execution/commitment/parallel_update.go @@ -17,6 +17,7 @@ package commitment import ( + "slices" "sync" ) @@ -87,7 +88,7 @@ func resetEntries(c []touchEntry) []touchEntry { func (pu *parallelUpdate) handOff() { pu.inflight.Wait() c := pu.pending - pu.pending, pu.spare = pu.spare[:0], c + pu.pending, pu.spare = slices.Grow(pu.spare[:0], touchChunkKeys), c pu.inflight.Go(func() { pu.insertChunk(c) clear(c) From 518b8b908c8d7361a045369a960bd56a1a711f18 Mon Sep 17 00:00:00 2001 From: awskii Date: Mon, 12 Oct 2026 13:15:54 +0700 Subject: [PATCH 17/18] commitment: compare chunk buffer addresses and pin arrival order for a key touched in several chunks --- execution/commitment/parallel_collect_test.go | 23 ++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/execution/commitment/parallel_collect_test.go b/execution/commitment/parallel_collect_test.go index 5775e3fd1ad..7e8f3528c8e 100644 --- a/execution/commitment/parallel_collect_test.go +++ b/execution/commitment/parallel_collect_test.go @@ -126,10 +126,10 @@ func TestChunkBuild_BuffersSurviveBatches(t *testing.T) { cases := randomTouchCases(777, touchChunkKeys+8, 0) pu := newParallelUpdate() - buffers := func() []*touchEntry { - return []*touchEntry{unsafe.SliceData(pu.pending), unsafe.SliceData(pu.spare)} + buffers := func() []uintptr { + return []uintptr{uintptr(unsafe.Pointer(unsafe.SliceData(pu.pending))), uintptr(unsafe.Pointer(unsafe.SliceData(pu.spare)))} } - var first []*touchEntry + var first []uintptr for batch := range 3 { pu.Reset() collectAll(pu, cases) @@ -144,6 +144,23 @@ func TestChunkBuild_BuffersSurviveBatches(t *testing.T) { require.ElementsMatch(t, first, buffers(), "later batches must reuse the same two buffers") } +func TestChunkBuild_KeyTouchedInSeveralChunksKeepsArrivalOrder(t *testing.T) { + t.Parallel() + + cases := randomTouchCases(41, 2*touchChunkKeys+5, 0) + hk := bytes.Repeat([]byte{0x0a}, 64) + for i, at := range []int{touchChunkKeys - 1, touchChunkKeys, 2*touchChunkKeys + 1} { + cases[at] = touchEntry{hashedKey: hk, plainKey: []byte("pk"), update: &Update{Flags: BalanceUpdate, Balance: *uint256.NewInt(uint64(i) + 1)}} + } + want := collectWalk(buildReferenceTrie(cases)) + reordered := slices.Concat(cases[touchChunkKeys:], cases[:touchChunkKeys]) + require.NotEqual(t, want, collectWalk(buildReferenceTrie(reordered)), "the fixture must be sensitive to chunk order") + + pu, backgrounded := buildChunked(cases) + require.True(t, backgrounded) + require.Equal(t, want, collectWalk(pu.trie), "a key touched in several chunks must keep its last update") +} + func TestChunkBuild_TouchHashedKeyCopiesCallerBuffer(t *testing.T) { t.Parallel() From e8190e045f84f492a781d05b3ac2ae0030e71cda Mon Sep 17 00:00:00 2001 From: awskii Date: Tue, 13 Oct 2026 15:27:30 +0700 Subject: [PATCH 18/18] commitment: build pending touch chunks before timing process-only benches; drop the redundant SetMode reset --- execution/commitment/commitment.go | 3 --- execution/commitment/parallel_streaming_bench_test.go | 4 ++++ 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/execution/commitment/commitment.go b/execution/commitment/commitment.go index b9220fa596f..f821a9a0a69 100644 --- a/execution/commitment/commitment.go +++ b/execution/commitment/commitment.go @@ -1380,9 +1380,6 @@ func NewUpdates(m Mode, tmpdir string, hasher keyHasher) *Updates { } func (t *Updates) SetMode(m Mode) { - if t.parallel != nil { - t.parallel.Reset() - } t.mode = m switch t.mode { case ModeDirect: diff --git a/execution/commitment/parallel_streaming_bench_test.go b/execution/commitment/parallel_streaming_bench_test.go index 5cd1bfc80d7..68976500547 100644 --- a/execution/commitment/parallel_streaming_bench_test.go +++ b/execution/commitment/parallel_streaming_bench_test.go @@ -98,6 +98,7 @@ func runParallelBenchWith(b *testing.B, pk [][]byte, updates []Update, workers i upds = NewUpdates(ModeParallel, "", KeyToHexNibbleHash) } else { upds = WrapKeyUpdates(b, ModeParallel, KeyToHexNibbleHash, pk, updates) + upds.parallel.Build() } b.StartTimer() @@ -418,6 +419,7 @@ func runParallelBenchGrain(b *testing.B, pk [][]byte, updates []Update, workers pph.SetForkGrain(grain) pph.RootTrie().Reset() upds := WrapKeyUpdates(b, ModeParallel, KeyToHexNibbleHash, pk, updates) + upds.parallel.Build() b.StartTimer() _, err := pph.Process(ctx, upds, "", nil, WarmupConfig{}) @@ -463,6 +465,7 @@ func runIncrementalParallelBenchGrain(b *testing.B, batch1, batch2 engineBatch, require.NoError(b, ms.applyPlainUpdates(batch2.keys, batch2.upds)) u2 := WrapKeyUpdates(b, ModeParallel, KeyToHexNibbleHash, batch2.keys, batch2.upds) + u2.parallel.Build() b.StartTimer() _, err = pph.Process(ctx, u2, "", nil, WarmupConfig{}) @@ -579,6 +582,7 @@ func runIncrementalParallelBench(b *testing.B, batch1, batch2 engineBatch, worke require.NoError(b, ms.applyPlainUpdates(batch2.keys, batch2.upds)) u2 := WrapKeyUpdates(b, ModeParallel, KeyToHexNibbleHash, batch2.keys, batch2.upds) + u2.parallel.Build() b.StartTimer() _, err = pph.Process(ctx, u2, "", nil, WarmupConfig{})