diff --git a/execution/commitment/parallel_mount.go b/execution/commitment/parallel_mount.go index 53544bf38e6..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().(*HexPatriciaHashed) + 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,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) + w.Release() 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) + w.Release() 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) + w.Release() return nil }) childIdx++ @@ -252,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 94956c82326..58fb0006ff5 100644 --- a/execution/commitment/parallel_patricia_hashed.go +++ b/execution/commitment/parallel_patricia_hashed.go @@ -31,10 +31,9 @@ import ( type ParallelPatriciaHashed struct { template *HexPatriciaHashed trieCtxFactory TrieContextFactory - workerPool sync.Pool - cfg TrieConfig accountKeyLen int16 + cfg TrieConfig numWorkers int rootHash atomic.Pointer[[]byte] @@ -51,23 +50,12 @@ func NewParallelPatriciaHashed(ctxFactory TrieContextFactory, accountKeyLen int1 template: NewHexPatriciaHashed(accountKeyLen, nil, cfg), trieCtxFactory: ctxFactory, accountKeyLen: accountKeyLen, - cfg: cfg, numWorkers: runtime.NumCPU(), + cfg: cfg, } - p.resetPool() 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 +98,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 +117,6 @@ func (p *ParallelPatriciaHashed) Release() { p.template = nil } p.rootHash.Store(nil) - p.resetPool() 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..a260cada6b3 100644 --- a/execution/commitment/parallel_patricia_hashed_test.go +++ b/execution/commitment/parallel_patricia_hashed_test.go @@ -126,6 +126,24 @@ func TestParallelPatriciaHashedSkeletonReset(t *testing.T) { require.NotNil(t, p.template, "Reset preserves the template") } +// Every checkout must be config-correct whether it hit the shared pool or +// constructed fresh — that fungibility is what lets workers cross instances. +func TestWorkerCheckoutAppliesConfig(t *testing.T) { + cfg := DefaultTrieConfig() + cfg.MemoizationOff = true + + stale := NewHexPatriciaHashed(2*length.Addr, nil, DefaultTrieConfig()) + stale.Release() + + for range 2 { + 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) + w.Release() + } +} + func TestParallelPatriciaHashedSkeletonRelease(t *testing.T) { p := NewParallelPatriciaHashed(nil, length.Addr, DefaultTrieConfig()) require.NotNil(t, p.template) @@ -872,6 +890,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..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 sync.Pool trie *prefixTrie splits map[byte]*splitState eagerFloor uint64 @@ -123,18 +122,9 @@ func NewStreamingCommitter(ctxFactory TrieContextFactory, accountKeyLen int16, c splits: make(map[byte]*splitState), eagerFloor: defaultEagerFold, } - sc.resetPool() 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 +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().(*HexPatriciaHashed) + 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))) @@ -593,8 +583,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) + w.Release() return c, deferred, ov.flushed, err } @@ -746,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().(*HexPatriciaHashed) + 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))) @@ -773,8 +762,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) + w.Release() for _, upd := range pu.deferredCombined { putDeferredUpdate(upd) } @@ -782,8 +770,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) + w.Release() for _, upd := range pu.deferredCombined { putDeferredUpdate(upd) } @@ -794,8 +781,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) + w.Release() s.mu.Lock() for _, upd := range s.deferred { @@ -814,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. @@ -942,7 +928,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 +952,4 @@ func (sc *StreamingCommitter) Release() { } sc.deferredForCaller = nil sc.rootValid, sc.rootSeeded = false, false - sc.resetPool() } diff --git a/execution/commitment/streaming_deep_fold.go b/execution/commitment/streaming_deep_fold.go index 5a8aec0aa84..de1cc4ed435 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, 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() { - w.resetForReuse() - pool.Put(w) + w.Release() if cleanup != nil { cleanup() }