eth/protocols/eth: use bloomfilter instead of set

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

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)
+	bloom.Add(testKey)
-	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()
 	}
 	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
+	// Add key in bloom 0
-	bloom.Put(testKey)
+	bloom.Add(testKey)
-	if !bloom.Contain(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)
 	}
 }
 

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
+	// transactionEvictionInterval specifies the interval in which a previously recorded
-	// before starting to randomly evict them.
+	// transaction hash is forgotten.
-	maxKnownTxs = 32768
+	transactionEvictionInterval = 10 * time.Minute
 
-	// maxKnownBlocks is the maximum block hashes to keep in the known list
+	// blockEvictionInterval specifies the interval in which a previously recorded
-	// before starting to randomly evict them.
+	// block hash is forgotten.
-	maxKnownBlocks = 1024
+	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
+	txpool      TxPool               // Transaction pool used by the broadcasters for liveness checks
-	knownTxs    *knownCache        // Set of transaction hashes known to be known by this peer
+	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
+	txBroadcast chan []common.Hash   // Channel used to queue transaction propagation requests
-	txAnnounce  chan []common.Hash // Channel used to queue transaction announcement 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),
+		knownTxs:        txBloom,
-		knownBlocks:     newKnownCache(maxKnownBlocks),
+		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
+		// Mark all the transactions as known
-		p.knownTxs.Add(hashes...)
+		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
+	// Mark all the transactions as known
-	p.knownTxs.Add(hashes...)
+	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
+	// Mark all the transactions as known
-	p.knownTxs.Add(hashes...)
+	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
+		// Mark all the transactions as known
-		p.knownTxs.Add(hashes...)
+		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
+	// Mark all the transactions as known
-	p.knownTxs.Add(hashes...)
+	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
+	// Mark all the block hashes as known
-	p.knownBlocks.Add(hashes...)
+	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
+		// Mark all the block hash as known
-		p.knownBlocks.Add(block.Hash())
+		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
+	// Mark all the block hash as known
-	p.knownBlocks.Add(block.Hash())
+	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
+		// Mark all the block hash as known
-		p.knownBlocks.Add(block.Hash())
+		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.
+// bloomHashWrapper wraps a common.Hash to be used in the bloom filter library.
-type knownCache struct {
+// It converts the common.Hash to a mini hash of size 8 bytes.
-	hashes mapset.Set[common.Hash]
+type bloomHashWrapper common.Hash
-	max    int
-}
 
-// newKnownCache creates a new knownCache with a max capacity.
+func (h bloomHashWrapper) Write(p []byte) (n int, err error) { panic("not implemented") }
-func newKnownCache(max int) *knownCache {
+func (h bloomHashWrapper) Sum(b []byte) []byte               { panic("not implemented") }
-	return &knownCache{
+func (h bloomHashWrapper) Reset()                            { panic("not implemented") }
-		max:    max,
+func (h bloomHashWrapper) BlockSize() int                    { panic("not implemented") }
-		hashes: mapset.NewSet[common.Hash](),
+func (h bloomHashWrapper) Size() int                         { return 8 }
-	}
+func (h bloomHashWrapper) Sum64() uint64 {
-}
+	return binary.BigEndian.Uint64(h[0:8])
-
-// 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()
 }

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