Proofs pool improvements

cmd/node/config/config.toml

@@ -373,6 +373,7 @@
 
 [ProofsPoolConfig]
     CleanupNonceDelta = 3
+    BucketSize = 100
 
 [BadBlocksCache]
     Name = "BadBlocksCache"

config/config.go

@@ -22,6 +22,7 @@ type HeadersPoolConfig struct {
 // ProofsPoolConfig will map the proofs cache configuration
 type ProofsPoolConfig struct {
 	CleanupNonceDelta uint64
+	BucketSize        int
 }
 
 // DBConfig will map the database configuration

dataRetriever/dataPool/proofsCache/export_test.go

@@ -0,0 +1,34 @@
+package proofscache
+
+import "github.com/multiversx/mx-chain-core-go/data"
+
+// NewProofsCache -
+func NewProofsCache(bucketSize int) *proofsCache {
+	return newProofsCache(bucketSize)
+}
+
+// HeadBucketSize -
+func (pc *proofsCache) FullProofsByNonceSize() int {
+	size := 0
+
+	for _, bucket := range pc.proofsByNonceBuckets {
+		size += bucket.size()
+	}
+
+	return size
+}
+
+// ProofsByHashSize -
+func (pc *proofsCache) ProofsByHashSize() int {
+	return len(pc.proofsByHash)
+}
+
+// AddProof -
+func (pc *proofsCache) AddProof(proof data.HeaderProofHandler) {
+	pc.addProof(proof)
+}
+
+// CleanupProofsBehindNonce -
+func (pc *proofsCache) CleanupProofsBehindNonce(nonce uint64) {
+	pc.cleanupProofsBehindNonce(nonce)
+}

dataRetriever/dataPool/proofsCache/proofsBucket.go

@@ -0,0 +1,26 @@
+package proofscache
+
+import "github.com/multiversx/mx-chain-core-go/data"
+
+type proofNonceBucket struct {
+	maxNonce      uint64
+	proofsByNonce map[uint64]string
+}
+
+func newProofBucket() *proofNonceBucket {
+	return &proofNonceBucket{
+		proofsByNonce: make(map[uint64]string),
+	}
+}
+
+func (p *proofNonceBucket) size() int {
+	return len(p.proofsByNonce)
+}
+
+func (p *proofNonceBucket) insert(proof data.HeaderProofHandler) {
+	p.proofsByNonce[proof.GetHeaderNonce()] = string(proof.GetHeaderHash())
+
+	if proof.GetHeaderNonce() > p.maxNonce {
+		p.maxNonce = proof.GetHeaderNonce()
+	}
+}

dataRetriever/dataPool/proofsCache/proofsCache.go

@@ -1,29 +1,24 @@
 package proofscache
 
 import (
-	"sort"
 	"sync"
 
 	"github.com/multiversx/mx-chain-core-go/core/check"
 	"github.com/multiversx/mx-chain-core-go/data"
 )
 
-type proofNonceMapping struct {
-	headerHash string
-	nonce      uint64
-}
-
 type proofsCache struct {
-	mutProofsCache sync.RWMutex
-	proofsByNonce  []*proofNonceMapping
-	proofsByHash   map[string]data.HeaderProofHandler
+	mutProofsCache       sync.RWMutex
+	proofsByNonceBuckets map[uint64]*proofNonceBucket
+	bucketSize           uint64
+	proofsByHash         map[string]data.HeaderProofHandler
 }
 
-func newProofsCache() *proofsCache {
+func newProofsCache(bucketSize int) *proofsCache {
 	return &proofsCache{
-		mutProofsCache: sync.RWMutex{},
-		proofsByNonce:  make([]*proofNonceMapping, 0),
-		proofsByHash:   make(map[string]data.HeaderProofHandler),
+		proofsByNonceBuckets: make(map[uint64]*proofNonceBucket),
+		bucketSize:           uint64(bucketSize),
+		proofsByHash:         make(map[string]data.HeaderProofHandler),
 	}
 }
 
@@ -47,18 +42,28 @@ func (pc *proofsCache) addProof(proof data.HeaderProofHandler) {
 	pc.mutProofsCache.Lock()
 	defer pc.mutProofsCache.Unlock()
 
-	pc.proofsByNonce = append(pc.proofsByNonce, &proofNonceMapping{
-		headerHash: string(proof.GetHeaderHash()),
-		nonce:      proof.GetHeaderNonce(),
-	})
-
-	sort.Slice(pc.proofsByNonce, func(i, j int) bool {
-		return pc.proofsByNonce[i].nonce < pc.proofsByNonce[j].nonce
-	})
+	pc.insertProofByNonce(proof)
 
 	pc.proofsByHash[string(proof.GetHeaderHash())] = proof
 }
 
+// getBucketKey will return bucket key as lower bound window value
+func (pc *proofsCache) getBucketKey(index uint64) uint64 {
+	return (index / pc.bucketSize) * pc.bucketSize
+}
+
+func (pc *proofsCache) insertProofByNonce(proof data.HeaderProofHandler) {
+	bucketKey := pc.getBucketKey(proof.GetHeaderNonce())
+
+	bucket, ok := pc.proofsByNonceBuckets[bucketKey]
+	if !ok {
+		bucket = newProofBucket()
+		pc.proofsByNonceBuckets[bucketKey] = bucket
+	}
+
+	bucket.insert(proof)
+}
+
 func (pc *proofsCache) cleanupProofsBehindNonce(nonce uint64) {
 	if nonce == 0 {
 		return
@@ -67,16 +72,16 @@ func (pc *proofsCache) cleanupProofsBehindNonce(nonce uint64) {
 	pc.mutProofsCache.Lock()
 	defer pc.mutProofsCache.Unlock()
 
-	proofsByNonce := make([]*proofNonceMapping, 0)
-
-	for _, proofInfo := range pc.proofsByNonce {
-		if proofInfo.nonce < nonce {
-			delete(pc.proofsByHash, proofInfo.headerHash)
-			continue
+	for key, bucket := range pc.proofsByNonceBuckets {
+		if nonce > bucket.maxNonce {
+			pc.cleanupProofsInBucket(bucket)
+			delete(pc.proofsByNonceBuckets, key)
 		}
-
-		proofsByNonce = append(proofsByNonce, proofInfo)
 	}
+}
 
-	pc.proofsByNonce = proofsByNonce
+func (pc *proofsCache) cleanupProofsInBucket(bucket *proofNonceBucket) {
+	for _, headerHash := range bucket.proofsByNonce {
+		delete(pc.proofsByHash, headerHash)
+	}
 }

dataRetriever/dataPool/proofsCache/proofsCache_bench_test.go

@@ -0,0 +1,66 @@
+package proofscache_test
+
+import (
+	"fmt"
+	"testing"
+
+	proofscache "github.com/multiversx/mx-chain-go/dataRetriever/dataPool/proofsCache"
+)
+
+func Benchmark_AddProof_Bucket10_Pool1000(b *testing.B) {
+	benchmarkAddProof(b, 10, 1000)
+}
+
+func Benchmark_AddProof_Bucket100_Pool10000(b *testing.B) {
+	benchmarkAddProof(b, 100, 10000)
+}
+
+func Benchmark_AddProof_Bucket1000_Pool100000(b *testing.B) {
+	benchmarkAddProof(b, 1000, 100000)
+}
+
+func benchmarkAddProof(b *testing.B, bucketSize int, nonceRange int) {
+	pc := proofscache.NewProofsCache(bucketSize)
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		b.StopTimer()
+		proof := generateProof()
+		nonce := generateRandomNonce(int64(nonceRange))
+
+		proof.HeaderNonce = nonce
+		proof.HeaderHash = []byte("hash_" + fmt.Sprintf("%d", nonce))
+		b.StartTimer()
+
+		pc.AddProof(proof)
+	}
+}
+
+func Benchmark_CleanupProofs_Bucket10_Pool1000(b *testing.B) {
+	benchmarkCleanupProofs(b, 10, 1000)
+}
+
+func Benchmark_CleanupProofs_Bucket100_Pool10000(b *testing.B) {
+	benchmarkCleanupProofs(b, 100, 10000)
+}
+
+func Benchmark_CleanupProofs_Bucket1000_Pool100000(b *testing.B) {
+	benchmarkCleanupProofs(b, 1000, 100000)
+}
+
+func benchmarkCleanupProofs(b *testing.B, bucketSize int, nonceRange int) {
+	pc := proofscache.NewProofsCache(bucketSize)
+
+	for i := uint64(0); i < uint64(nonceRange); i++ {
+		proof := generateProof()
+		proof.HeaderNonce = i
+		proof.HeaderHash = []byte("hash_" + fmt.Sprintf("%d", i))
+
+		pc.AddProof(proof)
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		pc.CleanupProofsBehindNonce(uint64(nonceRange))
+	}
+}

dataRetriever/dataPool/proofsCache/proofsCache_test.go

@@ -0,0 +1,152 @@
+package proofscache_test
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+
+	"github.com/multiversx/mx-chain-core-go/data/block"
+	proofscache "github.com/multiversx/mx-chain-go/dataRetriever/dataPool/proofsCache"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestProofsCache(t *testing.T) {
+	t.Parallel()
+
+	t.Run("incremental nonces, should cleanup all caches", func(t *testing.T) {
+		t.Parallel()
+
+		proof0 := &block.HeaderProof{HeaderHash: []byte{0}, HeaderNonce: 0}
+		proof1 := &block.HeaderProof{HeaderHash: []byte{1}, HeaderNonce: 1}
+		proof2 := &block.HeaderProof{HeaderHash: []byte{2}, HeaderNonce: 2}
+		proof3 := &block.HeaderProof{HeaderHash: []byte{3}, HeaderNonce: 3}
+		proof4 := &block.HeaderProof{HeaderHash: []byte{4}, HeaderNonce: 4}
+
+		pc := proofscache.NewProofsCache(4)
+
+		pc.AddProof(proof0)
+		pc.AddProof(proof1)
+		pc.AddProof(proof2)
+		pc.AddProof(proof3)
+
+		require.Equal(t, 4, pc.FullProofsByNonceSize())
+		require.Equal(t, 4, pc.ProofsByHashSize())
+
+		pc.AddProof(proof4) // added to new head bucket
+
+		require.Equal(t, 5, pc.ProofsByHashSize())
+
+		pc.CleanupProofsBehindNonce(4)
+		require.Equal(t, 1, pc.ProofsByHashSize())
+
+		pc.CleanupProofsBehindNonce(10)
+		require.Equal(t, 0, pc.ProofsByHashSize())
+	})
+
+	t.Run("non incremental nonces", func(t *testing.T) {
+		t.Parallel()
+
+		proof0 := &block.HeaderProof{HeaderHash: []byte{0}, HeaderNonce: 0}
+		proof1 := &block.HeaderProof{HeaderHash: []byte{1}, HeaderNonce: 1}
+		proof2 := &block.HeaderProof{HeaderHash: []byte{2}, HeaderNonce: 2}
+		proof3 := &block.HeaderProof{HeaderHash: []byte{3}, HeaderNonce: 3}
+		proof4 := &block.HeaderProof{HeaderHash: []byte{4}, HeaderNonce: 4}
+		proof5 := &block.HeaderProof{HeaderHash: []byte{5}, HeaderNonce: 5}
+
+		pc := proofscache.NewProofsCache(4)
+
+		pc.AddProof(proof4)
+		pc.AddProof(proof1)
+		pc.AddProof(proof2)
+		pc.AddProof(proof3)
+
+		require.Equal(t, 4, pc.FullProofsByNonceSize())
+		require.Equal(t, 4, pc.ProofsByHashSize())
+
+		pc.AddProof(proof0) // added to new head bucket
+
+		require.Equal(t, 5, pc.FullProofsByNonceSize())
+		require.Equal(t, 5, pc.ProofsByHashSize())
+
+		pc.CleanupProofsBehindNonce(4)
+
+		// cleanup up head bucket with only one proof
+		require.Equal(t, 1, pc.ProofsByHashSize())
+
+		pc.AddProof(proof5) // added to new head bucket
+
+		require.Equal(t, 2, pc.ProofsByHashSize())
+
+		pc.CleanupProofsBehindNonce(5) // will not remove any bucket
+		require.Equal(t, 2, pc.FullProofsByNonceSize())
+		require.Equal(t, 2, pc.ProofsByHashSize())
+
+		pc.CleanupProofsBehindNonce(10)
+		require.Equal(t, 0, pc.ProofsByHashSize())
+	})
+
+	t.Run("shuffled nonces, should cleanup all caches", func(t *testing.T) {
+		t.Parallel()
+
+		pc := proofscache.NewProofsCache(10)
+
+		nonces := generateShuffledNonces(100)
+		for _, nonce := range nonces {
+			proof := generateProof()
+			proof.HeaderNonce = nonce
+			proof.HeaderHash = []byte("hash_" + fmt.Sprintf("%d", nonce))
+
+			pc.AddProof(proof)
+		}
+
+		require.Equal(t, 100, pc.FullProofsByNonceSize())
+		require.Equal(t, 100, pc.ProofsByHashSize())
+
+		pc.CleanupProofsBehindNonce(100)
+		require.Equal(t, 0, pc.FullProofsByNonceSize())
+		require.Equal(t, 0, pc.ProofsByHashSize())
+	})
+}
+
+func TestProofsCache_Concurrency(t *testing.T) {
+	t.Parallel()
+
+	pc := proofscache.NewProofsCache(100)
+
+	numOperations := 1000
+
+	wg := sync.WaitGroup{}
+	wg.Add(numOperations)
+
+	for i := 0; i < numOperations; i++ {
+		go func(idx int) {
+			switch idx % 2 {
+			case 0:
+				pc.AddProof(generateProof())
+			case 1:
+				pc.CleanupProofsBehindNonce(generateRandomNonce(100))
+			default:
+				assert.Fail(t, "should have not beed called")
+			}
+
+			wg.Done()
+		}(i)
+	}
+
+	wg.Wait()
+}
+
+func generateShuffledNonces(n int) []uint64 {
+	nonces := make([]uint64, n)
+	for i := uint64(0); i < uint64(n); i++ {
+		nonces[i] = i
+	}
+
+	rand.Shuffle(len(nonces), func(i, j int) {
+		nonces[i], nonces[j] = nonces[j], nonces[i]
+	})
+
+	return nonces
+}