Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: (DutyScheduler) duties reset race condition #1741

Merged
merged 6 commits into from
Sep 17, 2024
Merged

fix: (DutyScheduler) duties reset race condition #1741

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
e8e5668
fix: (DutyScheduler) duties reset race condition
moshe-blox Sep 16, 2024
b4cc4a9
refactors
moshe-blox Sep 16, 2024
25a5efc
fix missing lock
moshe-blox Sep 16, 2024
38293bd
Merge branch 'stage' into fix/duty-scheduler-reset
moshe-blox Sep 16, 2024
fcd2a0e
fix test
moshe-blox Sep 16, 2024
697505f
optimization
moshe-blox Sep 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions operator/duties/attester.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,10 +80,6 @@ func (h *AttesterHandler) HandleDuties(ctx context.Context) {
h.logger.Debug("🛠 ticker event", zap.String("epoch_slot_pos", buildStr))

h.processExecution(currentEpoch, slot)
if h.indicesChanged {
h.duties.ResetEpoch(currentEpoch)
h.indicesChanged = false
}
h.processFetching(ctx, currentEpoch, slot)

slotsPerEpoch := h.network.Beacon.SlotsPerEpoch()
Expand Down Expand Up @@ -129,7 +125,6 @@ func (h *AttesterHandler) HandleDuties(ctx context.Context) {
buildStr := fmt.Sprintf("e%v-s%v-#%v", currentEpoch, slot, slot%32+1)
h.logger.Info("🔁 indices change received", zap.String("epoch_slot_pos", buildStr))

h.indicesChanged = true
h.fetchCurrentEpoch = true

// reset next epoch duties if in appropriate slot range
Expand Down Expand Up @@ -215,10 +210,17 @@ func (h *AttesterHandler) fetchAndProcessDuties(ctx context.Context, epoch phase
}

specDuties := make([]*spectypes.ValidatorDuty, 0, len(duties))
dutyDescriptors := make([]dutystore.DutyDescriptor[eth2apiv1.AttesterDuty], 0, len(duties))
for _, d := range duties {
h.duties.Add(epoch, d.Slot, d.ValidatorIndex, d, true)
dutyDescriptors = append(dutyDescriptors, dutystore.DutyDescriptor[eth2apiv1.AttesterDuty]{
Slot: d.Slot,
ValidatorIndex: d.ValidatorIndex,
Duty: d,
InCommittee: true,
})
specDuties = append(specDuties, h.toSpecDuty(d, spectypes.BNRoleAttester))
}
h.duties.Set(epoch, dutyDescriptors)

h.logger.Debug("🗂 got duties",
fields.Count(len(duties)),
Expand Down
35 changes: 17 additions & 18 deletions operator/duties/dutystore/duties.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,19 +11,21 @@ type Duty interface {
eth2apiv1.AttesterDuty | eth2apiv1.ProposerDuty | eth2apiv1.SyncCommitteeDuty
}

type dutyDescriptor[D Duty] struct {
duty *D
inCommittee bool
type DutyDescriptor[D Duty] struct {
Slot phase0.Slot
ValidatorIndex phase0.ValidatorIndex
Duty *D
InCommittee bool
}

type Duties[D Duty] struct {
mu sync.RWMutex
m map[phase0.Epoch]map[phase0.Slot]map[phase0.ValidatorIndex]dutyDescriptor[D]
m map[phase0.Epoch]map[phase0.Slot]map[phase0.ValidatorIndex]DutyDescriptor[D]
}

func NewDuties[D Duty]() *Duties[D] {
return &Duties[D]{
m: make(map[phase0.Epoch]map[phase0.Slot]map[phase0.ValidatorIndex]dutyDescriptor[D]),
m: make(map[phase0.Epoch]map[phase0.Slot]map[phase0.ValidatorIndex]DutyDescriptor[D]),
}
}

Expand All @@ -43,8 +45,8 @@ func (d *Duties[D]) CommitteeSlotDuties(epoch phase0.Epoch, slot phase0.Slot) []

var duties []*D
for _, descriptor := range descriptorMap {
if descriptor.inCommittee {
duties = append(duties, descriptor.duty)
if descriptor.InCommittee {
duties = append(duties, descriptor.Duty)
}
}

Expand All @@ -70,22 +72,19 @@ func (d *Duties[D]) ValidatorDuty(epoch phase0.Epoch, slot phase0.Slot, validato
return nil
}

return descriptor.duty
return descriptor.Duty
}

func (d *Duties[D]) Add(epoch phase0.Epoch, slot phase0.Slot, validatorIndex phase0.ValidatorIndex, duty *D, inCommittee bool) {
func (d *Duties[D]) Set(epoch phase0.Epoch, duties []DutyDescriptor[D]) {
d.mu.Lock()
defer d.mu.Unlock()

if _, ok := d.m[epoch]; !ok {
d.m[epoch] = make(map[phase0.Slot]map[phase0.ValidatorIndex]dutyDescriptor[D])
}
if _, ok := d.m[epoch][slot]; !ok {
d.m[epoch][slot] = make(map[phase0.ValidatorIndex]dutyDescriptor[D])
}
d.m[epoch][slot][validatorIndex] = dutyDescriptor[D]{
duty: duty,
inCommittee: inCommittee,
d.m[epoch] = make(map[phase0.Slot]map[phase0.ValidatorIndex]DutyDescriptor[D])
for _, duty := range duties {
if _, ok := d.m[epoch][duty.Slot]; !ok {
d.m[epoch][duty.Slot] = make(map[phase0.ValidatorIndex]DutyDescriptor[D])
}
d.m[epoch][duty.Slot][duty.ValidatorIndex] = duty
}
}

Expand Down
25 changes: 9 additions & 16 deletions operator/duties/dutystore/sync_committee.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,12 @@ import (

type SyncCommitteeDuties struct {
mu sync.RWMutex
m map[uint64]map[phase0.ValidatorIndex]dutyDescriptor[eth2apiv1.SyncCommitteeDuty]
m map[uint64]map[phase0.ValidatorIndex]DutyDescriptor[eth2apiv1.SyncCommitteeDuty]
}

func NewSyncCommitteeDuties() *SyncCommitteeDuties {
return &SyncCommitteeDuties{
m: make(map[uint64]map[phase0.ValidatorIndex]dutyDescriptor[eth2apiv1.SyncCommitteeDuty]),
m: make(map[uint64]map[phase0.ValidatorIndex]DutyDescriptor[eth2apiv1.SyncCommitteeDuty]),
}
}

Expand All @@ -29,8 +29,8 @@ func (d *SyncCommitteeDuties) CommitteePeriodDuties(period uint64) []*eth2apiv1.

var duties []*eth2apiv1.SyncCommitteeDuty
for _, descriptor := range descriptorMap {
if descriptor.inCommittee {
duties = append(duties, descriptor.duty)
if descriptor.InCommittee {
duties = append(duties, descriptor.Duty)
}
}

Expand All @@ -51,20 +51,13 @@ func (d *SyncCommitteeDuties) Duty(period uint64, validatorIndex phase0.Validato
return nil
}

return descriptor.duty
return descriptor.Duty
}

func (d *SyncCommitteeDuties) Add(period uint64, validatorIndex phase0.ValidatorIndex, duty *eth2apiv1.SyncCommitteeDuty, inCommittee bool) {
d.mu.Lock()
defer d.mu.Unlock()

if _, ok := d.m[period]; !ok {
d.m[period] = make(map[phase0.ValidatorIndex]dutyDescriptor[eth2apiv1.SyncCommitteeDuty])
}

d.m[period][validatorIndex] = dutyDescriptor[eth2apiv1.SyncCommitteeDuty]{
duty: duty,
inCommittee: inCommittee,
func (d *SyncCommitteeDuties) Set(period uint64, duties []DutyDescriptor[eth2apiv1.SyncCommitteeDuty]) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you removed the mutex lock here

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@y0sher good catch, fixed
i also pushed a bunch of refactors

d.m[period] = make(map[phase0.ValidatorIndex]DutyDescriptor[eth2apiv1.SyncCommitteeDuty])
for _, duty := range duties {
d.m[period][duty.ValidatorIndex] = duty
}
}

Expand Down
9 changes: 8 additions & 1 deletion operator/duties/proposer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,11 +181,18 @@ func (h *ProposerHandler) fetchAndProcessDuties(ctx context.Context, epoch phase
h.duties.ResetEpoch(epoch)

specDuties := make([]*spectypes.ValidatorDuty, 0, len(duties))
dutyDescriptors := make([]dutystore.DutyDescriptor[eth2apiv1.ProposerDuty], 0, len(duties))
for _, d := range duties {
_, inCommitteeDuty := selfIndicesSet[d.ValidatorIndex]
h.duties.Add(epoch, d.Slot, d.ValidatorIndex, d, inCommitteeDuty)
dutyDescriptors = append(dutyDescriptors, dutystore.DutyDescriptor[eth2apiv1.ProposerDuty]{
Slot: d.Slot,
ValidatorIndex: d.ValidatorIndex,
Duty: d,
InCommittee: inCommitteeDuty,
})
specDuties = append(specDuties, h.toSpecDuty(d, spectypes.BNRoleProposer))
}
h.duties.Set(epoch, dutyDescriptors)

h.logger.Debug("📚 got duties",
fields.Count(len(duties)),
Expand Down
9 changes: 7 additions & 2 deletions operator/duties/sync_committee.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,11 +218,16 @@ func (h *SyncCommitteeHandler) fetchAndProcessDuties(ctx context.Context, period
return fmt.Errorf("failed to fetch sync committee duties: %w", err)
}

h.duties.Reset(period)
dutyDescriptors := make([]dutystore.DutyDescriptor[eth2apiv1.SyncCommitteeDuty], 0, len(duties))
for _, d := range duties {
_, inCommitteeDuty := inCommitteeIndicesSet[d.ValidatorIndex]
h.duties.Add(period, d.ValidatorIndex, d, inCommitteeDuty)
dutyDescriptors = append(dutyDescriptors, dutystore.DutyDescriptor[eth2apiv1.SyncCommitteeDuty]{
ValidatorIndex: d.ValidatorIndex,
Duty: d,
InCommittee: inCommitteeDuty,
})
}
h.duties.Set(period, dutyDescriptors)

h.prepareDutiesResultLog(period, duties, start)

Expand Down
Loading