Merge branch 'minhd-vu/sensor-broadcast' into minhd-vu/rpc-methods

Author: minhd-vu

p2p/conns.go

@@ -318,19 +318,30 @@ func (c *Conns) PeerConnectedAt(peerID string) time.Time {
 	return time.Time{}
 }
 
-// AddTx adds a transaction to the shared cache for duplicate detection and serving.
-func (c *Conns) AddTx(hash common.Hash, tx *types.Transaction) {
-	c.txs.Add(hash, tx)
+// AddTxs adds multiple transactions to the shared cache in a single lock operation.
+// Returns the computed hashes for reuse by the caller.
+func (c *Conns) AddTxs(txs []*types.Transaction) []common.Hash {
+	if len(txs) == 0 {
+		return nil
+	}
+	hashes := make([]common.Hash, len(txs))
+	for i, tx := range txs {
+		hashes[i] = tx.Hash()
+	}
+	c.txs.AddBatch(hashes, txs)
+	return hashes
 }
 
-// GetTx retrieves a transaction from the shared cache.
-func (c *Conns) GetTx(hash common.Hash) (*types.Transaction, bool) {
-	return c.txs.Get(hash)
+// PeekTxs retrieves multiple transactions from the shared cache without updating LRU ordering.
+// Uses a single read lock for better concurrency when LRU ordering is not needed.
+func (c *Conns) PeekTxs(hashes []common.Hash) []*types.Transaction {
+	return c.txs.PeekMany(hashes)
 }
 
-// GetTxs retrieves multiple transactions from the shared cache in a single lock operation.
-func (c *Conns) GetTxs(hashes []common.Hash) []*types.Transaction {
-	return c.txs.GetMany(hashes)
+// PeekTxsWithHashes retrieves multiple transactions with their hashes from the cache.
+// Returns parallel slices of found hashes and transactions. Uses a single read lock.
+func (c *Conns) PeekTxsWithHashes(hashes []common.Hash) ([]common.Hash, []*types.Transaction) {
+	return c.txs.PeekManyWithKeys(hashes)
 }
 
 // Blocks returns the global blocks cache.

p2p/datastructures/bloomset_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 )
 
-func TestBloomSet_AddAndContains(t *testing.T) {
+func TestBloomSetAddAndContains(t *testing.T) {
 	b := NewBloomSet(DefaultBloomSetOptions())
 
 	hash1 := common.HexToHash("0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef")
@@ -33,7 +33,7 @@ func TestBloomSet_AddAndContains(t *testing.T) {
 	}
 }
 
-func TestBloomSet_AddMany(t *testing.T) {
+func TestBloomSetAddMany(t *testing.T) {
 	b := NewBloomSet(DefaultBloomSetOptions())
 
 	hashes := make([]common.Hash, 100)
@@ -51,7 +51,7 @@ func TestBloomSet_AddMany(t *testing.T) {
 	}
 }
 
-func TestBloomSet_FilterNotContained(t *testing.T) {
+func TestBloomSetFilterNotContained(t *testing.T) {
 	b := NewBloomSet(DefaultBloomSetOptions())
 
 	// Add some hashes
@@ -82,7 +82,7 @@ func TestBloomSet_FilterNotContained(t *testing.T) {
 	}
 }
 
-func TestBloomSet_Rotate(t *testing.T) {
+func TestBloomSetRotate(t *testing.T) {
 	b := NewBloomSet(DefaultBloomSetOptions())
 
 	hash1 := common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111")
@@ -116,7 +116,7 @@ func TestBloomSet_Rotate(t *testing.T) {
 	}
 }
 
-func TestBloomSet_MemoryUsage(t *testing.T) {
+func TestBloomSetMemoryUsage(t *testing.T) {
 	opts := DefaultBloomSetOptions()
 	b := NewBloomSet(opts)
 
@@ -136,13 +136,13 @@ func generateTestHash(seed uint64) common.Hash {
 	return crypto.Keccak256Hash(buf[:])
 }
 
-func TestBloomSet_FalsePositiveRate(t *testing.T) {
+func TestBloomSetFalsePositiveRate(t *testing.T) {
 	// Test that false positive rate is approximately as expected
 	b := NewBloomSet(DefaultBloomSetOptions())
 
 	// Add 32768 unique hashes (the design capacity)
 	// Use keccak256 to generate properly distributed hashes
-	for i := uint64(0); i < 32768; i++ {
+	for i := range uint64(32768) {
 		b.Add(generateTestHash(i))
 	}
 
@@ -162,7 +162,7 @@ func TestBloomSet_FalsePositiveRate(t *testing.T) {
 	t.Logf("False positive rate: %.2f%%", rate*100)
 }
 
-func BenchmarkBloomSet_Add(b *testing.B) {
+func BenchmarkBloomSetAdd(b *testing.B) {
 	bloom := NewBloomSet(DefaultBloomSetOptions())
 	hash := common.HexToHash("0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef")
 
@@ -171,7 +171,7 @@ func BenchmarkBloomSet_Add(b *testing.B) {
 	}
 }
 
-func BenchmarkBloomSet_Contains(b *testing.B) {
+func BenchmarkBloomSetContains(b *testing.B) {
 	bloom := NewBloomSet(DefaultBloomSetOptions())
 	hash := common.HexToHash("0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef")
 	bloom.Add(hash)
@@ -181,7 +181,7 @@ func BenchmarkBloomSet_Contains(b *testing.B) {
 	}
 }
 
-func BenchmarkBloomSet_FilterNotContained(b *testing.B) {
+func BenchmarkBloomSetFilterNotContained(b *testing.B) {
 	bloom := NewBloomSet(DefaultBloomSetOptions())
 
 	// Add 1000 hashes
@@ -196,29 +196,8 @@ func BenchmarkBloomSet_FilterNotContained(b *testing.B) {
 		batch[i] = common.BytesToHash([]byte{byte(i >> 8), byte(i)})
 	}
 
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
+	for b.Loop() {
 		bloom.FilterNotContained(batch)
 	}
 }
 
-func BenchmarkLRU_FilterNotContained(b *testing.B) {
-	cache := NewLRU[common.Hash, struct{}](LRUOptions{MaxSize: 32768})
-
-	// Add 1000 hashes
-	for i := range 1000 {
-		hash := common.BytesToHash([]byte{byte(i >> 8), byte(i)})
-		cache.Add(hash, struct{}{})
-	}
-
-	// Create a batch of 100 hashes (mix of known and unknown)
-	batch := make([]common.Hash, 100)
-	for i := range batch {
-		batch[i] = common.BytesToHash([]byte{byte(i >> 8), byte(i)})
-	}
-
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		cache.FilterNotContained(batch)
-	}
-}

p2p/datastructures/lru.go

@@ -100,16 +100,39 @@ func (c *LRU[K, V]) Get(key K) (V, bool) {
 	return e.value, true
 }
 
-// GetMany retrieves multiple values from the cache and updates LRU ordering.
-// Uses a single write lock for all lookups, reducing lock contention compared
-// to calling Get in a loop. Returns a slice of values for keys that were found.
-func (c *LRU[K, V]) GetMany(keys []K) []V {
+// Peek retrieves a value from the cache without updating LRU ordering.
+// Uses a read lock for better concurrency.
+func (c *LRU[K, V]) Peek(key K) (V, bool) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	elem, ok := c.items[key]
+	if !ok {
+		var zero V
+		return zero, false
+	}
+
+	e := elem.Value.(*entry[K, V])
+
+	if e.expiresAt != nil && time.Now().After(*e.expiresAt) {
+		var zero V
+		return zero, false
+	}
+
+	return e.value, true
+}
+
+// PeekMany retrieves multiple values from the cache without updating LRU ordering.
+// Uses a single read lock for all lookups, providing better concurrency than GetMany
+// when LRU ordering updates are not needed. Returns only found values (indices don't
+// correspond to input keys). Use PeekManyWithKeys if you need key-value pairs.
+func (c *LRU[K, V]) PeekMany(keys []K) []V {
 	if len(keys) == 0 {
 		return nil
 	}
 
-	c.mu.Lock()
-	defer c.mu.Unlock()
+	c.mu.RLock()
+	defer c.mu.RUnlock()
 
 	now := time.Now()
 	result := make([]V, 0, len(keys))
@@ -123,38 +146,47 @@ func (c *LRU[K, V]) GetMany(keys []K) []V {
 		e := elem.Value.(*entry[K, V])
 
 		if e.expiresAt != nil && now.After(*e.expiresAt) {
-			c.list.Remove(elem)
-			delete(c.items, key)
 			continue
 		}
 
-		c.list.MoveToFront(elem)
 		result = append(result, e.value)
 	}
 
 	return result
 }
 
-// Peek retrieves a value from the cache without updating LRU ordering.
-// Uses a read lock for better concurrency.
-func (c *LRU[K, V]) Peek(key K) (V, bool) {
+// PeekManyWithKeys retrieves multiple key-value pairs from the cache without updating
+// LRU ordering. Returns parallel slices of found keys and values. Uses a single read
+// lock for all lookups.
+func (c *LRU[K, V]) PeekManyWithKeys(keys []K) ([]K, []V) {
+	if len(keys) == 0 {
+		return nil, nil
+	}
+
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 
-	elem, ok := c.items[key]
-	if !ok {
-		var zero V
-		return zero, false
-	}
+	now := time.Now()
+	foundKeys := make([]K, 0, len(keys))
+	foundValues := make([]V, 0, len(keys))
 
-	e := elem.Value.(*entry[K, V])
+	for _, key := range keys {
+		elem, ok := c.items[key]
+		if !ok {
+			continue
+		}
 
-	if e.expiresAt != nil && time.Now().After(*e.expiresAt) {
-		var zero V
-		return zero, false
+		e := elem.Value.(*entry[K, V])
+
+		if e.expiresAt != nil && now.After(*e.expiresAt) {
+			continue
+		}
+
+		foundKeys = append(foundKeys, key)
+		foundValues = append(foundValues, e.value)
 	}
 
-	return e.value, true
+	return foundKeys, foundValues
 }
 
 // Update atomically updates a value in the cache using the provided update function.
@@ -209,26 +241,6 @@ func (c *LRU[K, V]) Update(key K, updateFn func(V) V) {
 	}
 }
 
-// Contains checks if a key exists in the cache and is not expired.
-// Uses a read lock and doesn't update LRU ordering.
-func (c *LRU[K, V]) Contains(key K) bool {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	elem, ok := c.items[key]
-	if !ok {
-		return false
-	}
-
-	e := elem.Value.(*entry[K, V])
-
-	if e.expiresAt != nil && time.Now().After(*e.expiresAt) {
-		return false
-	}
-
-	return true
-}
-
 // Remove removes a key from the cache and returns the value if it existed.
 func (c *LRU[K, V]) Remove(key K) (V, bool) {
 	c.mu.Lock()
@@ -245,66 +257,11 @@ func (c *LRU[K, V]) Remove(key K) (V, bool) {
 	return zero, false
 }
 
-// Len returns the number of items in the cache.
-func (c *LRU[K, V]) Len() int {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-	return c.list.Len()
-}
-
-// Purge clears all items from the cache.
-func (c *LRU[K, V]) Purge() {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	c.items = make(map[K]*list.Element)
-	c.list.Init()
-}
-
-// Keys returns all keys in the cache.
-func (c *LRU[K, V]) Keys() []K {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	keys := make([]K, 0, c.list.Len())
-	for elem := c.list.Front(); elem != nil; elem = elem.Next() {
-		e := elem.Value.(*entry[K, V])
-		keys = append(keys, e.key)
-	}
-	return keys
-}
-
-// FilterNotContained returns the subset of keys that are not in the cache.
-// Uses a single read lock for all lookups, reducing lock contention compared
-// to calling Contains in a loop.
-func (c *LRU[K, V]) FilterNotContained(keys []K) []K {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
-
-	now := time.Now()
-	result := make([]K, 0, len(keys))
-
-	for _, key := range keys {
-		elem, ok := c.items[key]
-		if !ok {
-			result = append(result, key)
-			continue
-		}
-
-		e := elem.Value.(*entry[K, V])
-		if e.expiresAt != nil && now.After(*e.expiresAt) {
-			result = append(result, key)
-		}
-	}
-
-	return result
-}
-
-// AddMany adds multiple keys with the same value to the cache.
+// AddBatch adds multiple key-value pairs to the cache.
 // Uses a single write lock for all additions, reducing lock contention
-// compared to calling Add in a loop.
-func (c *LRU[K, V]) AddMany(keys []K, value V) {
-	if len(keys) == 0 {
+// compared to calling Add in a loop. Keys and values must have the same length.
+func (c *LRU[K, V]) AddBatch(keys []K, values []V) {
+	if len(keys) == 0 || len(keys) != len(values) {
 		return
 	}
 
@@ -317,7 +274,9 @@ func (c *LRU[K, V]) AddMany(keys []K, value V) {
 		expiresAt = &t
 	}
 
-	for _, key := range keys {
+	for i, key := range keys {
+		value := values[i]
+
 		if elem, ok := c.items[key]; ok {
 			c.list.MoveToFront(elem)
 			e := elem.Value.(*entry[K, V])