From c3c126bc67aaea663bb32372c8369ac4513c636d Mon Sep 17 00:00:00 2001 From: simlecode <69969590+simlecode@users.noreply.github.com> Date: Tue, 15 Oct 2024 15:18:27 +0800 Subject: [PATCH] fix: send the last gpbft message for each new participant --- pkg/vf3/f3.go | 26 +++++++++++++++++--------- pkg/vf3/participation_lease.go | 7 +++++++ 2 files changed, 24 insertions(+), 9 deletions(-) diff --git a/pkg/vf3/f3.go b/pkg/vf3/f3.go index 39424f0473..b5b95be1ae 100644 --- a/pkg/vf3/f3.go +++ b/pkg/vf3/f3.go @@ -163,20 +163,28 @@ func (fff *F3) runSigningLoop(ctx context.Context) { msgCh := fff.inner.MessagesToSign() -loop: + var mb *gpbft.MessageBuilder + alreadyParticipated := make(map[uint64]struct{}) for ctx.Err() == nil { select { case <-ctx.Done(): return - case mb, ok := <-msgCh: - if !ok { - continue loop + case <-fff.leaser.notifyParticipation: + if mb == nil { + continue } - participants := fff.leaser.getParticipantsByInstance(mb.Payload.Instance) - for _, id := range participants { - if err := participateOnce(ctx, mb, id); err != nil { - log.Errorf("while participating for miner f0%d: %+v", id, err) - } + case mb = <-msgCh: // never closed + clear(alreadyParticipated) + } + + participants := fff.leaser.getParticipantsByInstance(mb.Payload.Instance) + for _, id := range participants { + if _, ok := alreadyParticipated[id]; ok { + continue + } else if err := participateOnce(ctx, mb, id); err != nil { + log.Errorf("while participating for miner f0%d: %+v", id, err) + } else { + alreadyParticipated[id] = struct{}{} } } } diff --git a/pkg/vf3/participation_lease.go b/pkg/vf3/participation_lease.go index 288e1d6113..9ddd686c53 100644 --- a/pkg/vf3/participation_lease.go +++ b/pkg/vf3/participation_lease.go @@ -22,6 +22,8 @@ type leaser struct { issuer peer.ID status f3Status maxLeasableInstances uint64 + // Signals that a lease was created and/or updated. + notifyParticipation chan struct{} } func newParticipationLeaser(nodeID peer.ID, status f3Status, maxLeasedInstances uint64) *leaser { @@ -30,6 +32,7 @@ func newParticipationLeaser(nodeID peer.ID, status f3Status, maxLeasedInstances issuer: nodeID, status: status, maxLeasableInstances: maxLeasedInstances, + notifyParticipation: make(chan struct{}, 1), } } @@ -98,6 +101,10 @@ func (l *leaser) participate(ticket types.F3ParticipationTicket) (types.F3Partic return types.F3ParticipationLease{}, types.ErrF3ParticipationTicketStartBeforeExisting } l.leases[newLease.MinerID] = newLease + select { + case l.notifyParticipation <- struct{}{}: + default: + } return newLease, nil }