forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 15:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

core/txpool/blobpool: refactor buffer callbacks

Browse source
This commit is contained in:
Felix Lange 2026-06-12 19:43:41 +02:00
parent 9527e04f48
commit a4ed392269
3 changed files with 30 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
core/txpool/blobpool/buffer.go
View file

@ -69,21 +69,22 @@ type BlobBuffer struct {
txs map[common.Hash]*txEntry txs map[common.Hash]*txEntry
cells map[common.Hash]*cellEntry cells map[common.Hash]*cellEntry
addToPool func(*BlobTxForPool) error
validateTx func(*types.Transaction) error
dropPeer func(string)
completed []*BlobTxForPool completed []*BlobTxForPool
completedCount atomic.Int32 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{ 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,
} }
} }
@ -102,7 +103,7 @@ func (b *BlobBuffer) Flush() ([]common.Hash, []error) {
errs := make([]error, len(b.completed)) errs := make([]error, len(b.completed))
for i, ptx := range b.completed { for i, ptx := range b.completed {
txs[i] = ptx.Tx.Hash() txs[i] = ptx.Tx.Hash()
errs[i] = b.addToPool(ptx) errs[i] = b.cb.AddToPool(ptx)
} }
b.completed = nil b.completed = nil
b.completedCount.Store(0) b.completedCount.Store(0)
@ -129,7 +130,7 @@ func (b *BlobBuffer) AddTx(txs []*types.Transaction, peer string) []error {
} }
// tx validation (basic w/o lock) // tx validation (basic w/o lock)
// error will be handled by tx fetcher // 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 errs[i] = err
continue continue
} }
@ -210,11 +211,11 @@ func (b *BlobBuffer) HasCells(hash common.Hash) bool {
} }
func (b *BlobBuffer) dropPeers(peers []string) { func (b *BlobBuffer) dropPeers(peers []string) {
if b.dropPeer == nil { if b.cb.DropPeer == nil {
return return
} }
for _, p := range peers { for _, p := range peers {
b.dropPeer(p) b.cb.DropPeer(p)
} }
} }

20
core/txpool/blobpool/buffer_test.go
View file

@ -39,11 +39,11 @@ func makePeerDelivery(t *testing.T, blobOffset, blobCount int, indices []uint64)
func newTestBuffer(t *testing.T) *BlobBuffer { func newTestBuffer(t *testing.T) *BlobBuffer {
t.Helper() 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) { func TestSortCells(t *testing.T) {
@ -171,11 +171,11 @@ func TestBadCell(t *testing.T) {
blobCount := 1 blobCount := 1
var dropped []string 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) tx := makeV1Tx(t, 0, blobCount, 0, key)
hash := tx.Hash() hash := tx.Hash()

6
eth/handler.go
View file

@ -190,7 +190,11 @@ func newHandler(config *handlerConfig) (*handler, error) {
} }
// Construct the blob buffer for assembling blob txs from separate tx and cell deliveries. // 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 { addTxs := func(txs []*types.Transaction) []error {
return h.txpool.Add(txs, false) return h.txpool.Add(txs, false)