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,43 @@
+package proofscache
+
+import "github.com/multiversx/mx-chain-core-go/data"
+
+// NewProofsCache -
+func NewProofsCache(bucketSize int) *proofsCache {
+	return newProofsCache(bucketSize)
+}
+
+// HeadBucketSize -
+func (pc *proofsCache) HeadBucketSize() int {
+	if len(pc.proofsByNonceBuckets) > 0 {
+		return len(pc.proofsByNonceBuckets[0].proofsByNonce)
+	}
+
+	return 0
+}
+
+// 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,44 @@
+package proofscache
+
+import "github.com/multiversx/mx-chain-core-go/data"
+
+type proofNonceBucket struct {
+	maxNonce      uint64
+	proofsByNonce []*proofNonceMapping
+	bucketSize    int
+}
+
+func newProofBucket(bucketSize int) *proofNonceBucket {
+	return &proofNonceBucket{
+		proofsByNonce: make([]*proofNonceMapping, 0),
+		bucketSize:    bucketSize,
+	}
+}
+
+func (p *proofNonceBucket) size() int {
+	return len(p.proofsByNonce)
+}
+
+func (p *proofNonceBucket) isFull() bool {
+	return len(p.proofsByNonce) >= p.bucketSize
+}
+
+func (p *proofNonceBucket) insertInNew(proof data.HeaderProofHandler) {
+	p.insert(proof)
+	p.maxNonce = proof.GetHeaderNonce()
+}
+
+func (p *proofNonceBucket) insertInExisting(proof data.HeaderProofHandler) {
+	p.insert(proof)
+
+	if proof.GetHeaderNonce() > p.maxNonce {
+		p.maxNonce = proof.GetHeaderNonce()
+	}
+}
+
+func (p *proofNonceBucket) insert(proof data.HeaderProofHandler) {
+	p.proofsByNonce = append(p.proofsByNonce, &proofNonceMapping{
+		headerHash: string(proof.GetHeaderHash()),
+		nonce:      proof.GetHeaderNonce(),
+	})
+}

dataRetriever/dataPool/proofsCache/proofsCache.go

@@ -1,7 +1,6 @@
 package proofscache
 
 import (
-	"sort"
 	"sync"
 
 	"github.com/multiversx/mx-chain-core-go/data"
@@ -13,22 +12,26 @@ type proofNonceMapping struct {
 }
 
 type proofsCache struct {
-	mutProofsCache sync.RWMutex
-	proofsByNonce  []*proofNonceMapping
-	proofsByHash   map[string]data.HeaderProofHandler
+	mutProofsByNonce     sync.RWMutex
+	proofsByNonceBuckets []*proofNonceBucket
+	bucketSize           int
+
+	proofsByHash    map[string]data.HeaderProofHandler
+	mutProofsByHash sync.RWMutex
 }
 
-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([]*proofNonceBucket, 0),
+		bucketSize:           bucketSize,
+		proofsByHash:         make(map[string]data.HeaderProofHandler),
 	}
 }
 
 func (pc *proofsCache) getProofByHash(headerHash []byte) (data.HeaderProofHandler, error) {
-	pc.mutProofsCache.RLock()
-	defer pc.mutProofsCache.RUnlock()
+	pc.mutProofsByHash.RLock()
+	defer pc.mutProofsByHash.RUnlock()
 
 	proof, ok := pc.proofsByHash[string(headerHash)]
 	if !ok {
@@ -43,39 +46,76 @@ func (pc *proofsCache) addProof(proof data.HeaderProofHandler) {
 		return
 	}
 
-	pc.mutProofsCache.Lock()
-	defer pc.mutProofsCache.Unlock()
+	pc.insertProofByNonce(proof)
 
-	pc.proofsByNonce = append(pc.proofsByNonce, &proofNonceMapping{
-		headerHash: string(proof.GetHeaderHash()),
-		nonce:      proof.GetHeaderNonce(),
-	})
+	pc.mutProofsByHash.Lock()
+	pc.proofsByHash[string(proof.GetHeaderHash())] = proof
+	pc.mutProofsByHash.Unlock()
+}
 
-	sort.Slice(pc.proofsByNonce, func(i, j int) bool {
-		return pc.proofsByNonce[i].nonce < pc.proofsByNonce[j].nonce
-	})
+func (pc *proofsCache) insertProofByNonce(proof data.HeaderProofHandler) {
+	pc.mutProofsByNonce.Lock()
+	defer pc.mutProofsByNonce.Unlock()
 
-	pc.proofsByHash[string(proof.GetHeaderHash())] = proof
+	if len(pc.proofsByNonceBuckets) == 0 {
+		pc.insertInNewBucket(proof)
+		return
+	}
+
+	headBucket := pc.proofsByNonceBuckets[0]
+
+	if headBucket.isFull() {
+		pc.insertInNewBucket(proof)
+		return
+	}
+
+	headBucket.insertInExisting(proof)
+}
+
+func (pc *proofsCache) insertInNewBucket(proof data.HeaderProofHandler) {
+	bucket := newProofBucket(pc.bucketSize)
+	bucket.insertInNew(proof)
+
+	pc.proofsByNonceBuckets = append([]*proofNonceBucket{bucket}, pc.proofsByNonceBuckets...)
 }
 
 func (pc *proofsCache) cleanupProofsBehindNonce(nonce uint64) {
 	if nonce == 0 {
 		return
 	}
 
-	pc.mutProofsCache.Lock()
-	defer pc.mutProofsCache.Unlock()
+	pc.mutProofsByNonce.Lock()
+	defer pc.mutProofsByNonce.Unlock()
+
+	buckets := make([]*proofNonceBucket, 0)
+
+	wg := &sync.WaitGroup{}
+
+	for _, bucket := range pc.proofsByNonceBuckets {
+		if nonce > bucket.maxNonce {
+			wg.Add(1)
 
-	proofsByNonce := make([]*proofNonceMapping, 0)
+			go func(bucket *proofNonceBucket) {
+				pc.cleanupProofsInBucket(bucket)
+				wg.Done()
+			}(bucket)
 
-	for _, proofInfo := range pc.proofsByNonce {
-		if proofInfo.nonce < nonce {
-			delete(pc.proofsByHash, proofInfo.headerHash)
 			continue
 		}
 
-		proofsByNonce = append(proofsByNonce, proofInfo)
+		buckets = append(buckets, bucket)
 	}
 
-	pc.proofsByNonce = proofsByNonce
+	wg.Wait()
+
+	pc.proofsByNonceBuckets = buckets
+}
+
+func (pc *proofsCache) cleanupProofsInBucket(bucket *proofNonceBucket) {
+	pc.mutProofsByHash.Lock()
+	defer pc.mutProofsByHash.Unlock()
+
+	for _, proofInfo := range bucket.proofsByNonce {
+		delete(pc.proofsByHash, proofInfo.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))
+	}
+}