From 5da4fe545802f2e73f49b01b2f01ade4cd3235ac Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Wed, 29 Jul 2026 10:23:32 +0700 Subject: [PATCH 1/5] execution/commitment: reuse pooled trie workers across Process calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit resetPool() replaced the whole sync.Pool object on every Reset and Release, discarding every cached worker trie. On a mainnet node heap profile (alloc_space) the resulting refills cost 943MB — 27% of all sync.Pool refill bytes process-wide, ~9.8K refills at ~93KB each, one full NewHexPatriciaHashed per refill — and the drain was self-inflicted, not GC: Reset runs once per commitment round via SharedDomains.ClearRam. Workers are already scrubbed with resetForReuse() before every Put, so pool contents are safe to reuse. Replace the raw sync.Pool with a small trieWorkerPool that: - keeps cached workers across Reset (accountKeyLen/cfg never change after construction, so there is nothing to invalidate); - re-applies accountKeyLen + cfg on every pooled get, making a pooled hit identical to fresh construction (previously an intra-Process recycled worker kept zeroed cfg/memoizationOff while a fresh one got the configured values); - on Release drains cached workers back to the global hphPool, so a successor instance (fresh SharedDomains after a commit cycle) refills from cached objects instead of allocating. put() centralizes the resetForReuse-before-Put scrub that every call site previously repeated by hand. Claude-Session: https://claude.ai/code/session_01UYCsHj9HYTJnqUCy8a7W91 --- execution/commitment/parallel_mount.go | 11 ++-- .../commitment/parallel_patricia_hashed.go | 60 +++++++++++++----- .../parallel_patricia_hashed_test.go | 63 +++++++++++++++++++ execution/commitment/streaming_commitment.go | 31 +++------ .../commitment/streaming_commitment_test.go | 16 +++++ execution/commitment/streaming_deep_fold.go | 8 +-- 6 files changed, 138 insertions(+), 51 deletions(-) diff --git a/execution/commitment/parallel_mount.go b/execution/commitment/parallel_mount.go index 53544bf38e6..997ac1c5915 100644 --- a/execution/commitment/parallel_mount.go +++ b/execution/commitment/parallel_mount.go @@ -127,7 +127,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up child := root.children[childIdx] ni, ch := nib, child g.Go(func() error { - w := p.workerPool.Get().(*HexPatriciaHashed) + w := p.workerPool.get() w.mountTo(base, ni) if p.template != nil && p.template.traceW != nil { w.traceW = tracePrefix(p.template.traceW, fmt.Sprintf("[mnt %x] ", ni)) @@ -154,8 +154,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up return foldStorageRoot(gctx, foldSem, p.newStorageWorker, pu, n, pth, accountFresh) }) if buildErr != nil { - w.resetForReuse() - p.workerPool.Put(w) + p.workerPool.put(w) return fmt.Errorf("mount[%x] build: %w", ni, buildErr) } var tf time.Time @@ -168,8 +167,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up foldDur[ni] = time.Since(tf) } if err != nil { - w.resetForReuse() - p.workerPool.Put(w) + p.workerPool.put(w) return fmt.Errorf("mount[%x] fold: %w", ni, err) } cells[ni] = c @@ -177,8 +175,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up if deferred := w.TakeDeferredUpdates(); len(deferred) > 0 { pu.appendDeferred(deferred) } - w.resetForReuse() - p.workerPool.Put(w) + p.workerPool.put(w) return nil }) childIdx++ diff --git a/execution/commitment/parallel_patricia_hashed.go b/execution/commitment/parallel_patricia_hashed.go index 94956c82326..497529b42dc 100644 --- a/execution/commitment/parallel_patricia_hashed.go +++ b/execution/commitment/parallel_patricia_hashed.go @@ -27,12 +27,49 @@ import ( "sync/atomic" ) +// trieWorkerPool caches scrubbed HexPatriciaHashed workers across Process calls; +// only put may insert, so every pooled worker is already reset for reuse. +type trieWorkerPool struct { + pool sync.Pool + accountKeyLen int16 + cfg TrieConfig +} + +func (wp *trieWorkerPool) init(accountKeyLen int16, cfg TrieConfig) { + wp.accountKeyLen = accountKeyLen + wp.cfg = cfg +} + +func (wp *trieWorkerPool) get() *HexPatriciaHashed { + if w, ok := wp.pool.Get().(*HexPatriciaHashed); ok { + w.accountKeyLen = wp.accountKeyLen + w.applyConfig(wp.cfg) + return w + } + return NewHexPatriciaHashed(wp.accountKeyLen, nil, wp.cfg) +} + +func (wp *trieWorkerPool) put(w *HexPatriciaHashed) { + w.resetForReuse() + wp.pool.Put(w) +} + +// drain hands cached workers back to the shared constructor pool instead of the GC. +func (wp *trieWorkerPool) drain() { + for { + w, ok := wp.pool.Get().(*HexPatriciaHashed) + if !ok { + return + } + w.Release() + } +} + // ParallelPatriciaHashed is the trie-side of the parallel commitment pipeline. type ParallelPatriciaHashed struct { template *HexPatriciaHashed trieCtxFactory TrieContextFactory - workerPool sync.Pool - cfg TrieConfig + workerPool trieWorkerPool accountKeyLen int16 numWorkers int @@ -51,23 +88,12 @@ func NewParallelPatriciaHashed(ctxFactory TrieContextFactory, accountKeyLen int1 template: NewHexPatriciaHashed(accountKeyLen, nil, cfg), trieCtxFactory: ctxFactory, accountKeyLen: accountKeyLen, - cfg: cfg, numWorkers: runtime.NumCPU(), } - p.resetPool() + p.workerPool.init(accountKeyLen, cfg) return p } -func (p *ParallelPatriciaHashed) resetPool() { - akl := p.accountKeyLen - cfg := p.cfg - p.workerPool = sync.Pool{ - New: func() any { - return NewHexPatriciaHashed(akl, nil, cfg) - }, - } -} - // SetNumWorkers overrides the worker count for the next Process call; n <= 0 falls back to runtime.NumCPU. func (p *ParallelPatriciaHashed) SetNumWorkers(n int) { if n <= 0 { @@ -110,13 +136,13 @@ func (p *ParallelPatriciaHashed) RootTrie() *HexPatriciaHashed { return p.template } -// Reset clears the published root hash, drops pooled workers, and resets the template so the instance can be reused. +// Reset clears the published root hash and resets the template so the instance +// can be reused; pooled workers stay cached for the next Process call. func (p *ParallelPatriciaHashed) Reset() { if p.template != nil { p.template.Reset() } p.rootHash.Store(nil) - p.resetPool() if p.streaming != nil { p.streaming.Reset() } @@ -129,7 +155,7 @@ func (p *ParallelPatriciaHashed) Release() { p.template = nil } p.rootHash.Store(nil) - p.resetPool() + p.workerPool.drain() if p.streaming != nil { p.streaming.Release() p.streaming = nil diff --git a/execution/commitment/parallel_patricia_hashed_test.go b/execution/commitment/parallel_patricia_hashed_test.go index 93e1f677456..07f23ee06dc 100644 --- a/execution/commitment/parallel_patricia_hashed_test.go +++ b/execution/commitment/parallel_patricia_hashed_test.go @@ -126,6 +126,22 @@ func TestParallelPatriciaHashedSkeletonReset(t *testing.T) { require.NotNil(t, p.template, "Reset preserves the template") } +func TestParallelWorkerPoolSurvivesReset(t *testing.T) { + p := NewParallelPatriciaHashed(nil, length.Addr, DefaultTrieConfig()) + defer p.Release() + + const tag = 7 + for range 4 { + w := NewHexPatriciaHashed(length.Addr, nil, DefaultTrieConfig()) + w.mountedNib = tag + p.workerPool.pool.Put(w) + } + p.Reset() + + got := p.workerPool.get() + assert.Equal(t, tag, got.mountedNib, "Reset must keep cached workers instead of discarding the pool") +} + func TestParallelPatriciaHashedSkeletonRelease(t *testing.T) { p := NewParallelPatriciaHashed(nil, length.Addr, DefaultTrieConfig()) require.NotNil(t, p.template) @@ -872,6 +888,53 @@ func requireIncrementalEquiv(t *testing.T, k1 [][]byte, u1 []Update, k2 [][]byte requireAllEnginesParity(t, k1, u1, k2, u2, workers) } +// Folds two batches on ONE trie instance with the per-block Reset + SetState +// lifecycle in between, so batch-2 runs on workers cached by batch-1. +func reusedInstanceIncrementalRoot(t *testing.T, variant TrieVariant, workers int, k1 [][]byte, u1 []Update, k2 [][]byte, u2 []Update) []byte { + t.Helper() + ctx := context.Background() + ms := NewMockState(t) + ms.SetConcurrentCommitment(true) + + cfg := DefaultTrieConfig() + cfg.Variant = variant + trie, ut := InitializeTrieAndUpdates(ModeDirect, t.TempDir(), cfg) + defer ut.Close() + defer trie.Release() + pt := trie.(*ParallelPatriciaHashed) + pt.SetNumWorkers(workers) + pt.SetTrieContextFactory(mockTrieCtxFactory(ms)) + pt.ResetContext(ms) + + process := func(keys [][]byte, upds []Update) ([]byte, []byte) { + require.NoError(t, ms.applyPlainUpdates(keys, upds)) + WrapKeyUpdatesInto(t, ut, keys, upds) + root, err := pt.Process(ctx, ut, "", nil, WarmupConfig{}) + require.NoError(t, err) + blob, err := pt.RootTrie().EncodeCurrentState(nil) + require.NoError(t, err) + return bytes.Clone(root), blob + } + + _, blob := process(k1, u1) + pt.Reset() + require.NoError(t, pt.RootTrie().SetState(blob)) + root2, _ := process(k2, u2) + return root2 +} + +func TestParallelReuseAcrossResetParity(t *testing.T) { + t.Parallel() + k1, u1 := genRandomAccountsStorage(256) + k2, u2 := sparseBatch2(k1, 3, false) + seqRoot, _ := incrementalRoot(t, modeSeq, 0, k1, u1, k2, u2) + + for _, variant := range []TrieVariant{VariantParallelHexPatricia, VariantStreamingHexPatricia} { + root := reusedInstanceIncrementalRoot(t, variant, 8, k1, u1, k2, u2) + require.Equalf(t, seqRoot, root, "reused-instance %s incremental root vs sequential", variant) + } +} + func TestVerifyParallel_WideNested(t *testing.T) { t.Parallel() keys, upds := genWideNested(t) diff --git a/execution/commitment/streaming_commitment.go b/execution/commitment/streaming_commitment.go index 7a45a9ac0da..2bc7f75faab 100644 --- a/execution/commitment/streaming_commitment.go +++ b/execution/commitment/streaming_commitment.go @@ -69,7 +69,7 @@ type StreamingCommitter struct { accountKeyLen int16 numWorkers int - workerPool sync.Pool + workerPool trieWorkerPool trie *prefixTrie splits map[byte]*splitState eagerFloor uint64 @@ -123,18 +123,10 @@ func NewStreamingCommitter(ctxFactory TrieContextFactory, accountKeyLen int16, c splits: make(map[byte]*splitState), eagerFloor: defaultEagerFold, } - sc.resetPool() + sc.workerPool.init(accountKeyLen, cfg) return sc } -func (sc *StreamingCommitter) resetPool() { - akl := sc.accountKeyLen - cfg := sc.cfg - sc.workerPool = sync.Pool{ - New: func() any { return NewHexPatriciaHashed(akl, nil, cfg) }, - } -} - // SetNumWorkers overrides the worker count for the next Process call. Values // <= 0 fall back to runtime.NumCPU. func (sc *StreamingCommitter) SetNumWorkers(n int) { @@ -566,7 +558,7 @@ func (sc *StreamingCommitter) markQueued(s *splitState, nib byte) { // foldKeys folds a snapshotted split's keys on a pooled worker whose overlay ctx // discards branch writes; the returned flushed flag reports a mid-fold self-flush. func (sc *StreamingCommitter) foldKeys(nib byte, keys []touchedKey) (cell, []*DeferredBranchUpdate, bool, error) { - w := sc.workerPool.Get().(*HexPatriciaHashed) + w := sc.workerPool.get() w.mountTo(sc.base, int(nib)) if sc.traceW != nil { w.SetTraceWriter(tracePrefix(sc.traceW, fmt.Sprintf("[fold %x] ", nib))) @@ -593,8 +585,7 @@ func (sc *StreamingCommitter) foldKeys(nib byte, keys []touchedKey) (cell, []*De c, err = w.foldMounted(sc.bgCtx, int(nib)) } deferred := w.TakeDeferredUpdates() - w.resetForReuse() - sc.workerPool.Put(w) + sc.workerPool.put(w) return c, deferred, ov.flushed, err } @@ -746,7 +737,7 @@ func stitchSplitCells(base *HexPatriciaHashed, cells *[16]cell, present *[16]boo // deferred set. func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore.Weighted, base *HexPatriciaHashed, s *splitState, child *prefixNode) error { ni := s.prefix[0] - w := sc.workerPool.Get().(*HexPatriciaHashed) + w := sc.workerPool.get() w.mountTo(base, int(ni)) if sc.traceW != nil { w.SetTraceWriter(tracePrefix(sc.traceW, fmt.Sprintf("[split %x] ", ni))) @@ -773,8 +764,7 @@ func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore. return sr, err } if err := dfsSubtreeDeep(w, child, path, deepStorageRoot); err != nil { - w.resetForReuse() - sc.workerPool.Put(w) + sc.workerPool.put(w) for _, upd := range pu.deferredCombined { putDeferredUpdate(upd) } @@ -782,8 +772,7 @@ func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore. } c, err := w.foldMounted(ctx, int(ni)) if err != nil { - w.resetForReuse() - sc.workerPool.Put(w) + sc.workerPool.put(w) for _, upd := range pu.deferredCombined { putDeferredUpdate(upd) } @@ -794,8 +783,7 @@ func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore. if d := w.TakeDeferredUpdates(); len(d) > 0 { newDeferred = append(newDeferred, d...) } - w.resetForReuse() - sc.workerPool.Put(w) + sc.workerPool.put(w) s.mu.Lock() for _, upd := range s.deferred { @@ -942,7 +930,6 @@ func (sc *StreamingCommitter) Reset() { } sc.deferredForCaller = nil sc.rootValid, sc.rootSeeded = false, false - sc.resetPool() } // releaseBase drops the scheduler's persistent base and its context. @@ -967,5 +954,5 @@ func (sc *StreamingCommitter) Release() { } sc.deferredForCaller = nil sc.rootValid, sc.rootSeeded = false, false - sc.resetPool() + sc.workerPool.drain() } diff --git a/execution/commitment/streaming_commitment_test.go b/execution/commitment/streaming_commitment_test.go index c8b32a91786..59adbedb792 100644 --- a/execution/commitment/streaming_commitment_test.go +++ b/execution/commitment/streaming_commitment_test.go @@ -51,6 +51,22 @@ func TestUpdates_NewEmpty_PreservesStreaming(t *testing.T) { require.NotZero(t, sink.touches, "rotated buffer did not forward touch to the streamer") } +func TestStreamingWorkerPoolSurvivesReset(t *testing.T) { + sc := NewStreamingCommitter(nil, length.Addr, DefaultTrieConfig()) + defer sc.Release() + + const tag = 7 + for range 4 { + w := NewHexPatriciaHashed(length.Addr, nil, DefaultTrieConfig()) + w.mountedNib = tag + sc.workerPool.pool.Put(w) + } + sc.Reset() + + got := sc.workerPool.get() + require.Equal(t, tag, got.mountedNib, "Reset must keep cached workers instead of discarding the pool") +} + func streamingRoot(t *testing.T, workers int, keys [][]byte, upds []Update, idxOrder []int) ([]byte, *MockState) { t.Helper() sc, ms := newStreamingFixture(t, keys, upds, workers) diff --git a/execution/commitment/streaming_deep_fold.go b/execution/commitment/streaming_deep_fold.go index 5a8aec0aa84..b2efe48b4aa 100644 --- a/execution/commitment/streaming_deep_fold.go +++ b/execution/commitment/streaming_deep_fold.go @@ -23,7 +23,6 @@ import ( "io" "math/bits" "runtime" - "sync" "golang.org/x/sync/errgroup" "golang.org/x/sync/semaphore" @@ -289,16 +288,15 @@ func storageRootFromSingleChild(base *HexPatriciaHashed) (cell, error) { // newDeferredStorageWorker yields a pooled trie worker for a deferring storage fold // and a release that returns it to the pool and frees its context. -func newDeferredStorageWorker(ctx context.Context, pool *sync.Pool, factory TrieContextFactory, traceW io.Writer) (*HexPatriciaHashed, func()) { - w := pool.Get().(*HexPatriciaHashed) +func newDeferredStorageWorker(ctx context.Context, pool *trieWorkerPool, factory TrieContextFactory, traceW io.Writer) (*HexPatriciaHashed, func()) { + w := pool.get() wctx, cleanup := factory(ctx) w.ResetContext(wctx) w.SetTraceWriter(traceW) w.branchEncoder.setDeferUpdates(true) w.SetLeaveDeferredForCaller(true) return w, func() { - w.resetForReuse() - pool.Put(w) + pool.put(w) if cleanup != nil { cleanup() } From 096bbdb9d02451e486a90a4f13f2880d4bc1dbd1 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Wed, 29 Jul 2026 10:29:17 +0700 Subject: [PATCH 2/5] execution/commitment: return checked-out worker to pool in survival tests Claude-Session: https://claude.ai/code/session_01UYCsHj9HYTJnqUCy8a7W91 --- execution/commitment/parallel_patricia_hashed_test.go | 1 + execution/commitment/streaming_commitment_test.go | 1 + 2 files changed, 2 insertions(+) diff --git a/execution/commitment/parallel_patricia_hashed_test.go b/execution/commitment/parallel_patricia_hashed_test.go index 07f23ee06dc..8378a29eabe 100644 --- a/execution/commitment/parallel_patricia_hashed_test.go +++ b/execution/commitment/parallel_patricia_hashed_test.go @@ -140,6 +140,7 @@ func TestParallelWorkerPoolSurvivesReset(t *testing.T) { got := p.workerPool.get() assert.Equal(t, tag, got.mountedNib, "Reset must keep cached workers instead of discarding the pool") + p.workerPool.put(got) } func TestParallelPatriciaHashedSkeletonRelease(t *testing.T) { diff --git a/execution/commitment/streaming_commitment_test.go b/execution/commitment/streaming_commitment_test.go index 59adbedb792..9f432d183d2 100644 --- a/execution/commitment/streaming_commitment_test.go +++ b/execution/commitment/streaming_commitment_test.go @@ -65,6 +65,7 @@ func TestStreamingWorkerPoolSurvivesReset(t *testing.T) { got := sc.workerPool.get() require.Equal(t, tag, got.mountedNib, "Reset must keep cached workers instead of discarding the pool") + sc.workerPool.put(got) } func streamingRoot(t *testing.T, workers int, keys [][]byte, upds []Update, idxOrder []int) ([]byte, *MockState) { From 765de5d06109c8c89b4edde3ee5eb13491afa211 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Wed, 29 Jul 2026 11:16:12 +0700 Subject: [PATCH 3/5] execution/commitment: replace slot-dependent pool tests with config invariant The tagged-worker identity assertion depends on sync.Pool per-P slot placement; goroutine migration (frequent under -race) makes it flaky. The property it guarded is structural now, so pin the checkout invariant instead: every get() is config-correct regardless of pool contents. Claude-Session: https://claude.ai/code/session_01UYCsHj9HYTJnqUCy8a7W91 --- .../parallel_patricia_hashed_test.go | 30 +++++++++++-------- .../commitment/streaming_commitment_test.go | 17 ----------- 2 files changed, 17 insertions(+), 30 deletions(-) diff --git a/execution/commitment/parallel_patricia_hashed_test.go b/execution/commitment/parallel_patricia_hashed_test.go index 8378a29eabe..ac432823940 100644 --- a/execution/commitment/parallel_patricia_hashed_test.go +++ b/execution/commitment/parallel_patricia_hashed_test.go @@ -126,21 +126,25 @@ func TestParallelPatriciaHashedSkeletonReset(t *testing.T) { require.NotNil(t, p.template, "Reset preserves the template") } -func TestParallelWorkerPoolSurvivesReset(t *testing.T) { - p := NewParallelPatriciaHashed(nil, length.Addr, DefaultTrieConfig()) - defer p.Release() +// Every checkout must be config-correct whether it hit the shared pool or +// constructed fresh — that fungibility is what lets workers cross instances. +func TestWorkerPoolCheckoutAppliesConfig(t *testing.T) { + cfg := DefaultTrieConfig() + cfg.MemoizationOff = true - const tag = 7 - for range 4 { - w := NewHexPatriciaHashed(length.Addr, nil, DefaultTrieConfig()) - w.mountedNib = tag - p.workerPool.pool.Put(w) - } - p.Reset() + var wp trieWorkerPool + wp.init(length.Addr, cfg) + + stale := NewHexPatriciaHashed(2*length.Addr, nil, DefaultTrieConfig()) + stale.Release() - got := p.workerPool.get() - assert.Equal(t, tag, got.mountedNib, "Reset must keep cached workers instead of discarding the pool") - p.workerPool.put(got) + for range 2 { + w := wp.get() + assert.Equal(t, int16(length.Addr), w.accountKeyLen) + assert.True(t, w.memoizationOff) + assert.Equal(t, cfg, w.cfg) + wp.put(w) + } } func TestParallelPatriciaHashedSkeletonRelease(t *testing.T) { diff --git a/execution/commitment/streaming_commitment_test.go b/execution/commitment/streaming_commitment_test.go index 9f432d183d2..c8b32a91786 100644 --- a/execution/commitment/streaming_commitment_test.go +++ b/execution/commitment/streaming_commitment_test.go @@ -51,23 +51,6 @@ func TestUpdates_NewEmpty_PreservesStreaming(t *testing.T) { require.NotZero(t, sink.touches, "rotated buffer did not forward touch to the streamer") } -func TestStreamingWorkerPoolSurvivesReset(t *testing.T) { - sc := NewStreamingCommitter(nil, length.Addr, DefaultTrieConfig()) - defer sc.Release() - - const tag = 7 - for range 4 { - w := NewHexPatriciaHashed(length.Addr, nil, DefaultTrieConfig()) - w.mountedNib = tag - sc.workerPool.pool.Put(w) - } - sc.Reset() - - got := sc.workerPool.get() - require.Equal(t, tag, got.mountedNib, "Reset must keep cached workers instead of discarding the pool") - sc.workerPool.put(got) -} - func streamingRoot(t *testing.T, workers int, keys [][]byte, upds []Update, idxOrder []int) ([]byte, *MockState) { t.Helper() sc, ms := newStreamingFixture(t, keys, upds, workers) From 646389025e8d6012466d8902d7c5038de59b40e3 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Wed, 29 Jul 2026 12:56:45 +0700 Subject: [PATCH 4/5] execution/commitment: back trieWorkerPool by the package-level hphPool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get() re-applies config on every checkout, so pooled workers are fungible across instances — the instance-local sync.Pool tier and its drain() duplicated what NewHexPatriciaHashed/Release already do against hphPool. Workers now become globally shareable at put() time instead of waiting for instance teardown. Claude-Session: https://claude.ai/code/session_01UYCsHj9HYTJnqUCy8a7W91 --- .../commitment/parallel_patricia_hashed.go | 26 +++---------------- execution/commitment/streaming_commitment.go | 1 - 2 files changed, 4 insertions(+), 23 deletions(-) diff --git a/execution/commitment/parallel_patricia_hashed.go b/execution/commitment/parallel_patricia_hashed.go index 497529b42dc..2b0bef79185 100644 --- a/execution/commitment/parallel_patricia_hashed.go +++ b/execution/commitment/parallel_patricia_hashed.go @@ -27,10 +27,10 @@ import ( "sync/atomic" ) -// trieWorkerPool caches scrubbed HexPatriciaHashed workers across Process calls; -// only put may insert, so every pooled worker is already reset for reuse. +// trieWorkerPool hands out HexPatriciaHashed workers backed by the package-level +// hphPool. Every checkout re-applies this instance's config, so workers are +// fungible across instances and survive instance teardown in the shared pool. type trieWorkerPool struct { - pool sync.Pool accountKeyLen int16 cfg TrieConfig } @@ -41,28 +41,11 @@ func (wp *trieWorkerPool) init(accountKeyLen int16, cfg TrieConfig) { } func (wp *trieWorkerPool) get() *HexPatriciaHashed { - if w, ok := wp.pool.Get().(*HexPatriciaHashed); ok { - w.accountKeyLen = wp.accountKeyLen - w.applyConfig(wp.cfg) - return w - } return NewHexPatriciaHashed(wp.accountKeyLen, nil, wp.cfg) } func (wp *trieWorkerPool) put(w *HexPatriciaHashed) { - w.resetForReuse() - wp.pool.Put(w) -} - -// drain hands cached workers back to the shared constructor pool instead of the GC. -func (wp *trieWorkerPool) drain() { - for { - w, ok := wp.pool.Get().(*HexPatriciaHashed) - if !ok { - return - } - w.Release() - } + w.Release() } // ParallelPatriciaHashed is the trie-side of the parallel commitment pipeline. @@ -155,7 +138,6 @@ func (p *ParallelPatriciaHashed) Release() { p.template = nil } p.rootHash.Store(nil) - p.workerPool.drain() if p.streaming != nil { p.streaming.Release() p.streaming = nil diff --git a/execution/commitment/streaming_commitment.go b/execution/commitment/streaming_commitment.go index 2bc7f75faab..b609d3e6869 100644 --- a/execution/commitment/streaming_commitment.go +++ b/execution/commitment/streaming_commitment.go @@ -954,5 +954,4 @@ func (sc *StreamingCommitter) Release() { } sc.deferredForCaller = nil sc.rootValid, sc.rootSeeded = false, false - sc.workerPool.drain() } From 99caa1fd57f8601819e24358970b81dba2de9bf0 Mon Sep 17 00:00:00 2001 From: Alexey Sharov Date: Wed, 29 Jul 2026 13:08:18 +0700 Subject: [PATCH 5/5] execution/commitment: dissolve trieWorkerPool into direct constructor calls The handle only carried accountKeyLen+cfg; call NewHexPatriciaHashed and Release directly, with cfg stored on the owning structs. Claude-Session: https://claude.ai/code/session_01UYCsHj9HYTJnqUCy8a7W91 --- execution/commitment/parallel_mount.go | 10 ++++---- .../commitment/parallel_patricia_hashed.go | 25 ++----------------- .../parallel_patricia_hashed_test.go | 9 +++---- execution/commitment/streaming_commitment.go | 16 ++++++------ execution/commitment/streaming_deep_fold.go | 6 ++--- 5 files changed, 20 insertions(+), 46 deletions(-) diff --git a/execution/commitment/parallel_mount.go b/execution/commitment/parallel_mount.go index 997ac1c5915..51edf0e0547 100644 --- a/execution/commitment/parallel_mount.go +++ b/execution/commitment/parallel_mount.go @@ -127,7 +127,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up child := root.children[childIdx] ni, ch := nib, child g.Go(func() error { - w := p.workerPool.get() + w := NewHexPatriciaHashed(p.accountKeyLen, nil, p.cfg) w.mountTo(base, ni) if p.template != nil && p.template.traceW != nil { w.traceW = tracePrefix(p.template.traceW, fmt.Sprintf("[mnt %x] ", ni)) @@ -154,7 +154,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up return foldStorageRoot(gctx, foldSem, p.newStorageWorker, pu, n, pth, accountFresh) }) if buildErr != nil { - p.workerPool.put(w) + w.Release() return fmt.Errorf("mount[%x] build: %w", ni, buildErr) } var tf time.Time @@ -167,7 +167,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up foldDur[ni] = time.Since(tf) } if err != nil { - p.workerPool.put(w) + w.Release() return fmt.Errorf("mount[%x] fold: %w", ni, err) } cells[ni] = c @@ -175,7 +175,7 @@ func (p *ParallelPatriciaHashed) processMounted(ctx context.Context, updates *Up if deferred := w.TakeDeferredUpdates(); len(deferred) > 0 { pu.appendDeferred(deferred) } - p.workerPool.put(w) + w.Release() return nil }) childIdx++ @@ -249,7 +249,7 @@ func (p *ParallelPatriciaHashed) newStorageWorker(ctx context.Context) (*HexPatr if p.template != nil { traceW = p.template.traceW } - return newDeferredStorageWorker(ctx, &p.workerPool, p.trieCtxFactory, traceW) + return newDeferredStorageWorker(ctx, p.accountKeyLen, p.cfg, p.trieCtxFactory, traceW) } // setAccountStorageRoot writes the folded storage-root cell sr onto the account leaf. diff --git a/execution/commitment/parallel_patricia_hashed.go b/execution/commitment/parallel_patricia_hashed.go index 2b0bef79185..58fb0006ff5 100644 --- a/execution/commitment/parallel_patricia_hashed.go +++ b/execution/commitment/parallel_patricia_hashed.go @@ -27,34 +27,13 @@ import ( "sync/atomic" ) -// trieWorkerPool hands out HexPatriciaHashed workers backed by the package-level -// hphPool. Every checkout re-applies this instance's config, so workers are -// fungible across instances and survive instance teardown in the shared pool. -type trieWorkerPool struct { - accountKeyLen int16 - cfg TrieConfig -} - -func (wp *trieWorkerPool) init(accountKeyLen int16, cfg TrieConfig) { - wp.accountKeyLen = accountKeyLen - wp.cfg = cfg -} - -func (wp *trieWorkerPool) get() *HexPatriciaHashed { - return NewHexPatriciaHashed(wp.accountKeyLen, nil, wp.cfg) -} - -func (wp *trieWorkerPool) put(w *HexPatriciaHashed) { - w.Release() -} - // ParallelPatriciaHashed is the trie-side of the parallel commitment pipeline. type ParallelPatriciaHashed struct { template *HexPatriciaHashed trieCtxFactory TrieContextFactory - workerPool trieWorkerPool accountKeyLen int16 + cfg TrieConfig numWorkers int rootHash atomic.Pointer[[]byte] @@ -72,8 +51,8 @@ func NewParallelPatriciaHashed(ctxFactory TrieContextFactory, accountKeyLen int1 trieCtxFactory: ctxFactory, accountKeyLen: accountKeyLen, numWorkers: runtime.NumCPU(), + cfg: cfg, } - p.workerPool.init(accountKeyLen, cfg) return p } diff --git a/execution/commitment/parallel_patricia_hashed_test.go b/execution/commitment/parallel_patricia_hashed_test.go index ac432823940..a260cada6b3 100644 --- a/execution/commitment/parallel_patricia_hashed_test.go +++ b/execution/commitment/parallel_patricia_hashed_test.go @@ -128,22 +128,19 @@ func TestParallelPatriciaHashedSkeletonReset(t *testing.T) { // Every checkout must be config-correct whether it hit the shared pool or // constructed fresh — that fungibility is what lets workers cross instances. -func TestWorkerPoolCheckoutAppliesConfig(t *testing.T) { +func TestWorkerCheckoutAppliesConfig(t *testing.T) { cfg := DefaultTrieConfig() cfg.MemoizationOff = true - var wp trieWorkerPool - wp.init(length.Addr, cfg) - stale := NewHexPatriciaHashed(2*length.Addr, nil, DefaultTrieConfig()) stale.Release() for range 2 { - w := wp.get() + w := NewHexPatriciaHashed(length.Addr, nil, cfg) assert.Equal(t, int16(length.Addr), w.accountKeyLen) assert.True(t, w.memoizationOff) assert.Equal(t, cfg, w.cfg) - wp.put(w) + w.Release() } } diff --git a/execution/commitment/streaming_commitment.go b/execution/commitment/streaming_commitment.go index b609d3e6869..44af52de98d 100644 --- a/execution/commitment/streaming_commitment.go +++ b/execution/commitment/streaming_commitment.go @@ -69,7 +69,6 @@ type StreamingCommitter struct { accountKeyLen int16 numWorkers int - workerPool trieWorkerPool trie *prefixTrie splits map[byte]*splitState eagerFloor uint64 @@ -123,7 +122,6 @@ func NewStreamingCommitter(ctxFactory TrieContextFactory, accountKeyLen int16, c splits: make(map[byte]*splitState), eagerFloor: defaultEagerFold, } - sc.workerPool.init(accountKeyLen, cfg) return sc } @@ -558,7 +556,7 @@ func (sc *StreamingCommitter) markQueued(s *splitState, nib byte) { // foldKeys folds a snapshotted split's keys on a pooled worker whose overlay ctx // discards branch writes; the returned flushed flag reports a mid-fold self-flush. func (sc *StreamingCommitter) foldKeys(nib byte, keys []touchedKey) (cell, []*DeferredBranchUpdate, bool, error) { - w := sc.workerPool.get() + w := NewHexPatriciaHashed(sc.accountKeyLen, nil, sc.cfg) w.mountTo(sc.base, int(nib)) if sc.traceW != nil { w.SetTraceWriter(tracePrefix(sc.traceW, fmt.Sprintf("[fold %x] ", nib))) @@ -585,7 +583,7 @@ func (sc *StreamingCommitter) foldKeys(nib byte, keys []touchedKey) (cell, []*De c, err = w.foldMounted(sc.bgCtx, int(nib)) } deferred := w.TakeDeferredUpdates() - sc.workerPool.put(w) + w.Release() return c, deferred, ov.flushed, err } @@ -737,7 +735,7 @@ func stitchSplitCells(base *HexPatriciaHashed, cells *[16]cell, present *[16]boo // deferred set. func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore.Weighted, base *HexPatriciaHashed, s *splitState, child *prefixNode) error { ni := s.prefix[0] - w := sc.workerPool.get() + w := NewHexPatriciaHashed(sc.accountKeyLen, nil, sc.cfg) w.mountTo(base, int(ni)) if sc.traceW != nil { w.SetTraceWriter(tracePrefix(sc.traceW, fmt.Sprintf("[split %x] ", ni))) @@ -764,7 +762,7 @@ func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore. return sr, err } if err := dfsSubtreeDeep(w, child, path, deepStorageRoot); err != nil { - sc.workerPool.put(w) + w.Release() for _, upd := range pu.deferredCombined { putDeferredUpdate(upd) } @@ -772,7 +770,7 @@ func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore. } c, err := w.foldMounted(ctx, int(ni)) if err != nil { - sc.workerPool.put(w) + w.Release() for _, upd := range pu.deferredCombined { putDeferredUpdate(upd) } @@ -783,7 +781,7 @@ func (sc *StreamingCommitter) foldSplit(ctx context.Context, foldSem *semaphore. if d := w.TakeDeferredUpdates(); len(d) > 0 { newDeferred = append(newDeferred, d...) } - sc.workerPool.put(w) + w.Release() s.mu.Lock() for _, upd := range s.deferred { @@ -802,7 +800,7 @@ func (sc *StreamingCommitter) DeepLocalFolds() uint64 { return sc.deepLocalFolds // newStorageWorker sources a concurrent-storage-fold worker; disjoint subtree // prefixes keep a mid-fold self-flush from racing another fold's writes. func (sc *StreamingCommitter) newStorageWorker(ctx context.Context) (*HexPatriciaHashed, func()) { - return newDeferredStorageWorker(ctx, &sc.workerPool, sc.trieCtxFactory, sc.traceW) + return newDeferredStorageWorker(ctx, sc.accountKeyLen, sc.cfg, sc.trieCtxFactory, sc.traceW) } // dropSplitDeferred returns every split's staged deferred branch updates to the pool. diff --git a/execution/commitment/streaming_deep_fold.go b/execution/commitment/streaming_deep_fold.go index b2efe48b4aa..de1cc4ed435 100644 --- a/execution/commitment/streaming_deep_fold.go +++ b/execution/commitment/streaming_deep_fold.go @@ -288,15 +288,15 @@ func storageRootFromSingleChild(base *HexPatriciaHashed) (cell, error) { // newDeferredStorageWorker yields a pooled trie worker for a deferring storage fold // and a release that returns it to the pool and frees its context. -func newDeferredStorageWorker(ctx context.Context, pool *trieWorkerPool, factory TrieContextFactory, traceW io.Writer) (*HexPatriciaHashed, func()) { - w := pool.get() +func newDeferredStorageWorker(ctx context.Context, accountKeyLen int16, cfg TrieConfig, factory TrieContextFactory, traceW io.Writer) (*HexPatriciaHashed, func()) { + w := NewHexPatriciaHashed(accountKeyLen, nil, cfg) wctx, cleanup := factory(ctx) w.ResetContext(wctx) w.SetTraceWriter(traceW) w.branchEncoder.setDeferUpdates(true) w.SetLeaveDeferredForCaller(true) return w, func() { - pool.put(w) + w.Release() if cleanup != nil { cleanup() }