/
contribution.go
90 lines (73 loc) · 2.67 KB
/
contribution.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
package synccommittee
import (
"strconv"
"github.com/pkg/errors"
"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
"github.com/prysmaticlabs/prysm/v4/container/queue"
ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
)
// To give two slots tolerance for objects that arrive earlier.
// This account for previous slot, current slot, two future slots.
const syncCommitteeMaxQueueSize = 4
// SaveSyncCommitteeContribution saves a sync committee contribution in to a priority queue.
// The priority queue is capped at syncCommitteeMaxQueueSize contributions.
func (s *Store) SaveSyncCommitteeContribution(cont *ethpb.SyncCommitteeContribution) error {
if cont == nil {
return errNilContribution
}
s.contributionLock.Lock()
defer s.contributionLock.Unlock()
item, err := s.contributionCache.PopByKey(syncCommitteeKey(cont.Slot))
if err != nil {
return err
}
copied := ethpb.CopySyncCommitteeContribution(cont)
// Contributions exist in the queue. Append instead of insert new.
if item != nil {
contributions, ok := item.Value.([]*ethpb.SyncCommitteeContribution)
if !ok {
return errors.New("not typed []ethpb.SyncCommitteeContribution")
}
contributions = append(contributions, copied)
savedSyncCommitteeContributionTotal.Inc()
return s.contributionCache.Push(&queue.Item{
Key: syncCommitteeKey(cont.Slot),
Value: contributions,
Priority: int64(cont.Slot),
})
}
// Contribution does not exist. Insert new.
if err := s.contributionCache.Push(&queue.Item{
Key: syncCommitteeKey(cont.Slot),
Value: []*ethpb.SyncCommitteeContribution{copied},
Priority: int64(cont.Slot),
}); err != nil {
return err
}
savedSyncCommitteeContributionTotal.Inc()
// Trim contributions in queue down to syncCommitteeMaxQueueSize.
if s.contributionCache.Len() > syncCommitteeMaxQueueSize {
if _, err := s.contributionCache.Pop(); err != nil {
return err
}
}
return nil
}
// SyncCommitteeContributions returns sync committee contributions by slot from the priority queue.
// Upon retrieval, the contribution is removed from the queue.
func (s *Store) SyncCommitteeContributions(slot primitives.Slot) ([]*ethpb.SyncCommitteeContribution, error) {
s.contributionLock.RLock()
defer s.contributionLock.RUnlock()
item := s.contributionCache.RetrieveByKey(syncCommitteeKey(slot))
if item == nil {
return []*ethpb.SyncCommitteeContribution{}, nil
}
contributions, ok := item.Value.([]*ethpb.SyncCommitteeContribution)
if !ok {
return nil, errors.New("not typed []ethpb.SyncCommitteeContribution")
}
return contributions, nil
}
func syncCommitteeKey(slot primitives.Slot) string {
return strconv.FormatUint(uint64(slot), 10)
}