Skip to content

execution/stagedsync: drop block write buffer from calcFees/finalize readers (#23140) #23147

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
sudeepdino008 wants to merge 1 commit into from
Closed

execution/stagedsync: drop block write buffer from calcFees/finalize readers (#23140) #23147

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 12 additions & 9 deletions execution/stagedsync/exec3_parallel.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2735,10 +2735,13 @@ func (be *blockExecutor) nextResult(ctx context.Context, pe *parallelExecutor, r
return nil, fmt.Errorf("apply loop: unexpected task type for tx %d: result.Task=%T", tx, txResult.Task)
}
if stateReader == nil {
// calcFees reads through NewVersionedStateReader(be.versionMap, ...),
// so the version map supplies all in-block state. This reader is only
// the pre-block base, so it does not need the block write buffer.
if txTask.IsHistoric() {
stateReader = state.NewHistoryReaderV3WithBlockCache(applyTx, pe.rs.Domains(), be.blockStateCache, txTask.Version().TxNum)
stateReader = state.NewHistoryReaderV3WithBlockCache(applyTx, pe.rs.Domains(), nil, txTask.Version().TxNum)
} else {
stateReader = state.NewCurrentCachedReaderV3(pe.rs.Domains().AsGetter(applyTx), be.blockStateCache)
stateReader = state.NewCurrentCachedReaderV3(pe.rs.Domains().AsGetter(applyTx), nil)
}
}
tipWrites, err := txResult.calcFees(taskVer, be.versionMap, stateReader, txTask.Rules())
Expand Down Expand Up @@ -2850,15 +2853,15 @@ func (be *blockExecutor) nextResult(ctx context.Context, pe *parallelExecutor, r
}

if stateReader == nil {
// finalize (finalizeTx and the block-end finalizeSystemTx) reads
// through NewVersionedStateReader(be.versionMap, ...), so all
// accumulated in-block state comes from the version map. This
// reader is only the pre-block base and does not need the block
// write buffer.
if txTask.IsHistoric() {
stateReader = state.NewHistoryReaderV3WithBlockCache(applyTx, pe.rs.Domains(), be.blockStateCache, txTask.Version().TxNum)
stateReader = state.NewHistoryReaderV3WithBlockCache(applyTx, pe.rs.Domains(), nil, txTask.Version().TxNum)
} else {
// Use CachedReaderV3 with readCurrent=true so the
// finalize (including system TXs) reads from the
// BlockStateCache write buffer. This ensures the
// system TX sees all accumulated state from prior
// TXs in the block, not stale sd.mem values.
stateReader = state.NewCurrentCachedReaderV3(pe.rs.Domains().AsGetterNoMetrics(applyTx), be.blockStateCache)
stateReader = state.NewCurrentCachedReaderV3(pe.rs.Domains().AsGetterNoMetrics(applyTx), nil)
}
}

Expand Down
Loading