Merge branch 'bs/cell-blobpool/sparse-v2' of https://github.com/healthykim/go-ethereum into bs/cell-blobpool/sparse-v2

core/txpool/blobpool/buffer.go

@@ -69,32 +69,35 @@ type BlobBuffer struct {
 	txs   map[common.Hash]*txEntry
 	cells map[common.Hash]*cellEntry
 
-	addToPool  func(*BlobTxForPool) error
-	validateTx func(*types.Transaction) error
-	dropPeer   func(string)
-
 	completed      []*BlobTxForPool
 	completedCount atomic.Int32
+	cb             BlobBufferFunctions
 }
 
-func NewBlobBuffer(validateTx func(*types.Transaction) error, addToPool func(*BlobTxForPool) error, dropPeer func(string)) *BlobBuffer {
+type BlobBufferFunctions struct {
+	ValidateTx func(*types.Transaction) error
+	AddToPool  func(*BlobTxForPool) error
+	DropPeer   func(string)
+}
+
+func NewBlobBuffer(cb BlobBufferFunctions) *BlobBuffer {
 	return &BlobBuffer{
-		txs:        make(map[common.Hash]*txEntry),
+		txs:   make(map[common.Hash]*txEntry),
-		cells:      make(map[common.Hash]*cellEntry),
+		cells: make(map[common.Hash]*cellEntry),
-		validateTx: validateTx,
+		cb:    cb,
-		addToPool:  addToPool,
-		dropPeer:   dropPeer,
 	}
 }
 
 // Flush adds all completed entries to the pool and returns the hashes
 // and corresponding errors (nil on success) for each attempted insert.
 func (b *BlobBuffer) Flush() ([]common.Hash, []error) {
-	// Read the count first and return early
+	// Read the count first and return early if there is nothing to do.
-	// todo: increase threshold ?
+	// Flush is called very frequently from the blob fetcher so this
+	// optimization is warranted.
 	if b.completedCount.Load() == 0 {
 		return nil, nil
 	}
+
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
@@ -102,7 +105,7 @@ func (b *BlobBuffer) Flush() ([]common.Hash, []error) {
 	errs := make([]error, len(b.completed))
 	for i, ptx := range b.completed {
 		txs[i] = ptx.Tx.Hash()
-		errs[i] = b.addToPool(ptx)
+		errs[i] = b.cb.AddToPool(ptx)
 	}
 	b.completed = nil
 	b.completedCount.Store(0)
@@ -129,12 +132,12 @@ func (b *BlobBuffer) AddTx(txs []*types.Transaction, peer string) []error {
 		}
 		// tx validation (basic w/o lock)
 		// error will be handled by tx fetcher
-		if err := b.validateTx(tx); err != nil {
+		if err := b.cb.ValidateTx(tx); err != nil {
 			errs[i] = err
 			continue
 		}
 		if entry, ok := b.cells[hash]; ok {
-			b.add(hash, tx, entry)
+			b.storeCompleted(hash, tx, entry)
 			continue
 		}
 		blobBufferTxFirstCounter.Inc(1)
@@ -159,13 +162,14 @@ func (b *BlobBuffer) AddCells(hash common.Hash, deliveries map[string]*PeerDeliv
 		added:      time.Now(),
 	}
 	if txe, ok := b.txs[hash]; ok {
-		b.add(hash, txe.tx, b.cells[hash])
+		b.storeCompleted(hash, txe.tx, b.cells[hash])
 	}
 	blobBufferCellsFirstCounter.Inc(1)
 }
 
-// add verifies cells per-peer, sorts them, and adds to the pool.
+// storeCompleted verifies cells per-peer, sorts them, and schedules them for
-func (b *BlobBuffer) add(hash common.Hash, tx *types.Transaction, cells *cellEntry) {
+// addition into the pool. The actual addition happens in Flush().
+func (b *BlobBuffer) storeCompleted(hash common.Hash, tx *types.Transaction, cells *cellEntry) {
 	sidecar := tx.BlobTxSidecar()
 
 	// Per-peer cell verification
@@ -210,11 +214,11 @@ func (b *BlobBuffer) HasCells(hash common.Hash) bool {
 }
 
 func (b *BlobBuffer) dropPeers(peers []string) {
-	if b.dropPeer == nil {
+	if b.cb.DropPeer == nil {
 		return
 	}
 	for _, p := range peers {
-		b.dropPeer(p)
+		b.cb.DropPeer(p)
 	}
 }
 

core/txpool/blobpool/buffer_test.go

@@ -39,11 +39,11 @@ func makePeerDelivery(t *testing.T, blobOffset, blobCount int, indices []uint64)
 
 func newTestBuffer(t *testing.T) *BlobBuffer {
 	t.Helper()
-	return NewBlobBuffer(
+	return NewBlobBuffer(BlobBufferFunctions{
-		func(tx *types.Transaction) error { return nil },
+		ValidateTx: func(tx *types.Transaction) error { return nil },
-		func(ptx *BlobTxForPool) error { return nil },
+		AddToPool:  func(ptx *BlobTxForPool) error { return nil },
-		func(peer string) {},
+		DropPeer:   func(peer string) {},
-	)
+	})
 }
 
 func TestSortCells(t *testing.T) {
@@ -171,11 +171,11 @@ func TestBadCell(t *testing.T) {
 	blobCount := 1
 
 	var dropped []string
-	buf := NewBlobBuffer(
+	buf := NewBlobBuffer(BlobBufferFunctions{
-		func(tx *types.Transaction) error { return nil },
+		ValidateTx: func(tx *types.Transaction) error { return nil },
-		func(ptx *BlobTxForPool) error { return nil },
+		AddToPool:  func(ptx *BlobTxForPool) error { return nil },
-		func(peer string) { dropped = append(dropped, peer) },
+		DropPeer:   func(peer string) { dropped = append(dropped, peer) },
-	)
+	})
 
 	tx := makeV1Tx(t, 0, blobCount, 0, key)
 	hash := tx.Hash()

core/txpool/blobpool/cache.go

@@ -81,11 +81,11 @@ type Cache struct {
 	needCell bool
 
 	// channels into loop
-	quit         chan struct{}
+	quit        chan struct{}
-	topkRequest  chan struct{}
+	topkRequest chan struct{}
-	topkTimer    mclock.Timer
+	topkTimer   mclock.Timer
-	hasBlobsCh   chan []common.Hash // list of tx hashes that should be pinned
+	hasBlobsCh  chan []common.Hash // list of tx hashes that should be pinned
-	enableCellCh chan struct{}      // signals the loop to switch to cell mode
+	cellModeCh  chan bool          // signals the loop to switch cell mode on/offo
 
 	step func() // test hook fired after each loop iteration
 
@@ -103,14 +103,14 @@ func NewCache(p *BlobPool) *Cache {
 // It allows injecting a clock and a step hook.
 func newCache(p *BlobPool, clock mclock.Clock, step func()) *Cache {
 	c := &Cache{
-		entries:      make(map[common.Hash]*cachedBlob),
+		entries:     make(map[common.Hash]*cachedBlob),
-		blobpool:     p,
+		blobpool:    p,
-		hasBlobsCh:   make(chan []common.Hash, 1),
+		hasBlobsCh:  make(chan []common.Hash, 1),
-		clock:        clock,
+		clock:       clock,
-		step:         step,
+		step:        step,
-		quit:         make(chan struct{}),
+		quit:        make(chan struct{}),
-		topkRequest:  make(chan struct{}, 1),
+		topkRequest: make(chan struct{}, 1),
-		enableCellCh: make(chan struct{}, 1),
+		cellModeCh:  make(chan bool, 1),
 	}
 
 	c.wg.Add(1)
@@ -311,9 +311,9 @@ func (c *Cache) GetCells(vhashes []common.Hash, mask types.CustodyBitmap) ([][]*
 // blobs. This means we can also cache cells that lack enough blobs to
 // recover. It signals the loop to switch to cell mode and re-select
 // transactions from this wider pool.
-func (c *Cache) EnableCell() {
+func (c *Cache) SetCellMode(on bool) {
 	select {
-	case c.enableCellCh <- struct{}{}:
+	case c.cellModeCh <- on:
 	case <-c.quit:
 	}
 }
@@ -335,13 +335,11 @@ func (c *Cache) loop() {
 			c.update(want)
 			c.triggerTopKAfter(topKTimeout)
 
-		case <-c.enableCellCh:
+		case on := <-c.cellModeCh:
-			// CL supports cell-based blob retrieval; switch to cell mode and
+			// This runs when the CL signals (non-)support for cell proofs. Enable/disable
-			// re-select immediately over the now-wider pool. needCell is only
+			// cell mode and re-select immediately to force an update.
-			// touched here in the loop, so a fresh selection and update observe
+			if c.needCell != on {
-			// a consistent value.
+				c.needCell = on
-			if !c.needCell {
-				c.needCell = true
 				c.triggerTopK()
 			}
 

eth/catalyst/api.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"slices"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -1198,17 +1199,13 @@ func (api *ConsensusAPI) checkFork(timestamp uint64, forks ...forks.Fork) bool {
 func (api *ConsensusAPI) ExchangeCapabilities(caps []string) []string {
 	valueT := reflect.TypeOf(api)
 
-	for _, cap := range caps {
+	// If the CL supports getBlobsV4, we call EnableCell() on the
-		if cap == "engine_getBlobsV4" {
+	// blob cache to skip the blob recovery process. This is a
-			// If the CL supports getBlobsV4, we call EnableCell() on the
+	// one-directional toggle, which assumes that once the CL
-			// blob cache to skip the blob recovery process. This is a
+	// supports getBlobsV4, it will not fall back to getBlobsV3
-			// one-directional toggle, which assumes that once the CL
+	// again.
-			// supports getBlobsV4, it will not fall back to getBlobsV3
+	cellmode := slices.Contains(caps, "engine_getBlobsV4")
-			// again.
+	api.eth.BlobCache().SetCellMode(cellmode)
-			api.eth.BlobCache().EnableCell()
-			break
-		}
-	}
 
 	ourCaps := make([]string, 0, valueT.NumMethod())
 	for i := 0; i < valueT.NumMethod(); i++ {

eth/fetcher/blob_fetcher.go

@@ -746,20 +746,20 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
 	if len(actives) == 0 {
 		return
 	}
-	// For each active peer, try to schedule some payload fetches
-	idle := len(f.requests) == 0
 
+	wasIdle := len(f.requests) == 0
+
+	// For each active peer, try to schedule some payload fetches.
 	f.forEachPeer(actives, func(peer string) {
 		if len(f.announces[peer]) == 0 || len(f.requests[peer]) != 0 {
 			return // continue
 		}
 		var (
-			hashes    = make([]common.Hash, 0, maxTxRetrievals)
+			hashes    []common.Hash
-			custodies = make([]types.CustodyBitmap, 0, maxTxRetrievals)
+			custodies []types.CustodyBitmap
 		)
 		f.forEachAnnounce(f.announces[peer], func(hash common.Hash, cells types.CustodyBitmap) bool {
 			var unfetched types.CustodyBitmap
-
 			if f.fetches[hash] == nil {
 				// tx is not being fetched
 				unfetched = cells
@@ -795,6 +795,7 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
 
 			return len(hashes) < maxPayloadRetrievals
 		})
+
 		// If any hashes were allocated, request them from the peer
 		if len(hashes) > 0 {
 			// Group hashes by custody bitmap
@@ -817,7 +818,7 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
 				request = append(request, cr)
 			}
 			f.requests[peer] = request
-			go func(peer string, request []*cellRequest) {
+			go func() {
 				for _, req := range request {
 					blobRequestOutMeter.Mark(int64(len(req.txs)))
 					if err := f.fn.FetchPayloads(peer, req.txs, req.cells); err != nil {
@@ -826,11 +827,12 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
 						break
 					}
 				}
-			}(peer, request)
+			}()
 		}
 	})
+
 	// If a new request was fired, schedule a timeout timer
-	if idle && len(f.requests) > 0 {
+	if wasIdle && len(f.requests) > 0 {
 		f.rescheduleTimeout(timer, timeout)
 	}
 }

eth/fetcher/tx_fetcher.go

@@ -379,6 +379,8 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction,
 			poolTxs = batch
 		}
 		batch = append(poolTxs, blobTxs...)
+
+		// Add regular tx to pool, blob tx to buffer.
 		errs := append(f.addTxs(poolTxs), f.buffer.AddTx(blobTxs, peer)...)
 
 		hashes := make([]common.Hash, len(batch))

eth/handler.go

@@ -190,7 +190,11 @@ func newHandler(config *handlerConfig) (*handler, error) {
 	}
 
 	// Construct the blob buffer for assembling blob txs from separate tx and cell deliveries.
-	blobBuffer := blobpool.NewBlobBuffer(h.blobpool.ValidateTxBasics, h.blobpool.AddPooledTx, h.removePeer)
+	blobBuffer := blobpool.NewBlobBuffer(blobpool.BlobBufferFunctions{
+		ValidateTx: h.blobpool.ValidateTxBasics,
+		AddToPool:  h.blobpool.AddPooledTx,
+		DropPeer:   h.removePeer,
+	})
 
 	addTxs := func(txs []*types.Transaction) []error {
 		return h.txpool.Add(txs, false)