diff --git a/common/expbloom/bloom.go b/common/expbloom/bloom.go index 853d60d441..a9a0c1401f 100644 --- a/common/expbloom/bloom.go +++ b/common/expbloom/bloom.go @@ -85,7 +85,7 @@ func (e *ExpiringBloom) Stop() { close(e.closeCh) } -func (e *ExpiringBloom) Put(key hash.Hash64) { +func (e *ExpiringBloom) Add(key hash.Hash64) { e.mu.RLock() defer e.mu.RUnlock() @@ -93,7 +93,7 @@ func (e *ExpiringBloom) Put(key hash.Hash64) { e.union.Add(key) } -func (e *ExpiringBloom) Contain(key hash.Hash64) bool { +func (e *ExpiringBloom) Contains(key hash.Hash64) bool { e.mu.RLock() defer e.mu.RUnlock() diff --git a/common/expbloom/bloom_test.go b/common/expbloom/bloom_test.go index a55908990c..35eaf15136 100644 --- a/common/expbloom/bloom_test.go +++ b/common/expbloom/bloom_test.go @@ -39,20 +39,20 @@ func TestBloom(t *testing.T) { bloom, _ := NewExpiringBloom(3, 1024, 10*time.Millisecond) testKey := hashable{[]byte{0x01}} - bloom.Put(testKey) - if !bloom.Contain(testKey) { + bloom.Add(testKey) + if !bloom.Contains(testKey) { t.Fatal() } time.Sleep(11 * time.Millisecond) - if !bloom.Contain(testKey) { + if !bloom.Contains(testKey) { t.Fatal() } time.Sleep(11 * time.Millisecond) - if !bloom.Contain(testKey) { + if !bloom.Contains(testKey) { t.Fatal() } time.Sleep(11 * time.Millisecond) - if bloom.Contain(testKey) { + if bloom.Contains(testKey) { t.Fatal() } } @@ -61,34 +61,34 @@ func TestBloom2(t *testing.T) { bloom, _ := NewExpiringBloom(3, 1024, 10*time.Second) testKey := hashable{[]byte{0x01}} - // Put key in bloom 0 - bloom.Put(testKey) - if !bloom.Contain(testKey) { + // Add key in bloom 0 + bloom.Add(testKey) + if !bloom.Contains(testKey) { t.Fatal() } // Override bloom 1 bloom.tick() - if !bloom.Contain(testKey) { + if !bloom.Contains(testKey) { t.Fatal() } // Override bloom 2 bloom.tick() - if !bloom.Contain(testKey) { + if !bloom.Contains(testKey) { t.Fatal() } // Override bloom 0 bloom.tick() - if bloom.Contain(testKey) { + if bloom.Contains(testKey) { t.Fatal() } } -func BenchmarkPut(b *testing.B) { +func BenchmarkAdd(b *testing.B) { bloom, _ := NewExpiringBloom(2, 1024, 10*time.Second) testKey := hashable{[]byte{0x01}} for i := 0; i < b.N; i++ { - bloom.Put(testKey) + bloom.Add(testKey) } } diff --git a/eth/protocols/eth/peer.go b/eth/protocols/eth/peer.go index 219f486c8e..09d1a2c161 100644 --- a/eth/protocols/eth/peer.go +++ b/eth/protocols/eth/peer.go @@ -17,25 +17,27 @@ package eth import ( + "encoding/binary" "math/big" "math/rand" "sync" + "time" - mapset "github.com/deckarep/golang-set/v2" "github.com/ethereum/go-ethereum/common" + bloom "github.com/ethereum/go-ethereum/common/expbloom" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/rlp" ) const ( - // maxKnownTxs is the maximum transactions hashes to keep in the known list - // before starting to randomly evict them. - maxKnownTxs = 32768 + // transactionEvictionInterval specifies the interval in which a previously recorded + // transaction hash is forgotten. + transactionEvictionInterval = 10 * time.Minute - // maxKnownBlocks is the maximum block hashes to keep in the known list - // before starting to randomly evict them. - maxKnownBlocks = 1024 + // blockEvictionInterval specifies the interval in which a previously recorded + // block hash is forgotten. + blockEvictionInterval = 10 * time.Minute // maxQueuedTxs is the maximum number of transactions to queue up before dropping // older broadcasts. @@ -75,14 +77,14 @@ type Peer struct { head common.Hash // Latest advertised head block hash td *big.Int // Latest advertised head block total difficulty - knownBlocks *knownCache // Set of block hashes known to be known by this peer + knownBlocks *bloom.ExpiringBloom // Set of block hashes known to be known by this peer queuedBlocks chan *blockPropagation // Queue of blocks to broadcast to the peer queuedBlockAnns chan *types.Block // Queue of blocks to announce to the peer - 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 *bloom.ExpiringBloom // 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 reqDispatch chan *request // Dispatch channel to send requests and track then until fulfilment reqCancel chan *cancel // Dispatch channel to cancel pending requests and untrack them @@ -95,13 +97,17 @@ type Peer struct { // NewPeer create a wrapper for a network connection and negotiated protocol // version. func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter, txpool TxPool) *Peer { + // https://hur.st/bloomfilter/?n=32768&p=1.0E-2&m=&k=4 + txBloom, _ := bloom.NewExpiringBloom(10, 42*1024, transactionEvictionInterval/10) + // https://hur.st/bloomfilter/?n=1024&p=1.0E-2&m=&k=4 + blockBloom, _ := bloom.NewExpiringBloom(10, 2*1024, 5*time.Minute) peer := &Peer{ id: p.ID().String(), Peer: p, rw: rw, version: version, - knownTxs: newKnownCache(maxKnownTxs), - knownBlocks: newKnownCache(maxKnownBlocks), + knownTxs: txBloom, + knownBlocks: blockBloom, queuedBlocks: make(chan *blockPropagation, maxQueuedBlocks), queuedBlockAnns: make(chan *types.Block, maxQueuedBlockAnns), txBroadcast: make(chan []common.Hash), @@ -158,26 +164,26 @@ func (p *Peer) SetHead(hash common.Hash, td *big.Int) { // KnownBlock returns whether peer is known to already have a block. func (p *Peer) KnownBlock(hash common.Hash) bool { - return p.knownBlocks.Contains(hash) + return p.knownBlocks.Contains(bloomHashWrapper(hash)) } // KnownTransaction returns whether peer is known to already have a transaction. func (p *Peer) KnownTransaction(hash common.Hash) bool { - return p.knownTxs.Contains(hash) + return p.knownTxs.Contains(bloomHashWrapper(hash)) } // markBlock marks a block as known for the peer, ensuring that the block will // never be propagated to this particular peer. func (p *Peer) markBlock(hash common.Hash) { // If we reached the memory allowance, drop a previously known block hash - p.knownBlocks.Add(hash) + p.knownBlocks.Add(bloomHashWrapper(hash)) } // markTransaction marks a transaction as known for the peer, ensuring that it // will never be propagated to this particular peer. func (p *Peer) markTransaction(hash common.Hash) { // If we reached the memory allowance, drop a previously known transaction hash - p.knownTxs.Add(hash) + p.knownTxs.Add(bloomHashWrapper(hash)) } // SendTransactions sends transactions to the peer and includes the hashes @@ -190,9 +196,9 @@ func (p *Peer) markTransaction(hash common.Hash) { // The reasons this is public is to allow packages using this protocol to write // tests that directly send messages without having to do the async queueing. func (p *Peer) SendTransactions(txs types.Transactions) error { - // Mark all the transactions as known, but ensure we don't overflow our limits + // Mark all the transactions as known for _, tx := range txs { - p.knownTxs.Add(tx.Hash()) + p.knownTxs.Add(bloomHashWrapper(tx.Hash())) } return p2p.Send(p.rw, TransactionsMsg, txs) } @@ -203,8 +209,10 @@ func (p *Peer) SendTransactions(txs types.Transactions) error { func (p *Peer) AsyncSendTransactions(hashes []common.Hash) { select { case p.txBroadcast <- hashes: - // Mark all the transactions as known, but ensure we don't overflow our limits - p.knownTxs.Add(hashes...) + // Mark all the transactions as known + for _, hash := range hashes { + p.knownTxs.Add(bloomHashWrapper(hash)) + } case <-p.term: p.Log().Debug("Dropping transaction propagation", "count", len(hashes)) } @@ -217,8 +225,10 @@ func (p *Peer) AsyncSendTransactions(hashes []common.Hash) { // directly as the queueing (memory) and transmission (bandwidth) costs should // not be managed directly. func (p *Peer) sendPooledTransactionHashes66(hashes []common.Hash) error { - // Mark all the transactions as known, but ensure we don't overflow our limits - p.knownTxs.Add(hashes...) + // Mark all the transactions as known + for _, hash := range hashes { + p.knownTxs.Add(bloomHashWrapper(hash)) + } return p2p.Send(p.rw, NewPooledTransactionHashesMsg, NewPooledTransactionHashesPacket66(hashes)) } @@ -230,8 +240,10 @@ func (p *Peer) sendPooledTransactionHashes66(hashes []common.Hash) error { // directly as the queueing (memory) and transmission (bandwidth) costs should // not be managed directly. func (p *Peer) sendPooledTransactionHashes68(hashes []common.Hash, types []byte, sizes []uint32) error { - // Mark all the transactions as known, but ensure we don't overflow our limits - p.knownTxs.Add(hashes...) + // Mark all the transactions as known + for _, hash := range hashes { + p.knownTxs.Add(bloomHashWrapper(hash)) + } return p2p.Send(p.rw, NewPooledTransactionHashesMsg, NewPooledTransactionHashesPacket68{Types: types, Sizes: sizes, Hashes: hashes}) } @@ -241,8 +253,10 @@ func (p *Peer) sendPooledTransactionHashes68(hashes []common.Hash, types []byte, func (p *Peer) AsyncSendPooledTransactionHashes(hashes []common.Hash) { select { case p.txAnnounce <- hashes: - // Mark all the transactions as known, but ensure we don't overflow our limits - p.knownTxs.Add(hashes...) + // Mark all the transactions as known + for _, hash := range hashes { + p.knownTxs.Add(bloomHashWrapper(hash)) + } case <-p.term: p.Log().Debug("Dropping transaction announcement", "count", len(hashes)) } @@ -250,8 +264,10 @@ func (p *Peer) AsyncSendPooledTransactionHashes(hashes []common.Hash) { // ReplyPooledTransactionsRLP is the eth/66 version of SendPooledTransactionsRLP. func (p *Peer) ReplyPooledTransactionsRLP(id uint64, hashes []common.Hash, txs []rlp.RawValue) error { - // Mark all the transactions as known, but ensure we don't overflow our limits - p.knownTxs.Add(hashes...) + // Mark all the transactions as known + for _, hash := range hashes { + p.knownTxs.Add(bloomHashWrapper(hash)) + } // Not packed into PooledTransactionsPacket to avoid RLP decoding return p2p.Send(p.rw, PooledTransactionsMsg, &PooledTransactionsRLPPacket66{ @@ -263,8 +279,10 @@ func (p *Peer) ReplyPooledTransactionsRLP(id uint64, hashes []common.Hash, txs [ // SendNewBlockHashes announces the availability of a number of blocks through // a hash notification. func (p *Peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error { - // Mark all the block hashes as known, but ensure we don't overflow our limits - p.knownBlocks.Add(hashes...) + // Mark all the block hashes as known + for _, hash := range hashes { + p.knownBlocks.Add(bloomHashWrapper(hash)) + } request := make(NewBlockHashesPacket, len(hashes)) for i := 0; i < len(hashes); i++ { @@ -280,8 +298,8 @@ func (p *Peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error func (p *Peer) AsyncSendNewBlockHash(block *types.Block) { select { case p.queuedBlockAnns <- block: - // Mark all the block hash as known, but ensure we don't overflow our limits - p.knownBlocks.Add(block.Hash()) + // Mark all the block hash as known + p.knownBlocks.Add(bloomHashWrapper(block.Hash())) default: p.Log().Debug("Dropping block announcement", "number", block.NumberU64(), "hash", block.Hash()) } @@ -289,8 +307,8 @@ func (p *Peer) AsyncSendNewBlockHash(block *types.Block) { // SendNewBlock propagates an entire block to a remote peer. func (p *Peer) SendNewBlock(block *types.Block, td *big.Int) error { - // Mark all the block hash as known, but ensure we don't overflow our limits - p.knownBlocks.Add(block.Hash()) + // Mark all the block hash as known + p.knownBlocks.Add(bloomHashWrapper(block.Hash())) return p2p.Send(p.rw, NewBlockMsg, &NewBlockPacket{ Block: block, TD: td, @@ -302,8 +320,8 @@ func (p *Peer) SendNewBlock(block *types.Block, td *big.Int) error { func (p *Peer) AsyncSendNewBlock(block *types.Block, td *big.Int) { select { case p.queuedBlocks <- &blockPropagation{block: block, td: td}: - // Mark all the block hash as known, but ensure we don't overflow our limits - p.knownBlocks.Add(block.Hash()) + // Mark all the block hash as known + p.knownBlocks.Add(bloomHashWrapper(block.Hash())) default: p.Log().Debug("Dropping block propagation", "number", block.NumberU64(), "hash", block.Hash()) } @@ -500,36 +518,15 @@ func (p *Peer) RequestTxs(hashes []common.Hash) error { }) } -// knownCache is a cache for known hashes. -type knownCache struct { - hashes mapset.Set[common.Hash] - max int -} +// bloomHashWrapper wraps a common.Hash to be used in the bloom filter library. +// It converts the common.Hash to a mini hash of size 8 bytes. +type bloomHashWrapper common.Hash -// newKnownCache creates a new knownCache with a max capacity. -func newKnownCache(max int) *knownCache { - return &knownCache{ - max: max, - hashes: mapset.NewSet[common.Hash](), - } -} - -// Add adds a list of elements to the set. -func (k *knownCache) Add(hashes ...common.Hash) { - for k.hashes.Cardinality() > max(0, k.max-len(hashes)) { - k.hashes.Pop() - } - for _, hash := range hashes { - k.hashes.Add(hash) - } -} - -// Contains returns whether the given item is in the set. -func (k *knownCache) Contains(hash common.Hash) bool { - return k.hashes.Contains(hash) -} - -// Cardinality returns the number of elements in the set. -func (k *knownCache) Cardinality() int { - return k.hashes.Cardinality() +func (h bloomHashWrapper) Write(p []byte) (n int, err error) { panic("not implemented") } +func (h bloomHashWrapper) Sum(b []byte) []byte { panic("not implemented") } +func (h bloomHashWrapper) Reset() { panic("not implemented") } +func (h bloomHashWrapper) BlockSize() int { panic("not implemented") } +func (h bloomHashWrapper) Size() int { return 8 } +func (h bloomHashWrapper) Sum64() uint64 { + return binary.BigEndian.Uint64(h[0:8]) } diff --git a/eth/protocols/eth/peer_test.go b/eth/protocols/eth/peer_test.go index efbbbc6fff..372d891f9a 100644 --- a/eth/protocols/eth/peer_test.go +++ b/eth/protocols/eth/peer_test.go @@ -21,9 +21,7 @@ package eth import ( "crypto/rand" - "testing" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" ) @@ -63,28 +61,3 @@ func (p *testPeer) close() { p.Peer.Close() p.app.Close() } - -func TestPeerSet(t *testing.T) { - size := 5 - s := newKnownCache(size) - - // add 10 items - for i := 0; i < size*2; i++ { - s.Add(common.Hash{byte(i)}) - } - - if s.Cardinality() != size { - t.Fatalf("wrong size, expected %d but found %d", size, s.Cardinality()) - } - - vals := []common.Hash{} - for i := 10; i < 20; i++ { - vals = append(vals, common.Hash{byte(i)}) - } - - // add item in batch - s.Add(vals...) - if s.Cardinality() < size { - t.Fatalf("bad size") - } -}