forked from prysmaticlabs/prysm
/
prepare_forkchoice.go
130 lines (113 loc) · 3.82 KB
/
prepare_forkchoice.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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
package attestations
import (
"bytes"
"context"
"errors"
"time"
ethpb "github.com/prysmaticlabs/ethereumapis/eth/v1alpha1"
"github.com/prysmaticlabs/go-bitfield"
"github.com/prysmaticlabs/go-ssz"
"github.com/prysmaticlabs/prysm/beacon-chain/core/helpers"
stateTrie "github.com/prysmaticlabs/prysm/beacon-chain/state"
"github.com/prysmaticlabs/prysm/shared/hashutil"
"github.com/prysmaticlabs/prysm/shared/params"
"go.opencensus.io/trace"
)
// Prepare attestations for fork choice at every half of the slot.
var prepareForkChoiceAttsPeriod = time.Duration(params.BeaconConfig().SecondsPerSlot/3) * time.Second
// This prepares fork choice attestations by running batchForkChoiceAtts
// every prepareForkChoiceAttsPeriod.
func (s *Service) prepareForkChoiceAtts() {
ticker := time.NewTicker(prepareForkChoiceAttsPeriod)
for {
ctx := context.Background()
select {
case <-ticker.C:
if err := s.batchForkChoiceAtts(ctx); err != nil {
log.WithError(err).Error("Could not prepare attestations for fork choice")
}
case <-s.ctx.Done():
log.Debug("Context closed, exiting routine")
return
}
}
}
// This gets the attestations from the unaggregated, aggregated and block
// pool. Then finds the common data, aggregate and batch them for fork choice.
// The resulting attestations are saved in the fork choice pool.
func (s *Service) batchForkChoiceAtts(ctx context.Context) error {
_, span := trace.StartSpan(ctx, "Operations.attestations.batchForkChoiceAtts")
defer span.End()
attsByDataRoot := make(map[[32]byte][]*ethpb.Attestation)
atts := append(s.pool.UnaggregatedAttestations(), s.pool.AggregatedAttestations()...)
atts = append(atts, s.pool.BlockAttestations()...)
atts = append(atts, s.pool.ForkchoiceAttestations()...)
// Consolidate attestations by aggregating them by similar data root.
for _, att := range atts {
seen, err := s.seen(att)
if err != nil {
return err
}
if seen {
continue
}
attDataRoot, err := ssz.HashTreeRoot(att.Data)
if err != nil {
return err
}
attsByDataRoot[attDataRoot] = append(attsByDataRoot[attDataRoot], att)
}
for _, atts := range attsByDataRoot {
if err := s.aggregateAndSaveForkChoiceAtts(atts); err != nil {
return err
}
}
for _, a := range s.pool.BlockAttestations() {
if err := s.pool.DeleteBlockAttestation(a); err != nil {
return err
}
}
return nil
}
// This aggregates a list of attestations using the aggregation algorithm defined in AggregateAttestations
// and saves the attestations for fork choice.
func (s *Service) aggregateAndSaveForkChoiceAtts(atts []*ethpb.Attestation) error {
clonedAtts := make([]*ethpb.Attestation, len(atts))
for i, a := range atts {
clonedAtts[i] = stateTrie.CopyAttestation(a)
}
aggregatedAtts, err := helpers.AggregateAttestations(clonedAtts)
if err != nil {
return err
}
if err := s.pool.SaveForkchoiceAttestations(aggregatedAtts); err != nil {
return err
}
return nil
}
// This checks if the attestation has previously been aggregated for fork choice
// return true if yes, false if no.
func (s *Service) seen(att *ethpb.Attestation) (bool, error) {
attRoot, err := hashutil.HashProto(att.Data)
if err != nil {
return false, err
}
incomingBits := att.AggregationBits
savedBits, ok := s.forkChoiceProcessedRoots.Get(attRoot)
if ok {
savedBitlist, ok := savedBits.(bitfield.Bitlist)
if !ok {
return false, errors.New("not a bit field")
}
if savedBitlist.Len() == incomingBits.Len() {
// Returns true if the node has seen all the bits in the new bit field of the incoming attestation.
if bytes.Equal(savedBitlist, incomingBits) || savedBitlist.Contains(incomingBits) {
return true, nil
}
// Update the bit fields by Or'ing them with the new ones.
incomingBits = incomingBits.Or(savedBitlist)
}
}
s.forkChoiceProcessedRoots.Add(attRoot, incomingBits)
return false, nil
}