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