Cycle the BLS changes pool when falling below a threshold (#11873)

* Cycle the BLS changes pool when falling below a threshold

* move cycle logic within pool

* set threshold at 2000

* fix conflict

* more fixes

---------

Co-authored-by: kasey <489222+kasey@users.noreply.github.com>
Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

beacon-chain/operations/blstoexec/pool.go

@@ -14,6 +14,11 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+// We recycle the BLS changes pool to avoid the backing map growing without
+// bound. The cycling operation is expensive because it copies all elements, so
+// we only do it when the map is smaller than this upper bound.
+const blsChangesPoolThreshold = 2000
+
 // PoolManager maintains pending and seen BLS-to-execution-change objects.
 // This pool is used by proposers to insert BLS-to-execution-change objects into new blocks.
 type PoolManager interface {
@@ -39,6 +44,15 @@ func NewPool() *Pool {
 	}
 }
 
+// Copies the internal map and returns a new one.
+func (p *Pool) cycleMap() {
+	newMap := make(map[primitives.ValidatorIndex]*doublylinkedlist.Node[*ethpb.SignedBLSToExecutionChange])
+	for k, v := range p.m {
+		newMap[k] = v
+	}
+	p.m = newMap
+}
+
 // PendingBLSToExecChanges returns all objects from the pool.
 func (p *Pool) PendingBLSToExecChanges() ([]*ethpb.SignedBLSToExecutionChange, error) {
 	p.lock.RLock()
@@ -150,6 +164,9 @@ func (p *Pool) MarkIncluded(change *ethpb.SignedBLSToExecutionChange) {
 
 	delete(p.m, change.Message.ValidatorIndex)
 	p.pending.Remove(node)
+	if p.numPending() == blsChangesPoolThreshold {
+		p.cycleMap()
+	}
 }
 
 // ValidatorExists checks if the bls to execution change object exists
@@ -162,3 +179,8 @@ func (p *Pool) ValidatorExists(idx primitives.ValidatorIndex) bool {
 
 	return node != nil
 }
+
+// numPending returns the number of pending bls to execution changes in the pool
+func (p *Pool) numPending() int {
+	return p.pending.Len()
+}

beacon-chain/operations/blstoexec/pool_test.go

@@ -407,3 +407,29 @@ func TestValidatorExists(t *testing.T) {
 		assert.Equal(t, false, pool.ValidatorExists(30))
 	})
 }
+
+func TestPoolCycleMap(t *testing.T) {
+	pool := NewPool()
+	firstChange := &eth.SignedBLSToExecutionChange{
+		Message: &eth.BLSToExecutionChange{
+			ValidatorIndex: primitives.ValidatorIndex(0),
+		}}
+	pool.InsertBLSToExecChange(firstChange)
+	secondChange := &eth.SignedBLSToExecutionChange{
+		Message: &eth.BLSToExecutionChange{
+			ValidatorIndex: primitives.ValidatorIndex(10),
+		}}
+	pool.InsertBLSToExecChange(secondChange)
+	thirdChange := &eth.SignedBLSToExecutionChange{
+		Message: &eth.BLSToExecutionChange{
+			ValidatorIndex: primitives.ValidatorIndex(30),
+		}}
+	pool.InsertBLSToExecChange(thirdChange)
+
+	pool.cycleMap()
+	require.Equal(t, true, pool.ValidatorExists(0))
+	require.Equal(t, true, pool.ValidatorExists(10))
+	require.Equal(t, true, pool.ValidatorExists(30))
+	require.Equal(t, false, pool.ValidatorExists(20))
+
+}

container/doubly-linked-list/list.go

@@ -24,6 +24,21 @@ type Node[T any] struct {
 	next  *Node[T]
 }
 
+// Copy returns a copy of the origina list.
+func (l *List[T]) Copy() *List[T] {
+	if l == nil {
+		return nil
+	}
+	list := &List[T]{}
+	if l.len == 0 {
+		return list
+	}
+	for n := l.First(); n != nil; n = n.next {
+		list.Append(n.Copy())
+	}
+	return list
+}
+
 // First gets the reference to the first node in the list.
 func (l *List[T]) First() *Node[T] {
 	return l.first
@@ -111,3 +126,12 @@ func (n *Node[T]) Value() (T, error) {
 	}
 	return n.value, nil
 }
+
+// Copy copies the given node and returns a new one. It does not do a deep copy
+// of T.
+func (n *Node[T]) Copy() *Node[T] {
+	if n == nil {
+		return nil
+	}
+	return NewNode(n.value)
+}

container/doubly-linked-list/list_test.go

@@ -123,3 +123,36 @@ func TestRemove(t *testing.T) {
 		require.Equal(t, 2, list.len)
 	})
 }
+
+func TestNodeCopy(t *testing.T) {
+	first := NewNode(1)
+	second := first.Copy()
+	v, err := second.Value()
+	require.NoError(t, err)
+	require.Equal(t, first.value, v)
+}
+
+func TestListCopy(t *testing.T) {
+	list := &List[int]{}
+	first := NewNode(1)
+	second := NewNode(2)
+	third := NewNode(3)
+	list.Append(first)
+	list.Append(second)
+	list.Append(third)
+
+	copied := list.Copy()
+	require.Equal(t, 3, copied.Len())
+	m := copied.First()
+	for n := list.First(); n != nil; n = n.next {
+		nv, err := n.Value()
+		require.NoError(t, err)
+		mv, err := m.Value()
+		require.NoError(t, err)
+		require.Equal(t, nv, mv)
+
+		require.NotEqual(t, n, m)
+		m, err = m.Next()
+		require.NoError(t, err)
+	}
+}