diff --git a/core/txpool/blobpool/blobpool.go b/core/txpool/blobpool/blobpool.go index 679529b286..35470d970f 100644 --- a/core/txpool/blobpool/blobpool.go +++ b/core/txpool/blobpool/blobpool.go @@ -341,8 +341,6 @@ type BlobPool struct { chain BlockChain // Chain object to access the state through cQueue *conversionQueue // The queue for performing legacy sidecar conversion (TODO: remove after Osaka) - sidecarMigrationDoneCh chan struct{} // Channel for signaling when all txs at the boundary have been converted (TODO: remove after Osaka) - head atomic.Pointer[types.Header] // Current head of the chain state *state.StateDB // Current state at the head of the chain gasTip atomic.Pointer[uint256.Int] // Currently accepted minimum gas tip @@ -1046,9 +1044,6 @@ func (p *BlobPool) convertLegacySidecars(ids map[common.Address]map[uint64]uint6 } } } - if p.sidecarMigrationDoneCh != nil { - close(p.sidecarMigrationDoneCh) - } log.Info("Completed blob transaction conversion", "discarded", failure, "injected", success, "elapsed", common.PrettyDuration(time.Since(start))) } diff --git a/core/txpool/blobpool/blobpool_test.go b/core/txpool/blobpool/blobpool_test.go index 29db9f1141..9e96365fff 100644 --- a/core/txpool/blobpool/blobpool_test.go +++ b/core/txpool/blobpool/blobpool_test.go @@ -2196,7 +2196,6 @@ func TestSidecarConversion(t *testing.T) { } // Kick off migration. - pool.sidecarMigrationDoneCh = make(chan struct{}) pool.Reset(header0, header1) // Add the v0 sidecar tx, but don't block so we can keep doing other stuff @@ -2245,7 +2244,7 @@ func TestSidecarConversion(t *testing.T) { } // Wait for the pool migration to complete. - <-pool.sidecarMigrationDoneCh + <-pool.cQueue.anyBillyConversionDone // Verify all transactions in the pool were converted and verify the // subsequent cell proofs. diff --git a/core/txpool/blobpool/conversion.go b/core/txpool/blobpool/conversion.go index f8d7948021..95828d83b2 100644 --- a/core/txpool/blobpool/conversion.go +++ b/core/txpool/blobpool/conversion.go @@ -43,19 +43,27 @@ type txConvert struct { // it is performed in the background by a single thread, ensuring the main Geth // process is not overloaded. type conversionQueue struct { - tasks chan *txConvert - billy chan func() - quit chan struct{} - closed chan struct{} + tasks chan *txConvert + startBilly chan func() + quit chan struct{} + closed chan struct{} + + billyQueue []func() + billyTaskDone chan struct{} + + // This channel will be closed when the first billy conversion finishes. + // It's added for unit tests to synchronize with the conversion progress. + anyBillyConversionDone chan struct{} } // newConversionQueue constructs the conversion queue. func newConversionQueue() *conversionQueue { q := &conversionQueue{ - tasks: make(chan *txConvert), - billy: make(chan func()), - quit: make(chan struct{}), - closed: make(chan struct{}), + tasks: make(chan *txConvert), + startBilly: make(chan func()), + quit: make(chan struct{}), + closed: make(chan struct{}), + anyBillyConversionDone: make(chan struct{}), } go q.loop() return q @@ -78,7 +86,7 @@ func (q *conversionQueue) convert(tx *types.Transaction) error { // launchBillyConversion starts a conversion task in the background. func (q *conversionQueue) launchBillyConversion(fn func()) error { select { - case q.billy <- fn: + case q.startBilly <- fn: return nil case <-q.closed: return errors.New("conversion queue closed") @@ -122,16 +130,17 @@ func (q *conversionQueue) loop() { defer close(q.closed) var ( - billyDone = make(chan struct{}) - billyTasks int // Count of running billy conversion tasks - done chan struct{} // Non-nil if background routine is active - interrupt *atomic.Int32 // Flag to signal conversion interruption + done chan struct{} // Non-nil if background routine is active + interrupt *atomic.Int32 // Flag to signal conversion interruption // The pending tasks for sidecar conversion. We assume the number of legacy // blob transactions requiring conversion will not be excessive. However, // a hard cap is applied as a protective measure. txTasks []*txConvert + + firstBilly = true ) + for { select { case t := <-q.tasks: @@ -150,18 +159,19 @@ func (q *conversionQueue) loop() { go q.run(tasks, done, interrupt) } - case fn := <-q.billy: - billyTasks++ - go func() { - fn() - billyDone <- struct{}{} - }() - case <-done: done, interrupt = nil, nil - case <-billyDone: - billyTasks-- + case fn := <-q.startBilly: + q.billyQueue = append(q.billyQueue, fn) + q.runNextBillyTask() + + case <-q.billyTaskDone: + if firstBilly { + close(q.anyBillyConversionDone) + firstBilly = false + } + q.runNextBillyTask() case <-q.quit: if done != nil { @@ -169,12 +179,25 @@ func (q *conversionQueue) loop() { interrupt.Store(1) <-done } - for billyTasks > 0 { + if q.billyTaskDone != nil { log.Debug("Waiting for blobpool billy conversion to exit") - <-billyDone - billyTasks-- + <-q.billyTaskDone } return } } } + +func (q *conversionQueue) runNextBillyTask() { + if len(q.billyQueue) == 0 { + q.billyTaskDone = nil + return + } + + fn := q.billyQueue[0] + q.billyQueue = append(q.billyQueue[:0], q.billyQueue[1:]...) + + done := make(chan struct{}) + go func() { defer close(done); fn() }() + q.billyTaskDone = done +}