core/txpool/blobpool: queue up billy conversion tasks

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)))
 }
 

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.

core/txpool/blobpool/conversion.go

@@ -44,18 +44,26 @@ type txConvert struct {
 // process is not overloaded.
 type conversionQueue struct {
 	tasks      chan *txConvert
-	billy  chan func()
+	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()),
+		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,8 +130,6 @@ 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
 
@@ -131,7 +137,10 @@ func (q *conversionQueue) loop() {
 		// 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:
+		case fn := <-q.startBilly:
-			billyTasks--
+			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
+				<-q.billyTaskDone
-				billyTasks--
 			}
 			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
+}