execution/stagedsync: reclaim superseded fee-merge write sets off the apply loop

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)

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.