From fe67bfdde39f7850398376e7d12f50006ebab2ea Mon Sep 17 00:00:00 2001 From: Bui Quang Minh Date: Tue, 4 Mar 2025 16:29:26 +0700 Subject: [PATCH] eth: avoid transaction pool query when annoucing transaction With eth/68, transaction announcement must have transaction type and size. So in announceTransactions, we need to query the transaction from transaction pool with its hash. This creates overhead in case of blob transaction which needs to load data from limbo and RLP decode. This commit tries to avoid transaction pool query by providing the transaction type and size information along with hash right from the beginning. --- core/types/transaction.go | 13 +++++++++++++ eth/handler.go | 19 +++++++++++++------ eth/protocols/eth/broadcast.go | 24 +++++++++++++++--------- eth/protocols/eth/peer.go | 20 +++++++++++--------- eth/sync.go | 15 ++++++++++----- 5 files changed, 62 insertions(+), 29 deletions(-) diff --git a/core/types/transaction.go b/core/types/transaction.go index 4e48248b4a..44dd5c9ac7 100644 --- a/core/types/transaction.go +++ b/core/types/transaction.go @@ -56,6 +56,12 @@ type Transaction struct { inner TxData // Consensus contents of a transaction time time.Time // Time first seen locally (spam avoidance) + // This is only set in blob transaction without sidecar + // so that we can quickly get the size of whole blob + // transaction without needing to query to database to get + // back the whole blob transaction with sidecar + sidecarSize uint64 + // caches hash atomic.Pointer[common.Hash] size atomic.Uint64 @@ -444,6 +450,9 @@ func (tx *Transaction) WithoutBlobTxSidecar() *Transaction { inner: blobtx.withoutSidecar(), time: tx.time, } + if sc := blobtx.Sidecar; sc != nil { + cpy.sidecarSize = rlp.ListSize(sc.encodedSize()) + } // Note: tx.size cache not carried over because the sidecar is included in size! if h := tx.hash.Load(); h != nil { cpy.hash.Store(h) @@ -563,6 +572,10 @@ func (tx *Transaction) Size() uint64 { return size } +func (tx *Transaction) SidecarSize() uint64 { + return tx.sidecarSize +} + // WithSignature returns a new transaction with the given signature. // This signature needs to be in the [R || S || V] format where V is 0 or 1. func (tx *Transaction) WithSignature(signer Signer, sig []byte) (*Transaction, error) { diff --git a/eth/handler.go b/eth/handler.go index 6ac890902b..6f1e3e7e7c 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -465,8 +465,8 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) { directCount int // Number of transactions sent directly to peers (duplicates included) annCount int // Number of transactions announced across all peers (duplicates included) - txset = make(map[*ethPeer][]common.Hash) // Set peer->hash to transfer directly - annos = make(map[*ethPeer][]common.Hash) // Set peer->hash to announce + txset = make(map[*ethPeer][]common.Hash) // Set peer->hash to transfer directly + annos = make(map[*ethPeer][]*eth.TxAnnouncement) // Set peer->txAnnouncement to announce ) // Broadcast transactions to a batch of peers not knowing about it direct := big.NewInt(int64(math.Sqrt(float64(h.peers.len())))) // Approximate number of peers to broadcast to @@ -515,7 +515,14 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) { if broadcast { txset[peer] = append(txset[peer], tx.Hash()) } else { - annos[peer] = append(annos[peer], tx.Hash()) + annos[peer] = append(annos[peer], ð.TxAnnouncement{ + Hash: tx.Hash(), + Type: tx.Type(), + // If this is a blob transaction, tx is already a + // transaction without sidecar. So we need to add + // the sidecar size to get the correct size. + Size: tx.Size() + tx.SidecarSize(), + }) } } } @@ -523,9 +530,9 @@ func (h *handler) BroadcastTransactions(txs types.Transactions) { directCount += len(hashes) peer.AsyncSendTransactions(hashes) } - for peer, hashes := range annos { - annCount += len(hashes) - peer.AsyncSendPooledTransactionHashes(hashes) + for peer, announces := range annos { + annCount += len(announces) + peer.AsyncSendPooledTransactionHashes(announces) } log.Debug("Distributed transactions", "plaintxs", len(txs)-blobTxs-largeTxs, "blobtxs", blobTxs, "largetxs", largeTxs, "bcastpeers", len(txset), "bcastcount", directCount, "annpeers", len(annos), "anncount", annCount) diff --git a/eth/protocols/eth/broadcast.go b/eth/protocols/eth/broadcast.go index f0ed1d6bc9..6e809b738b 100644 --- a/eth/protocols/eth/broadcast.go +++ b/eth/protocols/eth/broadcast.go @@ -27,6 +27,12 @@ const ( maxTxPacketSize = 100 * 1024 ) +type TxAnnouncement struct { + Hash common.Hash + Type uint8 + Size uint64 +} + // broadcastTransactions is a write loop that schedules transaction broadcasts // to the remote peer. The goal is to have an async writer that does not lock up // node internals and at the same time rate limits queued data. @@ -99,7 +105,7 @@ func (p *Peer) broadcastTransactions() { // node internals and at the same time rate limits queued data. func (p *Peer) announceTransactions() { var ( - queue []common.Hash // Queue of hashes to announce as transaction stubs + queue []*TxAnnouncement // Queue of hashes to announce as transaction stubs done chan struct{} // Non-nil if background announcer is running fail = make(chan error, 1) // Channel used to receive network error failed bool // Flag whether a send failed, discard everything onward @@ -116,12 +122,12 @@ func (p *Peer) announceTransactions() { size common.StorageSize ) for count = 0; count < len(queue) && size < maxTxPacketSize; count++ { - if tx := p.txpool.Get(queue[count]); tx != nil { - pending = append(pending, queue[count]) - pendingTypes = append(pendingTypes, tx.Type()) - pendingSizes = append(pendingSizes, uint32(tx.Size())) - size += common.HashLength - } + announce := queue[count] + + pending = append(pending, announce.Hash) + pendingTypes = append(pendingTypes, announce.Type) + pendingSizes = append(pendingSizes, uint32(announce.Size)) + size += common.HashLength } // Shift and trim queue queue = queue[:copy(queue, queue[count:])] @@ -141,13 +147,13 @@ func (p *Peer) announceTransactions() { } // Transfer goroutine may or may not have been started, listen for events select { - case hashes := <-p.txAnnounce: + case announces := <-p.txAnnounce: // If the connection failed, discard all transaction events if failed { continue } // New batch of transactions to be broadcast, queue them (with cap) - queue = append(queue, hashes...) + queue = append(queue, announces...) if len(queue) > maxQueuedTxAnns { // Fancy copy and resize to ensure buffer doesn't grow indefinitely queue = queue[:copy(queue, queue[len(queue)-maxQueuedTxAnns:])] diff --git a/eth/protocols/eth/peer.go b/eth/protocols/eth/peer.go index 31a35eb186..5e05c463aa 100644 --- a/eth/protocols/eth/peer.go +++ b/eth/protocols/eth/peer.go @@ -48,10 +48,10 @@ type Peer struct { rw p2p.MsgReadWriter // Input/output streams for snap version uint // Protocol version negotiated - txpool TxPool // Transaction pool used by the broadcasters for liveness checks - knownTxs *knownCache // Set of transaction hashes known to be known by this peer - txBroadcast chan []common.Hash // Channel used to queue transaction propagation requests - txAnnounce chan []common.Hash // Channel used to queue transaction announcement requests + txpool TxPool // Transaction pool used by the broadcasters for liveness checks + knownTxs *knownCache // Set of transaction hashes known to be known by this peer + txBroadcast chan []common.Hash // Channel used to queue transaction propagation requests + txAnnounce chan []*TxAnnouncement // Channel used to queue transaction announcement requests reqDispatch chan *request // Dispatch channel to send requests and track then until fulfillment reqCancel chan *cancel // Dispatch channel to cancel pending requests and untrack them @@ -70,7 +70,7 @@ func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter, txpool TxPool) *Pe version: version, knownTxs: newKnownCache(maxKnownTxs), txBroadcast: make(chan []common.Hash), - txAnnounce: make(chan []common.Hash), + txAnnounce: make(chan []*TxAnnouncement), reqDispatch: make(chan *request), reqCancel: make(chan *cancel), resDispatch: make(chan *response), @@ -160,13 +160,15 @@ func (p *Peer) sendPooledTransactionHashes(hashes []common.Hash, types []byte, s // AsyncSendPooledTransactionHashes queues a list of transactions hashes to eventually // announce to a remote peer. The number of pending sends are capped (new ones // will force old sends to be dropped) -func (p *Peer) AsyncSendPooledTransactionHashes(hashes []common.Hash) { +func (p *Peer) AsyncSendPooledTransactionHashes(txAnnounce []*TxAnnouncement) { select { - case p.txAnnounce <- hashes: + case p.txAnnounce <- txAnnounce: // Mark all the transactions as known, but ensure we don't overflow our limits - p.knownTxs.Add(hashes...) + for _, announce := range txAnnounce { + p.knownTxs.Add(announce.Hash) + } case <-p.term: - p.Log().Debug("Dropping transaction announcement", "count", len(hashes)) + p.Log().Debug("Dropping transaction announcement", "count", len(txAnnounce)) } } diff --git a/eth/sync.go b/eth/sync.go index 61f2b2b376..bfc650d069 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -17,21 +17,26 @@ package eth import ( - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/txpool" "github.com/ethereum/go-ethereum/eth/protocols/eth" ) // syncTransactions starts sending all currently pending transactions to the given peer. func (h *handler) syncTransactions(p *eth.Peer) { - var hashes []common.Hash + var announces []*eth.TxAnnouncement for _, batch := range h.txpool.Pending(txpool.PendingFilter{OnlyPlainTxs: true}) { for _, tx := range batch { - hashes = append(hashes, tx.Hash) + resolved := tx.Resolve() + + announces = append(announces, ð.TxAnnouncement{ + Hash: tx.Hash, + Size: resolved.Size(), + Type: resolved.Type(), + }) } } - if len(hashes) == 0 { + if len(announces) == 0 { return } - p.AsyncSendPooledTransactionHashes(hashes) + p.AsyncSendPooledTransactionHashes(announces) }