diff --git a/execution/stagedsync/exec3_fee_merge_temp_test.go b/execution/stagedsync/exec3_fee_merge_temp_test.go index 1b7262abe35..942ae076439 100644 --- a/execution/stagedsync/exec3_fee_merge_temp_test.go +++ b/execution/stagedsync/exec3_fee_merge_temp_test.go @@ -59,6 +59,7 @@ func TestRecordFeeMergeReleasesSupersededTemp(t *testing.T) { // superseded and reclaimed. temp2 := feeMergeTestWrites(t, addr, 3) be.recordFeeMerge(0, temp1, temp2) + be.awaitMapReleases() require.Same(t, temp2, be.feeMergeTemp[0]) require.Equal(t, 0, temp1.Count(), "superseded fee-merge temp must be released") require.Equal(t, 1, temp2.Count()) @@ -68,6 +69,7 @@ func TestRecordFeeMergeReleasesSupersededTemp(t *testing.T) { txOut2 := feeMergeTestWrites(t, addr, 4) temp3 := feeMergeTestWrites(t, addr, 5) be.recordFeeMerge(0, txOut2, temp3) + be.awaitMapReleases() require.Same(t, temp3, be.feeMergeTemp[0]) require.Equal(t, 1, txOut2.Count(), "TxOut must survive the fee merge") require.Equal(t, 1, temp2.Count(), "a temp that is not prev must not be released") @@ -90,6 +92,7 @@ func TestRecordFeeMergeReleaseKeepsSharedWrites(t *testing.T) { merged := temp1.MergeInto(tipWrites) require.Same(t, tipWrites, merged) be.recordFeeMerge(0, temp1, merged) + be.awaitMapReleases() require.Equal(t, 0, temp1.Count()) vw, ok := merged.GetBalance(shared) diff --git a/execution/stagedsync/exec3_parallel.go b/execution/stagedsync/exec3_parallel.go index ffa50cecdd7..56f7ba4cab4 100644 --- a/execution/stagedsync/exec3_parallel.go +++ b/execution/stagedsync/exec3_parallel.go @@ -2303,6 +2303,8 @@ type blockExecutor struct { // recorded set is some execResult's TxOut, which stays live. feeMergeTemp map[int]*state.WriteSet + mapReleasing sync.WaitGroup + // settledInput[tx]==true marks a task that was dispatched when every // preceding task had already validated — so it executed against fully // settled MVCC state, with no lower-indexed worker still in flight. @@ -2479,11 +2481,45 @@ func (be *blockExecutor) invalidBlockResult(err error) *blockResult { // so pooling prev's maps leaves the writes merged now holds intact. func (be *blockExecutor) recordFeeMerge(tx int, prev, merged *state.WriteSet) { if temp := be.feeMergeTemp[tx]; temp != nil && temp == prev && merged != temp { - temp.ReleaseMaps() + be.queueMapRelease(temp) } be.feeMergeTemp[tx] = merged } +// ReleaseMaps clears every map before pooling it, which is O(entries), and a +// superseded fee-merge set holds the whole tx's writes. Keep it off the apply +// loop, which is the serial stage the workers wait behind. +type mapRelease struct { + ws *state.WriteSet + pending *sync.WaitGroup +} + +var ( + mapReleases = make(chan mapRelease, 4096) + mapReleaseStart sync.Once +) + +func (be *blockExecutor) queueMapRelease(ws *state.WriteSet) { + mapReleaseStart.Do(func() { + go func() { + for r := range mapReleases { + r.ws.ReleaseMaps() + r.pending.Done() + } + }() + }) + be.mapReleasing.Add(1) + select { + case mapReleases <- mapRelease{ws, &be.mapReleasing}: + default: + // Releaser is behind; inline costs less than blocking the apply loop. + ws.ReleaseMaps() + be.mapReleasing.Done() + } +} + +func (be *blockExecutor) awaitMapReleases() { be.mapReleasing.Wait() } + // tooManyRetries returns an invalid-block result when tx has exceeded its // retry budget, otherwise nil. origin may be nil (validator-invalid path) // or carry the worker's underlying error.